Added interface for reporting metrics

Author: cccs-cat001

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.MetricsReporter;
 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 MetricsReporter metricsReporter;
 
   @Inject
   public IcebergCatalogAdapter(
@@ -167,7 +169,8 @@ public IcebergCatalogAdapter(
       CatalogHandlerUtils catalogHandlerUtils,
       @Any Instance<ExternalCatalogFactory> externalCatalogFactories,
       PolarisEventListener polarisEventListener,
-      AccessConfigProvider accessConfigProvider) {
+      AccessConfigProvider accessConfigProvider,
+      MetricsReporter 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;
   }
 
   /**
@@ -224,7 +228,8 @@ IcebergCatalogHandler newHandlerWrapper(SecurityContext securityContext, String
         catalogHandlerUtils,
         externalCatalogFactories,
         polarisEventListener,
-        accessConfigProvider);
+        accessConfigProvider,
+        metricsReporter);
   }
 
   @Override
@@ -755,7 +760,13 @@ public Response reportMetrics(
       ReportMetricsRequest reportMetricsRequest,
       RealmContext realmContext,
       SecurityContext securityContext) {
-    return Response.status(Response.Status.NO_CONTENT).build();
+    return withCatalog(
+        securityContext,
+        prefix,
+        catalog -> {
+          catalog.reportMetric(prefix, namespace, table, reportMetricsRequest);
+          return Response.status(Response.Status.NO_CONTENT).build();
+        });
   }
 
   @Override

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

@@ -68,6 +68,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.CreateNamespaceResponse;
@@ -112,6 +113,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.MetricsReporter;
 import org.apache.polaris.service.types.NotificationRequest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -140,7 +142,7 @@ public class IcebergCatalogHandler extends CatalogHandler implements AutoCloseab
   private final CatalogHandlerUtils catalogHandlerUtils;
   private final PolarisEventListener polarisEventListener;
   private final AccessConfigProvider accessConfigProvider;
-
+  private final MetricsReporter metricsReporter;
   // Catalog instance will be initialized after authorizing resolver successfully resolves
   // the catalog entity.
   protected Catalog baseCatalog = null;
@@ -164,7 +166,8 @@ public IcebergCatalogHandler(
       CatalogHandlerUtils catalogHandlerUtils,
       Instance<ExternalCatalogFactory> externalCatalogFactories,
       PolarisEventListener polarisEventListener,
-      AccessConfigProvider accessConfigProvider) {
+      AccessConfigProvider accessConfigProvider,
+      MetricsReporter metricsReporter) {
     super(
         diagnostics,
         callContext,
@@ -180,6 +183,7 @@ public IcebergCatalogHandler(
     this.catalogHandlerUtils = catalogHandlerUtils;
     this.polarisEventListener = polarisEventListener;
     this.accessConfigProvider = accessConfigProvider;
+    this.metricsReporter = metricsReporter;
   }
 
   private CatalogEntity getResolvedCatalogEntity() {
@@ -589,6 +593,11 @@ public LoadTableResponse registerTable(Namespace namespace, RegisterTableRequest
     return catalogHandlerUtils.registerTable(baseCatalog, namespace, request);
   }
 
+  public void reportMetric(
+      String prefix, String namespace, String table, ReportMetricsRequest reportMetricsRequest) {
+    metricsReporter.reportMetric(prefix, namespace, table, reportMetricsRequest);
+  }
+
   public boolean sendNotification(TableIdentifier identifier, NotificationRequest request) {
     PolarisAuthorizableOperation op = PolarisAuthorizableOperation.SEND_NOTIFICATIONS;
 

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

@@ -79,6 +79,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.MetricsReporter;
+import org.apache.polaris.service.reporting.MetricsReportingConfiguration;
 import org.apache.polaris.service.secrets.SecretsManagerConfiguration;
 import org.apache.polaris.service.storage.StorageConfiguration;
 import org.apache.polaris.service.storage.aws.S3AccessConfig;
@@ -405,4 +407,11 @@ public PolarisCredentialManager polarisCredentialManager(
   public void closeTaskExecutor(@Disposes @Identifier("task-executor") ManagedExecutor executor) {
     executor.close();
   }
+
+  @Produces
+  @RequestScoped
+  public MetricsReporter metricsReporter(
+      MetricsReportingConfiguration config, @Any Instance<MetricsReporter> 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,19 @@
+package org.apache.polaris.service.reporting;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.RequestScoped;
+import org.apache.iceberg.rest.requests.ReportMetricsRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RequestScoped
+@Identifier("default")
+public class DefaultMetricsReporter implements MetricsReporter {
+  private static final Logger LOGGER = LoggerFactory.getLogger(DefaultMetricsReporter.class);
+
+  @Override
+  public void reportMetric(
+      String prefix, String namespace, String table, ReportMetricsRequest reportMetricsRequest) {
+    // Do Nothing
+  }
+}

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

@@ -0,0 +1,22 @@
+package org.apache.polaris.service.reporting;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.RequestScoped;
+import org.apache.iceberg.rest.requests.ReportMetricsRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RequestScoped
+@Identifier("logging")
+public class LoggingMetricsReporter implements MetricsReporter {
+  private static final Logger LOGGER = LoggerFactory.getLogger(LoggingMetricsReporter.class);
+
+  @Override
+  public void reportMetric(
+      String prefix, String namespace, String table, ReportMetricsRequest reportMetricsRequest) {
+    LOGGER.info(prefix);
+    LOGGER.info(namespace);
+    LOGGER.info(table);
+    LOGGER.info(reportMetricsRequest.toString());
+  }
+}

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

@@ -0,0 +1,8 @@
+package org.apache.polaris.service.reporting;
+
+import org.apache.iceberg.rest.requests.ReportMetricsRequest;
+
+public interface MetricsReporter {
+  public void reportMetric(
+      String prefix, String namespace, String table, ReportMetricsRequest reportMetricsRequest);
+}

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

@@ -0,0 +1,10 @@
+package org.apache.polaris.service.reporting;
+
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+
+@ConfigMapping(prefix = "polaris.metrics.reporting")
+public interface MetricsReportingConfiguration {
+  @WithDefault("default")
+  String type();
+}

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

@@ -0,0 +1,46 @@
+package org.apache.polaris.service.reporting;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.common.collect.ImmutableMap;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import jakarta.inject.Inject;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+@QuarkusTest
+public class MetricsReportingConfigurationTest {
+  @Inject MetricsReportingConfiguration metricsConfig;
+
+  @Test
+  @TestProfile(MetricsReportingConfiguration.DefaultProfile.class)
+  void testDefault() {
+    String type = metricsConfig.type();
+    assertThat(type).isEqualTo("default");
+  }
+
+  @Test
+  @TestProfile(MetricsReportingConfiguration.LoggingProfile.class)
+  void testLogging() {
+    assertThat(metricsConfig.type()).isEqualTo("logging");
+  }
+
+  public static class DefaultProfile implements QuarkusTestProfile {
+    @Override
+    public Map<String, String> getConfigOverrides() {
+      return ImmutableMap.<String, String>builder()
+          .put("polaris.metrics.reporting.type", "default")
+          .build();
+    }
+  }
+
+  public static class LoggingProfile implements QuarkusTestProfile {
+    @Override
+    public Map<String, String> getConfigOverrides() {
+      return ImmutableMap.<String, String>builder()
+          .put("polaris.metrics.reporting.type", "logging")
+          .build();
+    }
+  }
+}