[opa] Correlate OPA server log with Polaris requestId; add observability for non-200 errors from OPA server

extensions/auth/opa/src/main/java/org/apache/polaris/extension/auth/opa/OpaPolarisAuthorizer.java

@@ -68,6 +68,8 @@
 import org.apache.polaris.extension.auth.opa.token.BearerTokenProvider;
 import org.jspecify.annotations.NonNull;
 import org.jspecify.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * OPA-based implementation of {@link PolarisAuthorizer}.
@@ -81,10 +83,13 @@
  * environments.
  */
 class OpaPolarisAuthorizer implements PolarisAuthorizer {
+  private static final Logger LOGGER = LoggerFactory.getLogger(OpaPolarisAuthorizer.class);
+
   private final URI policyUri;
   private final BearerTokenProvider tokenProvider;
   private final CloseableHttpClient httpClient;
   private final ObjectMapper objectMapper;
+  private final String requestId;
   private final String realm;
 
   /**
@@ -97,19 +102,24 @@ class OpaPolarisAuthorizer implements PolarisAuthorizer {
    * @param objectMapper Jackson ObjectMapper for JSON serialization (required). Shared across
    *     authorizer instances to avoid initialization overhead.
    * @param tokenProvider Token provider for authentication (optional)
+   * @param requestId The server-generated request ID (optional), used to correlate OPA queries with
+   *     the originating HTTP request. Resolved once by the caller since this authorizer is
+   *     constructed fresh per request.
    * @param realm The realm identifier (from RealmContext) for isolation in OPA policies.
    */
   public OpaPolarisAuthorizer(
       @NonNull URI policyUri,
       @NonNull CloseableHttpClient httpClient,
       @NonNull ObjectMapper objectMapper,
       @Nullable BearerTokenProvider tokenProvider,
+      @Nullable String requestId,
       @NonNull String realm) {
 
     this.policyUri = policyUri;
     this.tokenProvider = tokenProvider;
     this.httpClient = httpClient;
     this.objectMapper = objectMapper;
+    this.requestId = requestId;
     this.realm = realm;
   }
 
@@ -281,6 +291,7 @@ <T> T httpClientExecute(
   private boolean queryOpaCheckResponse(ClassicHttpResponse response) throws IOException {
     int statusCode = response.getCode();
     if (statusCode != 200) {
+      LOGGER.warn("OPA returned unexpected HTTP status {}, treating as deny", statusCode);
       return false;
     }
 
@@ -338,7 +349,10 @@ private ImmutableActor buildActor(PolarisPrincipal principal) {
   }
 
   private ImmutableContext buildContext() {
-    return ImmutableContext.builder().requestId(UUID.randomUUID().toString()).realm(realm).build();
+    return ImmutableContext.builder()
+        .requestId(requestId != null ? requestId : UUID.randomUUID().toString())
+        .realm(realm)
+        .build();
   }
 
   private ImmutableResource buildResource(

extensions/auth/opa/src/main/java/org/apache/polaris/extension/auth/opa/OpaPolarisAuthorizerFactory.java

@@ -35,6 +35,7 @@
 import org.apache.polaris.core.auth.PolarisAuthorizerFactory;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.context.RequestIdSupplier;
 import org.apache.polaris.extension.auth.opa.token.BearerTokenProvider;
 import org.apache.polaris.extension.auth.opa.token.FileBearerTokenProvider;
 import org.apache.polaris.extension.auth.opa.token.StaticBearerTokenProvider;
@@ -53,6 +54,7 @@ class OpaPolarisAuthorizerFactory implements PolarisAuthorizerFactory {
   private final Clock clock;
   private final ObjectMapper objectMapper;
   private final AsyncExec asyncExec;
+  private final RequestIdSupplier requestIdSupplier;
   private final RealmContext realmContext;
   private CloseableHttpClient httpClient;
   private BearerTokenProvider bearerTokenProvider;
@@ -62,10 +64,12 @@ public OpaPolarisAuthorizerFactory(
       OpaAuthorizationConfig opaConfig,
       Clock clock,
       AsyncExec asyncExec,
+      RequestIdSupplier requestIdSupplier,
       RealmContext realmContext) {
     this.opaConfig = opaConfig;
     this.clock = clock;
     this.asyncExec = asyncExec;
+    this.requestIdSupplier = requestIdSupplier;
     this.realmContext = realmContext;
     this.objectMapper = JsonMapper.builder().build();
   }
@@ -107,6 +111,7 @@ public PolarisAuthorizer create(RealmConfig realmConfig) {
         httpClient,
         objectMapper,
         bearerTokenProvider,
+        requestIdSupplier.getRequestId(),
         realmContext.getRealmIdentifier());
   }
 

extensions/auth/opa/src/test/java/org/apache/polaris/extension/auth/opa/OpaPolarisAuthorizerFactoryTest.java

@@ -37,12 +37,15 @@
 import java.nio.file.Path;
 import java.time.Clock;
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.polaris.core.auth.PolarisAuthorizableOperation;
 import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
 import org.apache.polaris.extension.auth.opa.token.FileBearerTokenProvider;
 import org.apache.polaris.nosql.async.java.JavaPoolAsyncExec;
@@ -80,7 +83,8 @@ public void testFactoryWithStaticTokenConfiguration() {
     try (JavaPoolAsyncExec asyncExec = new JavaPoolAsyncExec()) {
       RealmContext realmContext = () -> "test-realm";
       OpaPolarisAuthorizerFactory factory =
-          new OpaPolarisAuthorizerFactory(opaConfig, Clock.systemUTC(), asyncExec, realmContext);
+          new OpaPolarisAuthorizerFactory(
+              opaConfig, Clock.systemUTC(), asyncExec, () -> null, realmContext);
 
       // Create authorizer
       RealmConfig realmConfig = mock(RealmConfig.class);
@@ -125,7 +129,8 @@ public void testFactoryWithFileBasedTokenConfiguration() throws IOException {
     try (JavaPoolAsyncExec asyncExec = new JavaPoolAsyncExec()) {
       RealmContext realmContext = () -> "test-realm";
       OpaPolarisAuthorizerFactory factory =
-          new OpaPolarisAuthorizerFactory(opaConfig, Clock.systemUTC(), asyncExec, realmContext);
+          new OpaPolarisAuthorizerFactory(
+              opaConfig, Clock.systemUTC(), asyncExec, () -> null, realmContext);
 
       // Create authorizer
       RealmConfig realmConfig = mock(RealmConfig.class);
@@ -171,7 +176,8 @@ public void testFactoryWithNoTokenConfiguration() {
     try (JavaPoolAsyncExec asyncExec = new JavaPoolAsyncExec()) {
       RealmContext realmContext = () -> "test-realm";
       OpaPolarisAuthorizerFactory factory =
-          new OpaPolarisAuthorizerFactory(opaConfig, Clock.systemUTC(), asyncExec, realmContext);
+          new OpaPolarisAuthorizerFactory(
+              opaConfig, Clock.systemUTC(), asyncExec, () -> null, realmContext);
 
       // Create authorizer
       RealmConfig realmConfig = mock(RealmConfig.class);
@@ -208,7 +214,8 @@ public void testFactoryPassesRealmToAuthorizerContext() throws Exception {
       try (JavaPoolAsyncExec asyncExec = new JavaPoolAsyncExec()) {
         RealmContext realmContext = () -> "factory-realm";
         OpaPolarisAuthorizerFactory factory =
-            new OpaPolarisAuthorizerFactory(opaConfig, Clock.systemUTC(), asyncExec, realmContext);
+            new OpaPolarisAuthorizerFactory(
+                opaConfig, Clock.systemUTC(), asyncExec, () -> null, realmContext);
 
         factory.initialize();
         OpaPolarisAuthorizer authorizer =
@@ -263,7 +270,8 @@ public void testFactoryUsesDistinctRealmValues() throws Exception {
       try (JavaPoolAsyncExec asyncExec = new JavaPoolAsyncExec()) {
         RealmContext realmContext = () -> "realm-b";
         OpaPolarisAuthorizerFactory factory =
-            new OpaPolarisAuthorizerFactory(opaConfig, Clock.systemUTC(), asyncExec, realmContext);
+            new OpaPolarisAuthorizerFactory(
+                opaConfig, Clock.systemUTC(), asyncExec, () -> null, realmContext);
 
         factory.initialize();
         OpaPolarisAuthorizer authorizer =
@@ -290,6 +298,148 @@ public void testFactoryUsesDistinctRealmValues() throws Exception {
     }
   }
 
+  @Test
+  public void testFactoryPassesResolvedRequestIdToAuthorizerContext() throws Exception {
+    final String[] capturedRequestBody = new String[1];
+
+    HttpServer server = createServerWithRequestCapture(capturedRequestBody);
+    try {
+      OpaAuthorizationConfig opaConfig =
+          ImmutableOpaAuthorizationConfig.builder()
+              .policyUri(
+                  URI.create(
+                      "http://localhost:"
+                          + server.getAddress().getPort()
+                          + "/v1/data/polaris/allow"))
+              .auth(
+                  ImmutableAuthenticationConfig.builder()
+                      .type(OpaAuthorizationConfig.AuthenticationType.NONE)
+                      .build())
+              .http(
+                  ImmutableHttpConfig.builder()
+                      .timeout(Duration.ofSeconds(2))
+                      .verifySsl(true)
+                      .build())
+              .build();
+
+      try (JavaPoolAsyncExec asyncExec = new JavaPoolAsyncExec()) {
+        RealmContext realmContext = () -> "test-realm";
+        OpaPolarisAuthorizerFactory factory =
+            new OpaPolarisAuthorizerFactory(
+                opaConfig,
+                Clock.systemUTC(),
+                asyncExec,
+                () -> "factory-test-request-id",
+                realmContext);
+        factory.initialize();
+
+        RealmConfig realmConfig = mock(RealmConfig.class);
+        OpaPolarisAuthorizer authorizer = (OpaPolarisAuthorizer) factory.create(realmConfig);
+
+        PolarisPrincipal principal =
+            PolarisPrincipal.of("eve", Map.of("department", "finance"), Set.of("auditor"));
+        PolarisResolvedPathWrapper target = new PolarisResolvedPathWrapper(List.of());
+        PolarisResolvedPathWrapper secondary = new PolarisResolvedPathWrapper(List.of());
+
+        assertThatNoException()
+            .isThrownBy(
+                () ->
+                    authorizer.authorizeOrThrow(
+                        principal,
+                        Set.<PolarisBaseEntity>of(),
+                        PolarisAuthorizableOperation.LOAD_VIEW,
+                        target,
+                        secondary));
+
+        JsonNode root = JsonMapper.builder().build().readTree(capturedRequestBody[0]);
+        assertThat(root.at("/input/context/request_id").asText())
+            .isEqualTo("factory-test-request-id");
+      }
+    } finally {
+      server.stop(0);
+    }
+  }
+
+  @Test
+  public void testFactoryResolvesFreshRequestIdPerCreateCall() throws Exception {
+    final String[] capturedRequestBody = new String[1];
+
+    HttpServer server = createServerWithRequestCapture(capturedRequestBody);
+    try {
+      OpaAuthorizationConfig opaConfig =
+          ImmutableOpaAuthorizationConfig.builder()
+              .policyUri(
+                  URI.create(
+                      "http://localhost:"
+                          + server.getAddress().getPort()
+                          + "/v1/data/polaris/allow"))
+              .auth(
+                  ImmutableAuthenticationConfig.builder()
+                      .type(OpaAuthorizationConfig.AuthenticationType.NONE)
+                      .build())
+              .http(
+                  ImmutableHttpConfig.builder()
+                      .timeout(Duration.ofSeconds(2))
+                      .verifySsl(true)
+                      .build())
+              .build();
+
+      // Simulates the CDI proxy resolving to a different value on each request, the way the real
+      // RequestIdSupplier does via CurrentRequestManager.
+      AtomicInteger requestCounter = new AtomicInteger();
+      try (JavaPoolAsyncExec asyncExec = new JavaPoolAsyncExec()) {
+        RealmContext realmContext = () -> "test-realm";
+        OpaPolarisAuthorizerFactory factory =
+            new OpaPolarisAuthorizerFactory(
+                opaConfig,
+                Clock.systemUTC(),
+                asyncExec,
+                () -> "request-" + requestCounter.incrementAndGet(),
+                realmContext);
+
+        factory.initialize();
+
+        RealmConfig realmConfig = mock(RealmConfig.class);
+        PolarisPrincipal principal =
+            PolarisPrincipal.of("eve", Map.of("department", "finance"), Set.of("auditor"));
+        PolarisResolvedPathWrapper target = new PolarisResolvedPathWrapper(List.of());
+        PolarisResolvedPathWrapper secondary = new PolarisResolvedPathWrapper(List.of());
+
+        // First "request": factory.create() is called fresh, as it would be for each incoming
+        // HTTP request via the @RequestScoped PolarisAuthorizer producer.
+        OpaPolarisAuthorizer firstAuthorizer = (OpaPolarisAuthorizer) factory.create(realmConfig);
+        assertThatNoException()
+            .isThrownBy(
+                () ->
+                    firstAuthorizer.authorizeOrThrow(
+                        principal,
+                        Set.<PolarisBaseEntity>of(),
+                        PolarisAuthorizableOperation.LOAD_VIEW,
+                        target,
+                        secondary));
+        JsonNode firstRoot = JsonMapper.builder().build().readTree(capturedRequestBody[0]);
+        assertThat(firstRoot.at("/input/context/request_id").asText()).isEqualTo("request-1");
+
+        // Second "request": a new authorizer is created against the same long-lived factory,
+        // and must pick up a distinct, freshly-resolved request ID rather than reusing the first.
+        OpaPolarisAuthorizer secondAuthorizer = (OpaPolarisAuthorizer) factory.create(realmConfig);
+        assertThatNoException()
+            .isThrownBy(
+                () ->
+                    secondAuthorizer.authorizeOrThrow(
+                        principal,
+                        Set.<PolarisBaseEntity>of(),
+                        PolarisAuthorizableOperation.LOAD_VIEW,
+                        target,
+                        secondary));
+        JsonNode secondRoot = JsonMapper.builder().build().readTree(capturedRequestBody[0]);
+        assertThat(secondRoot.at("/input/context/request_id").asText()).isEqualTo("request-2");
+      }
+    } finally {
+      server.stop(0);
+    }
+  }
+
   private HttpServer createServerWithRequestCapture(String[] capturedRequestBody)
       throws IOException {
     HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);