Added interface for reporting metrics (#2887)

Co-authored-by: Alexandre Dutra <[email protected]>

Author: cccs-cat001

CHANGELOG.md

@@ -29,6 +29,13 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 
 ### Highlights
 
+- Support for [Iceberg Metrics Reporting] has been introduced in Polaris. Out of the box, metrics can
+  be printed to the logs by setting the `org.apache.polaris.service.reporting` logger level to `INFO` (it's
+  set to `OFF` by default). Custom reporters can be implemented and configured to send metrics to
+  external systems for further analysis and monitoring.
+
+[Iceberg Metrics Reporting]: https://iceberg.apache.org/docs/latest/metrics-reporting/
+
 ### Upgrade notes
 
 - The legacy management endpoints at `/metrics` and `/healthcheck` have been removed. Please use the

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java

@@ -65,9 +65,14 @@
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.ResolvingFileIO;
+import org.apache.iceberg.metrics.ImmutableScanReport;
+import org.apache.iceberg.metrics.ScanMetrics;
+import org.apache.iceberg.metrics.ScanMetricsResult;
+import org.apache.iceberg.metrics.ScanReport;
 import org.apache.iceberg.rest.RESTCatalog;
 import org.apache.iceberg.rest.RESTUtil;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.ReportMetricsRequest;
 import org.apache.iceberg.rest.responses.ErrorResponse;
 import org.apache.iceberg.rest.responses.ListNamespacesResponse;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
@@ -882,6 +887,27 @@ public void testCreateAndLoadTableWithReturnedEtag() {
     }
   }
 
+  @Test
+  public void testSendMetricsReport() {
+    ScanReport scanReport =
+        ImmutableScanReport.builder()
+            .tableName("tbl1")
+            .schemaId(4)
+            .addProjectedFieldIds(1, 2, 3)
+            .addProjectedFieldNames("c1", "c2", "c3")
+            .snapshotId(23L)
+            .filter(Expressions.alwaysTrue())
+            .scanMetrics(ScanMetricsResult.fromScanMetrics(ScanMetrics.noop()))
+            .build();
+    Invocation.Builder metricEndpoint =
+        catalogApi.request(
+            "v1/{cat}/namespaces/ns1/tables/tbl1/metrics", Map.of("cat", currentCatalogName));
+    try (Response response =
+        metricEndpoint.post(Entity.json(ReportMetricsRequest.of(scanReport)))) {
+      assertThat(response).returns(Response.Status.NO_CONTENT.getStatusCode(), Response::getStatus);
+    }
+  }
+
   @Test
   public void testSendNotificationInternalCatalog() {
     Map<String, String> payload =

runtime/defaults/src/main/resources/application.properties

@@ -242,6 +242,11 @@ polaris.oidc.principal-roles-mapper.type=default
 # Polaris Credential Manager Config
 polaris.credential-manager.type=default
 
+# Configuration for the behaviour of the metrics endpoint
+polaris.iceberg-metrics.reporting.type=default
+# Set to INFO if you want to see iceberg metric reports logged
+quarkus.log.category."org.apache.polaris.service.reporting".level=OFF
+
 quarkus.arc.ignored-split-packages=\
   org.apache.polaris.service.catalog.api,\
   org.apache.polaris.service.catalog.api.impl,\

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

@@ -87,6 +87,7 @@
 import org.apache.polaris.service.events.listeners.PolarisEventListener;
 import org.apache.polaris.service.http.IcebergHttpUtil;
 import org.apache.polaris.service.http.IfNoneMatch;
+import org.apache.polaris.service.reporting.PolarisMetricsReporter;
 import org.apache.polaris.service.types.CommitTableRequest;
 import org.apache.polaris.service.types.CommitViewRequest;
 import org.apache.polaris.service.types.NotificationRequest;
@@ -150,6 +151,7 @@ public class IcebergCatalogAdapter
   private final Instance<ExternalCatalogFactory> externalCatalogFactories;
   private final PolarisEventListener polarisEventListener;
   private final AccessConfigProvider accessConfigProvider;
+  private final PolarisMetricsReporter metricsReporter;
 
   @Inject
   public IcebergCatalogAdapter(
@@ -167,7 +169,8 @@ public IcebergCatalogAdapter(
       CatalogHandlerUtils catalogHandlerUtils,
       @Any Instance<ExternalCatalogFactory> externalCatalogFactories,
       PolarisEventListener polarisEventListener,
-      AccessConfigProvider accessConfigProvider) {
+      AccessConfigProvider accessConfigProvider,
+      PolarisMetricsReporter metricsReporter) {
     this.diagnostics = diagnostics;
     this.realmContext = realmContext;
     this.callContext = callContext;
@@ -184,6 +187,7 @@ public IcebergCatalogAdapter(
     this.externalCatalogFactories = externalCatalogFactories;
     this.polarisEventListener = polarisEventListener;
     this.accessConfigProvider = accessConfigProvider;
+    this.metricsReporter = metricsReporter;
   }
 
   /**
@@ -755,6 +759,11 @@ public Response reportMetrics(
       ReportMetricsRequest reportMetricsRequest,
       RealmContext realmContext,
       SecurityContext securityContext) {
+    String catalogName = prefixParser.prefixToCatalogName(realmContext, prefix);
+    Namespace ns = decodeNamespace(namespace);
+    TableIdentifier tableIdentifier = TableIdentifier.of(ns, RESTUtil.decodeString(table));
+
+    metricsReporter.reportMetric(catalogName, tableIdentifier, reportMetricsRequest.report());
     return Response.status(Response.Status.NO_CONTENT).build();
   }
 

runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java

@@ -83,6 +83,8 @@
 import org.apache.polaris.service.ratelimiter.RateLimiterFilterConfiguration;
 import org.apache.polaris.service.ratelimiter.TokenBucketConfiguration;
 import org.apache.polaris.service.ratelimiter.TokenBucketFactory;
+import org.apache.polaris.service.reporting.MetricsReportingConfiguration;
+import org.apache.polaris.service.reporting.PolarisMetricsReporter;
 import org.apache.polaris.service.secrets.SecretsManagerConfiguration;
 import org.apache.polaris.service.storage.StorageConfiguration;
 import org.apache.polaris.service.storage.aws.S3AccessConfig;
@@ -441,4 +443,11 @@ public PolarisCredentialManager polarisCredentialManager(
   public void closeTaskExecutor(@Disposes @Identifier("task-executor") ManagedExecutor executor) {
     executor.close();
   }
+
+  @Produces
+  @ApplicationScoped
+  public PolarisMetricsReporter metricsReporter(
+      MetricsReportingConfiguration config, @Any Instance<PolarisMetricsReporter> reporters) {
+    return reporters.select(Identifier.Literal.of(config.type())).get();
+  }
 }

runtime/service/src/main/java/org/apache/polaris/service/reporting/DefaultMetricsReporter.java

@@ -0,0 +1,52 @@
+/*
+ * 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.reporting;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.ApplicationScoped;
+import org.apache.commons.lang3.function.TriConsumer;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.metrics.MetricsReport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@ApplicationScoped
+@Identifier("default")
+public class DefaultMetricsReporter implements PolarisMetricsReporter {
+  private static final Logger LOGGER = LoggerFactory.getLogger(DefaultMetricsReporter.class);
+
+  private final TriConsumer<String, TableIdentifier, MetricsReport> reportConsumer;
+
+  public DefaultMetricsReporter() {
+    this(
+        (catalogName, table, metricsReport) ->
+            LOGGER.info("{}.{}: {}", catalogName, table, metricsReport));
+  }
+
+  @VisibleForTesting
+  DefaultMetricsReporter(TriConsumer<String, TableIdentifier, MetricsReport> reportConsumer) {
+    this.reportConsumer = reportConsumer;
+  }
+
+  @Override
+  public void reportMetric(String catalogName, TableIdentifier table, MetricsReport metricsReport) {
+    reportConsumer.accept(catalogName, table, metricsReport);
+  }
+}

runtime/service/src/main/java/org/apache/polaris/service/reporting/MetricsReportingConfiguration.java

@@ -0,0 +1,28 @@
+/*
+ * 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.reporting;
+
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+
+@ConfigMapping(prefix = "polaris.iceberg-metrics.reporting")
+public interface MetricsReportingConfiguration {
+  @WithDefault("default")
+  String type();
+}

runtime/service/src/main/java/org/apache/polaris/service/reporting/PolarisMetricsReporter.java

@@ -0,0 +1,26 @@
+/*
+ * 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.reporting;
+
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.metrics.MetricsReport;
+
+public interface PolarisMetricsReporter {
+  public void reportMetric(String catalogName, TableIdentifier table, MetricsReport metricsReport);
+}

runtime/service/src/test/java/org/apache/polaris/service/reporting/DefaultMetricsReporterTest.java

@@ -0,0 +1,43 @@
+/*
+ * 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.reporting;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import org.apache.commons.lang3.function.TriConsumer;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.metrics.MetricsReport;
+import org.junit.jupiter.api.Test;
+
+public class DefaultMetricsReporterTest {
+
+  @Test
+  void testLogging() {
+    TriConsumer<String, TableIdentifier, MetricsReport> mockConsumer = mock(TriConsumer.class);
+    DefaultMetricsReporter reporter = new DefaultMetricsReporter(mockConsumer);
+    String warehouse = "testWarehouse";
+    TableIdentifier table = TableIdentifier.of("testNamespace", "testTable");
+    MetricsReport metricsReport = mock(MetricsReport.class);
+
+    reporter.reportMetric(warehouse, table, metricsReport);
+
+    verify(mockConsumer).accept(warehouse, table, metricsReport);
+  }
+}

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

@@ -82,6 +82,7 @@
 import org.apache.polaris.service.events.listeners.TestPolarisEventListener;
 import org.apache.polaris.service.identity.provider.DefaultServiceIdentityProvider;
 import org.apache.polaris.service.persistence.InMemoryPolarisMetaStoreManagerFactory;
+import org.apache.polaris.service.reporting.DefaultMetricsReporter;
 import org.apache.polaris.service.secrets.UnsafeInMemorySecretsManagerFactory;
 import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
 import org.apache.polaris.service.task.TaskExecutor;
@@ -281,7 +282,8 @@ public TestServices build() {
               catalogHandlerUtils,
               externalCatalogFactory,
               polarisEventListener,
-              accessConfigProvider);
+              accessConfigProvider,
+              new DefaultMetricsReporter());
 
       IcebergRestCatalogApi restApi = new IcebergRestCatalogApi(catalogService);
       IcebergRestConfigurationApi restConfigurationApi =