feat(metrics): Evolve PolarisMetricsReporter interface with timestamp parameter and comprehensive documentation

CHANGELOG.md

@@ -55,6 +55,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 
 ### New Features
 
+- Added `ENABLE_METRICS_EVENT_EMISSION` feature flag (default: false) to control the emission of `BEFORE_REPORT_METRICS` and `AFTER_REPORT_METRICS` events when the Iceberg REST catalog API's `reportMetrics()` method is called. When enabled, event listeners can receive metrics report data for use cases like audit logging and metrics persistence. Can be configured via `polaris.features."ENABLE_METRICS_EVENT_EMISSION"=true`.
 - Added `--no-sts` flag to CLI to support S3-compatible storage systems that do not have Security Token Service available.
 - Support credential vending for federated catalogs. `ALLOW_FEDERATED_CATALOGS_CREDENTIAL_VENDING` (default: true) was added to toggle this feature.
 - Enhanced catalog federation with SigV4 authentication support, additional authentication types for credential vending, and location-based access restrictions to block credential vending for remote tables outside allowed location lists.

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -516,4 +516,23 @@ public static void enforceFeatureEnabledOrThrow(
                   + "Helps prevent thundering herd when multiple requests fail simultaneously.")
           .defaultValue(0.5)
           .buildFeatureConfiguration();
+
+  /**
+   * Feature flag to control the emission of BEFORE_REPORT_METRICS and AFTER_REPORT_METRICS events
+   * when the Iceberg REST catalog API's reportMetrics() method is called. When disabled (default),
+   * the reportMetrics() method calls the delegate directly without emitting any events. When
+   * enabled, BEFORE_REPORT_METRICS and AFTER_REPORT_METRICS events are emitted, allowing event
+   * listeners to receive metrics report data for use cases like audit logging and metrics
+   * persistence.
+   */
+  public static final FeatureConfiguration<Boolean> ENABLE_METRICS_EVENT_EMISSION =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ENABLE_METRICS_EVENT_EMISSION")
+          .description(
+              "If set to true, emit BEFORE_REPORT_METRICS and AFTER_REPORT_METRICS events when "
+                  + "the reportMetrics() API is called. This enables event listeners to receive "
+                  + "metrics report data for use cases like audit logging and metrics persistence. "
+                  + "Defaults to false to ensure backward compatibility.")
+          .defaultValue(false)
+          .buildFeatureConfiguration();
 }

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRestCatalogEventServiceDelegator.java

@@ -41,6 +41,8 @@
 import org.apache.iceberg.rest.responses.LoadTableResponse;
 import org.apache.iceberg.rest.responses.LoadViewResponse;
 import org.apache.iceberg.rest.responses.UpdateNamespacePropertiesResponse;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.service.catalog.CatalogPrefixParser;
 import org.apache.polaris.service.catalog.api.IcebergRestCatalogApiService;
@@ -64,18 +66,21 @@ public class IcebergRestCatalogEventServiceDelegator
   @Inject PolarisEventListener polarisEventListener;
   @Inject PolarisEventMetadataFactory eventMetadataFactory;
   @Inject CatalogPrefixParser prefixParser;
+  @Inject RealmConfig realmConfig;
 
   // Constructor for testing - allows manual dependency injection
   @VisibleForTesting
   public IcebergRestCatalogEventServiceDelegator(
       IcebergCatalogAdapter delegate,
       PolarisEventListener polarisEventListener,
       PolarisEventMetadataFactory eventMetadataFactory,
-      CatalogPrefixParser prefixParser) {
+      CatalogPrefixParser prefixParser,
+      RealmConfig realmConfig) {
     this.delegate = delegate;
     this.polarisEventListener = polarisEventListener;
     this.eventMetadataFactory = eventMetadataFactory;
     this.prefixParser = prefixParser;
+    this.realmConfig = realmConfig;
   }
 
   // Default constructor for CDI
@@ -805,8 +810,41 @@ public Response reportMetrics(
       ReportMetricsRequest reportMetricsRequest,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    return delegate.reportMetrics(
-        prefix, namespace, table, reportMetricsRequest, realmContext, securityContext);
+    // Check if metrics event emission is enabled
+    boolean metricsEventEmissionEnabled =
+        realmConfig.getConfig(FeatureConfiguration.ENABLE_METRICS_EVENT_EMISSION);
+
+    // If metrics event emission is disabled, call delegate directly without emitting events
+    if (!metricsEventEmissionEnabled) {
+      return delegate.reportMetrics(
+          prefix, namespace, table, reportMetricsRequest, realmContext, securityContext);
+    }
+
+    // Emit events when feature is enabled
+    String catalogName = prefixParser.prefixToCatalogName(realmContext, prefix);
+    Namespace namespaceObj = decodeNamespace(namespace);
+    polarisEventListener.onEvent(
+        new PolarisEvent(
+            PolarisEventType.BEFORE_REPORT_METRICS,
+            eventMetadataFactory.create(),
+            new AttributeMap()
+                .put(EventAttributes.CATALOG_NAME, catalogName)
+                .put(EventAttributes.NAMESPACE, namespaceObj)
+                .put(EventAttributes.TABLE_NAME, table)
+                .put(EventAttributes.REPORT_METRICS_REQUEST, reportMetricsRequest)));
+    Response resp =
+        delegate.reportMetrics(
+            prefix, namespace, table, reportMetricsRequest, realmContext, securityContext);
+    polarisEventListener.onEvent(
+        new PolarisEvent(
+            PolarisEventType.AFTER_REPORT_METRICS,
+            eventMetadataFactory.create(),
+            new AttributeMap()
+                .put(EventAttributes.CATALOG_NAME, catalogName)
+                .put(EventAttributes.NAMESPACE, namespaceObj)
+                .put(EventAttributes.TABLE_NAME, table)
+                .put(EventAttributes.REPORT_METRICS_REQUEST, reportMetricsRequest)));
+    return resp;
   }
 
   @Override

runtime/service/src/main/java/org/apache/polaris/service/events/EventAttributes.java

@@ -28,6 +28,7 @@
 import org.apache.iceberg.rest.requests.CreateViewRequest;
 import org.apache.iceberg.rest.requests.RegisterTableRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.iceberg.rest.requests.ReportMetricsRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
 import org.apache.iceberg.rest.responses.ConfigResponse;
@@ -230,4 +231,8 @@ private EventAttributes() {}
       new AttributeKey<>("detach_policy_request", DetachPolicyRequest.class);
   public static final AttributeKey<GetApplicablePoliciesResponse> GET_APPLICABLE_POLICIES_RESPONSE =
       new AttributeKey<>("get_applicable_policies_response", GetApplicablePoliciesResponse.class);
+
+  // Metrics reporting attributes
+  public static final AttributeKey<ReportMetricsRequest> REPORT_METRICS_REQUEST =
+      new AttributeKey<>("report_metrics_request", ReportMetricsRequest.class);
 }

runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventType.java

@@ -200,4 +200,8 @@ public enum PolarisEventType {
 
   // Rate Limiting Events
   BEFORE_LIMIT_REQUEST_RATE,
+
+  // Metrics Reporting Events
+  BEFORE_REPORT_METRICS,
+  AFTER_REPORT_METRICS,
 }

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/ReportMetricsEventTest.java

@@ -0,0 +1,238 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog.iceberg;
+
+import static org.apache.polaris.service.admin.PolarisAuthzTestBase.SCHEMA;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import jakarta.ws.rs.core.Response;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.metrics.ImmutableScanReport;
+import org.apache.iceberg.metrics.ScanMetrics;
+import org.apache.iceberg.metrics.ScanMetricsResult;
+import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.ReportMetricsRequest;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.CatalogProperties;
+import org.apache.polaris.core.admin.model.CreateCatalogRequest;
+import org.apache.polaris.core.admin.model.FileStorageConfigInfo;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.service.TestServices;
+import org.apache.polaris.service.events.EventAttributes;
+import org.apache.polaris.service.events.PolarisEvent;
+import org.apache.polaris.service.events.PolarisEventType;
+import org.apache.polaris.service.events.listeners.TestPolarisEventListener;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Unit tests for verifying that reportMetrics() emits BEFORE_REPORT_METRICS and
+ * AFTER_REPORT_METRICS events.
+ */
+public class ReportMetricsEventTest {
+  private static final String NAMESPACE = "test_ns";
+  private static final String CATALOG = "test-catalog";
+  private static final String TABLE = "test-table";
+
+  private String catalogLocation;
+
+  @BeforeEach
+  public void setUp(@TempDir Path tempDir) {
+    catalogLocation = tempDir.toAbsolutePath().toUri().toString();
+    if (catalogLocation.endsWith("/")) {
+      catalogLocation = catalogLocation.substring(0, catalogLocation.length() - 1);
+    }
+  }
+
+  @Test
+  void testReportMetricsEmitsBeforeAndAfterEventsWhenEnabled() {
+    // Create test services with ENABLE_METRICS_EVENT_EMISSION enabled
+    TestServices testServices = createTestServicesWithMetricsEmissionEnabled(true);
+    createCatalogAndNamespace(testServices);
+    createTable(testServices, TABLE);
+
+    // Create a ScanReport for testing
+    ImmutableScanReport scanReport =
+        ImmutableScanReport.builder()
+            .schemaId(0)
+            .tableName(NAMESPACE + "." + TABLE)
+            .snapshotId(100L)
+            .addProjectedFieldIds(1)
+            .addProjectedFieldNames("id")
+            .filter(Expressions.alwaysTrue())
+            .scanMetrics(ScanMetricsResult.fromScanMetrics(ScanMetrics.noop()))
+            .build();
+
+    ReportMetricsRequest request = ReportMetricsRequest.of(scanReport);
+
+    // Call reportMetrics
+    try (Response response =
+        testServices
+            .restApi()
+            .reportMetrics(
+                CATALOG,
+                NAMESPACE,
+                TABLE,
+                request,
+                testServices.realmContext(),
+                testServices.securityContext())) {
+      assertThat(response.getStatus()).isEqualTo(Response.Status.NO_CONTENT.getStatusCode());
+    }
+
+    // Verify that BEFORE_REPORT_METRICS and AFTER_REPORT_METRICS events were emitted
+    TestPolarisEventListener testEventListener =
+        (TestPolarisEventListener) testServices.polarisEventListener();
+
+    PolarisEvent beforeEvent = testEventListener.getLatest(PolarisEventType.BEFORE_REPORT_METRICS);
+    assertThat(beforeEvent).isNotNull();
+    assertThat(beforeEvent.attributes().getRequired(EventAttributes.CATALOG_NAME))
+        .isEqualTo(CATALOG);
+    assertThat(beforeEvent.attributes().getRequired(EventAttributes.NAMESPACE))
+        .isEqualTo(Namespace.of(NAMESPACE));
+    assertThat(beforeEvent.attributes().getRequired(EventAttributes.TABLE_NAME)).isEqualTo(TABLE);
+    assertThat(beforeEvent.attributes().getRequired(EventAttributes.REPORT_METRICS_REQUEST))
+        .isNotNull();
+
+    PolarisEvent afterEvent = testEventListener.getLatest(PolarisEventType.AFTER_REPORT_METRICS);
+    assertThat(afterEvent).isNotNull();
+    assertThat(afterEvent.attributes().getRequired(EventAttributes.CATALOG_NAME))
+        .isEqualTo(CATALOG);
+    assertThat(afterEvent.attributes().getRequired(EventAttributes.NAMESPACE))
+        .isEqualTo(Namespace.of(NAMESPACE));
+    assertThat(afterEvent.attributes().getRequired(EventAttributes.TABLE_NAME)).isEqualTo(TABLE);
+    assertThat(afterEvent.attributes().getRequired(EventAttributes.REPORT_METRICS_REQUEST))
+        .isNotNull();
+  }
+
+  @Test
+  void testReportMetricsDoesNotEmitEventsWhenDisabled() {
+    // Create test services with ENABLE_METRICS_EVENT_EMISSION disabled (default)
+    TestServices testServices = createTestServicesWithMetricsEmissionEnabled(false);
+    createCatalogAndNamespace(testServices);
+    createTable(testServices, TABLE);
+
+    // Create a ScanReport for testing
+    ImmutableScanReport scanReport =
+        ImmutableScanReport.builder()
+            .schemaId(0)
+            .tableName(NAMESPACE + "." + TABLE)
+            .snapshotId(100L)
+            .addProjectedFieldIds(1)
+            .addProjectedFieldNames("id")
+            .filter(Expressions.alwaysTrue())
+            .scanMetrics(ScanMetricsResult.fromScanMetrics(ScanMetrics.noop()))
+            .build();
+
+    ReportMetricsRequest request = ReportMetricsRequest.of(scanReport);
+
+    // Call reportMetrics
+    try (Response response =
+        testServices
+            .restApi()
+            .reportMetrics(
+                CATALOG,
+                NAMESPACE,
+                TABLE,
+                request,
+                testServices.realmContext(),
+                testServices.securityContext())) {
+      assertThat(response.getStatus()).isEqualTo(Response.Status.NO_CONTENT.getStatusCode());
+    }
+
+    // Verify that BEFORE_REPORT_METRICS and AFTER_REPORT_METRICS events were NOT emitted
+    TestPolarisEventListener testEventListener =
+        (TestPolarisEventListener) testServices.polarisEventListener();
+
+    assertThat(testEventListener.hasEvent(PolarisEventType.BEFORE_REPORT_METRICS)).isFalse();
+    assertThat(testEventListener.hasEvent(PolarisEventType.AFTER_REPORT_METRICS)).isFalse();
+  }
+
+  private TestServices createTestServicesWithMetricsEmissionEnabled(boolean enabled) {
+    Map<String, Object> config =
+        Map.of(
+            "ALLOW_INSECURE_STORAGE_TYPES",
+            "true",
+            "SUPPORTED_CATALOG_STORAGE_TYPES",
+            List.of("FILE"),
+            "ENABLE_METRICS_EVENT_EMISSION",
+            String.valueOf(enabled));
+    return TestServices.builder().config(config).withEventDelegator(true).build();
+  }
+
+  private void createCatalogAndNamespace(TestServices services) {
+    CatalogProperties.Builder propertiesBuilder =
+        CatalogProperties.builder()
+            .setDefaultBaseLocation(String.format("%s/%s", catalogLocation, CATALOG));
+
+    StorageConfigInfo config =
+        FileStorageConfigInfo.builder()
+            .setStorageType(StorageConfigInfo.StorageTypeEnum.FILE)
+            .build();
+    Catalog catalogObject =
+        new Catalog(
+            Catalog.TypeEnum.INTERNAL, CATALOG, propertiesBuilder.build(), 0L, 0L, 1, config);
+    try (Response response =
+        services
+            .catalogsApi()
+            .createCatalog(
+                new CreateCatalogRequest(catalogObject),
+                services.realmContext(),
+                services.securityContext())) {
+      assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode());
+    }
+
+    CreateNamespaceRequest createNamespaceRequest =
+        CreateNamespaceRequest.builder().withNamespace(Namespace.of(NAMESPACE)).build();
+    try (Response response =
+        services
+            .restApi()
+            .createNamespace(
+                CATALOG,
+                createNamespaceRequest,
+                services.realmContext(),
+                services.securityContext())) {
+      assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode());
+    }
+  }
+
+  private void createTable(TestServices services, String tableName) {
+    CreateTableRequest createTableRequest =
+        CreateTableRequest.builder()
+            .withName(tableName)
+            .withLocation(
+                String.format("%s/%s/%s/%s", catalogLocation, CATALOG, NAMESPACE, tableName))
+            .withSchema(SCHEMA)
+            .build();
+    services
+        .restApi()
+        .createTable(
+            CATALOG,
+            NAMESPACE,
+            createTableRequest,
+            null,
+            services.realmContext(),
+            services.securityContext());
+  }
+}

runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java

@@ -361,7 +361,8 @@ public String getAuthenticationScheme() {
                 catalogService,
                 polarisEventListener,
                 eventMetadataFactory,
-                new DefaultCatalogPrefixParser());
+                new DefaultCatalogPrefixParser(),
+                realmConfig);
         finalRestConfigurationService =
             new IcebergRestConfigurationEventServiceDelegator(
                 catalogService, polarisEventListener, eventMetadataFactory);