Add CloudMonitoringMetricsReporter to export Iceberg metrics to Google Cloud Monitoring

Author: talatuyarer

build.gradle

@@ -758,6 +758,7 @@ project(':iceberg-gcp') {
     compileOnly platform(libs.google.libraries.bom)
     compileOnly "com.google.cloud:google-cloud-storage"
     compileOnly "com.google.cloud:google-cloud-kms"
+    compileOnly "com.google.cloud:google-cloud-monitoring"
     compileOnly(libs.gcs.analytics.core)
 
     testImplementation "com.google.cloud:google-cloud-nio"

gcp-bundle/build.gradle

@@ -29,6 +29,7 @@ project(":iceberg-gcp-bundle") {
     implementation "com.google.cloud:google-cloud-bigquery"
     implementation "com.google.cloud:google-cloud-core"
     implementation "com.google.cloud:google-cloud-kms"
+    implementation "com.google.cloud:google-cloud-monitoring"
     implementation libs.gcs.analytics.core
   }
 

gcp/src/integration/java/org/apache/iceberg/gcp/metrics/TestCloudMonitoringMetricsReporterIntegration.java

@@ -0,0 +1,109 @@
+/*
+ * 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.iceberg.gcp.metrics;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.cloud.ServiceOptions;
+import com.google.cloud.monitoring.v3.MetricServiceClient;
+import com.google.monitoring.v3.ListTimeSeriesRequest;
+import com.google.monitoring.v3.ProjectName;
+import com.google.monitoring.v3.TimeInterval;
+import com.google.protobuf.util.Timestamps;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.Test;
+
+public class TestCloudMonitoringMetricsReporterIntegration {
+
+  @Test
+  void testMetricsReporting() throws Exception {
+    String projectId =
+        System.getProperty("iceberg.gcp.project", ServiceOptions.getDefaultProjectId());
+    Assumptions.assumeTrue(projectId != null, "GCP project ID is not set in environment");
+
+    Map<String, String> properties = new HashMap<>();
+    properties.put(CatalogProperties.URI, "https://biglake.googleapis.com/iceberg/v1/restcatalog");
+    properties.put(CatalogProperties.WAREHOUSE_LOCATION, "gs://biglake-public-nyc-taxi-iceberg");
+    properties.put(CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO");
+    properties.put("header.x-goog-user-project", projectId);
+    properties.put("header.X-Iceberg-Access-Delegation", "vended-credentials");
+    properties.put("rest.auth.type", "org.apache.iceberg.gcp.auth.GoogleAuthManager");
+    properties.put("scan-planning-mode", "client");
+
+    properties.put(
+        CatalogProperties.METRICS_REPORTER_IMPL,
+        "org.apache.iceberg.gcp.metrics.CloudMonitoringMetricsReporter");
+    properties.put("gcp.monitoring.project-id", projectId);
+
+    RESTCatalog catalog = new RESTCatalog();
+    catalog.initialize("my_catalog", properties);
+
+    TableIdentifier tableIdent = TableIdentifier.of("public_data", "nyc_taxicab");
+    org.apache.iceberg.Table table = catalog.loadTable(tableIdent);
+
+    long startTime = System.currentTimeMillis();
+
+    // Plan scan to trigger metrics
+    try (org.apache.iceberg.io.CloseableIterable<org.apache.iceberg.FileScanTask> tasks =
+        table.newScan().planFiles()) {
+      tasks.forEach(task -> {});
+    }
+
+    // Wait for async reporting
+    Thread.sleep(10000);
+
+    // Verify metrics in Cloud Monitoring
+    try (MetricServiceClient client = MetricServiceClient.create()) {
+      TimeInterval interval =
+          TimeInterval.newBuilder()
+              .setStartTime(Timestamps.fromMillis(startTime))
+              .setEndTime(Timestamps.fromMillis(System.currentTimeMillis()))
+              .build();
+
+      ListTimeSeriesRequest request =
+          ListTimeSeriesRequest.newBuilder()
+              .setName(ProjectName.of(projectId).toString())
+              .setFilter("metric.type=\"custom.googleapis.com/iceberg/scan_metrics\"")
+              .setInterval(interval)
+              .build();
+
+      MetricServiceClient.ListTimeSeriesPagedResponse response = client.listTimeSeries(request);
+
+      java.util.Iterator<com.google.monitoring.v3.TimeSeries> iterator =
+          response.iterateAll().iterator();
+
+      boolean found = false;
+      while (iterator.hasNext()) {
+        com.google.monitoring.v3.TimeSeries ts = iterator.next();
+        if ("scan_operations".equals(ts.getMetric().getLabelsOrDefault("metric_name", ""))) {
+          assertThat(ts.getMetric().getLabelsOrThrow("table"))
+              .isEqualTo("my_catalog.public_data.nyc_taxicab");
+          found = true;
+          break;
+        }
+      }
+      assertThat(found).isTrue();
+    }
+  }
+}

gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java

@@ -73,6 +73,18 @@ public class GCPProperties implements Serializable {
   /** Controls whether analytics core library is enabled or not. Defaults to false. */
   public static final String GCS_ANALYTICS_CORE_ENABLED = "gcs.analytics-core.enabled";
 
+  /**
+   * Controls the Google Cloud project ID used for Google Cloud Monitoring metrics. Defaults to the
+   * GCS project ID.
+   */
+  public static final String GCP_MONITORING_PROJECT_ID = "gcp.monitoring.project-id";
+
+  /** Controls the metric prefix used for Google Cloud Monitoring metrics. */
+  public static final String GCP_MONITORING_METRIC_PREFIX = "gcp.monitoring.metric-prefix";
+
+  /** Default metric prefix for Google Cloud Monitoring metrics. */
+  public static final String GCP_MONITORING_METRIC_PREFIX_DEFAULT = "custom.googleapis.com/iceberg";
+
   /**
    * Max possible batch size for deletion. Currently, a max of 100 keys is advised, so we default to
    * a number below that. https://cloud.google.com/storage/docs/batch
@@ -99,6 +111,9 @@ public class GCPProperties implements Serializable {
   private boolean gcsOauth2RefreshCredentialsEnabled;
   private boolean gcsAnalyticsCoreEnabled;
 
+  private String gcpMonitoringProjectId;
+  private String gcpMonitoringMetricPrefix;
+
   private String gcsImpersonateServiceAccount;
   private int gcsImpersonateLifetimeSeconds;
   private List<String> gcsImpersonateDelegates;
@@ -197,6 +212,10 @@ public GCPProperties(Map<String, String> properties) {
 
     gcsAnalyticsCoreEnabled =
         PropertyUtil.propertyAsBoolean(properties, GCS_ANALYTICS_CORE_ENABLED, false);
+
+    gcpMonitoringProjectId = properties.getOrDefault(GCP_MONITORING_PROJECT_ID, projectId);
+    gcpMonitoringMetricPrefix =
+        properties.getOrDefault(GCP_MONITORING_METRIC_PREFIX, GCP_MONITORING_METRIC_PREFIX_DEFAULT);
   }
 
   public Optional<Integer> channelReadChunkSize() {
@@ -278,4 +297,12 @@ public Map<String, String> properties() {
   public boolean isGcsAnalyticsCoreEnabled() {
     return gcsAnalyticsCoreEnabled;
   }
+
+  public String gcpMonitoringProjectId() {
+    return gcpMonitoringProjectId;
+  }
+
+  public String gcpMonitoringMetricPrefix() {
+    return gcpMonitoringMetricPrefix;
+  }
 }