GCP: Add CloudMonitoringMetricsReporter to export Iceberg metrics to Cloud Monitoring

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"

docs/docs/metrics-reporting.md

@@ -120,6 +120,75 @@ This is the default when using the [`RESTCatalog`](https://github.com/apache/ice
 
 Sending metrics via REST can be controlled with the `rest-metrics-reporting-enabled` (defaults to `true`) property.
 
+### `CloudMonitoringMetricsReporter` (GCP)
+
+A `MetricsReporter` implementation that exports `ScanReport` and `CommitReport` as
+[Cloud Monitoring](https://cloud.google.com/monitoring) time series, shipped in the `iceberg-gcp`
+module. Reports are sent asynchronously on a daemon executor; failures are logged and never
+propagate to the caller.
+
+Each report emits one operation series (`scan_operations` or `commit_operations`) plus one series
+per non-null counter/timer in the corresponding `ScanMetricsResult` / `CommitMetricsResult`. All
+series share a fixed, low-cardinality label set:
+
+- `table` — fully qualified table name
+- `catalog` — catalog impl class name (empty when unknown)
+- `metric_name` — stable name of the underlying metric (independent of upstream Java method names)
+- `operation` — commit operation, on commit metrics only
+
+Query predicates, projected column names and field IDs are **not** attached as labels. Each unique
+label combination is a billable Cloud Monitoring time series; the label set is intentionally
+minimal to bound cardinality and cost.
+
+Wire the reporter via the [catalog property](catalog-properties.md) `metrics-reporter-impl`:
+
+```properties
+metrics-reporter-impl=org.apache.iceberg.gcp.metrics.CloudMonitoringMetricsReporter
+gcp.monitoring.project-id=my-gcp-project
+# Optional:
+# gcp.monitoring.metric-prefix=custom.googleapis.com/iceberg
+# gcp.monitoring.endpoint=monitoring.googleapis.com:443
+# gcp.monitoring.quota-project-id=my-billing-project
+# gcp.monitoring.resource-type=global
+```
+
+| Property | Default | Description |
+| --- | --- | --- |
+| `gcp.monitoring.project-id` | falls back to `gcs.project-id` | GCP project that owns the time series |
+| `gcp.monitoring.metric-prefix` | `custom.googleapis.com/iceberg` | Prefix for metric types (`<prefix>/scan_metrics`, `<prefix>/commit_metrics`) |
+| `gcp.monitoring.endpoint` | — | Cloud Monitoring API endpoint override (emulator, private endpoint, VPC-SC) |
+| `gcp.monitoring.quota-project-id` | — | Billing/quota project for the Cloud Monitoring API calls |
+| `gcp.monitoring.resource-type` | `global` | `MonitoredResource` type used for emitted series |
+| `gcp.monitoring.predicate-tracking-enabled` | `false` | Opt-in: also emit per-column predicate and projection time series (see *Predicate tracking* below) |
+| `gcp.monitoring.column-cap` | `32` | Per-scan cap on the union of predicate + projection columns; predicate/projection emission is skipped above this cap |
+
+Property namespaces in `iceberg-gcp`: `gcs.*` covers Google Cloud Storage (FileIO, KMS, OAuth);
+`gcp.*` covers cross-service GCP features such as Cloud Monitoring.
+
+> **Reporter lifecycle.** The reporter owns a `MetricServiceClient` that should be released via
+> `close()`. Catalogs extending `BaseMetastoreCatalog` (Hive, JDBC, Glue, Nessie, ...) call
+> `close()` automatically. **`RESTCatalog` does not** close the reporter today — its `close()` only
+> drains an internal closeables group that the reporter is not registered with. Applications using
+> `RESTCatalog` should close the reporter explicitly (or accept that the gRPC client lives until
+> JVM exit; the channel uses daemon threads, so the JVM still shuts down cleanly).
+
+#### Predicate tracking (opt-in)
+
+Setting `gcp.monitoring.predicate-tracking-enabled=true` adds two metric types:
+
+- `<prefix>/predicate_columns` — labels `{table, catalog, column, op_class}`. One point with value `1`
+  per `(column, op_class)` pair seen in `ScanReport.filter()`. `op_class` buckets Iceberg's
+  `Expression.Operation` into clustering-relevant groups: `point` (`EQ`, `IN`), `range`
+  (`LT`/`LT_EQ`/`GT`/`GT_EQ`), `null` (`IS_NULL`/`NOT_NULL`/`IS_NAN`/`NOT_NAN`), and `other`
+  (everything else, including negations and `STARTS_WITH`).
+- `<prefix>/projection_columns` — labels `{table, catalog, column}`. One point with value `1` per
+  projected column name from `ScanReport.projectedFieldNames()`.
+
+Cardinality budget: the predicate metric type has at most `tables × columns × 4` active series; the
+projection metric type has `tables × columns`. For deployments with many wide tables this can
+exceed the default per-metric-type cap (~6,000 active series); request a quota increase from Google
+or rely on `gcp.monitoring.column-cap` to drop emission for unusually wide scans.
+
 ## Implementing a custom Metrics Reporter
 
 Implementing the [`MetricsReporter`](https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/metrics/MetricsReporter.java) API gives full flexibility in dealing with incoming [`MetricsReport`](https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/metrics/MetricsReport.java) instances. For example, it would be possible to send results to a Prometheus endpoint or any other observability framework/system.

gcp-bundle/build.gradle

@@ -35,6 +35,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,157 @@
+/*
+ * 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 static org.assertj.core.api.Assumptions.assumeThat;
+
+import com.google.cloud.ServiceOptions;
+import com.google.cloud.monitoring.v3.MetricServiceClient;
+import com.google.cloud.monitoring.v3.MetricServiceClient.ListTimeSeriesPagedResponse;
+import com.google.monitoring.v3.ListTimeSeriesRequest;
+import com.google.monitoring.v3.ProjectName;
+import com.google.monitoring.v3.TimeInterval;
+import com.google.monitoring.v3.TimeSeries;
+import com.google.protobuf.util.Timestamps;
+import java.time.Duration;
+import java.util.Map;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Integration test that exercises {@link CloudMonitoringMetricsReporter} end-to-end against a real
+ * Cloud Monitoring project and a configured catalog. Skipped automatically unless the required
+ * properties below are provided.
+ *
+ * <p>Required configuration (system properties or environment variables):
+ *
+ * <ul>
+ *   <li>{@code iceberg.gcp.project} (or default GCP credentials project)
+ *   <li>{@code iceberg.gcp.it.catalog-uri} — REST catalog URI
+ *   <li>{@code iceberg.gcp.it.warehouse} — warehouse location (e.g. {@code gs://...})
+ *   <li>{@code iceberg.gcp.it.namespace} and {@code iceberg.gcp.it.table} — table to scan
+ * </ul>
+ *
+ * Optional: {@code iceberg.gcp.it.catalog-properties} as a comma-separated list of {@code k=v}
+ * pairs for catalog-specific properties (e.g. headers, scan-planning-mode).
+ */
+public class TestCloudMonitoringMetricsReporterIntegration {
+
+  private static final String SYS_PROJECT = "iceberg.gcp.project";
+  private static final String SYS_CATALOG_URI = "iceberg.gcp.it.catalog-uri";
+  private static final String SYS_WAREHOUSE = "iceberg.gcp.it.warehouse";
+  private static final String SYS_NAMESPACE = "iceberg.gcp.it.namespace";
+  private static final String SYS_TABLE = "iceberg.gcp.it.table";
+  private static final String SYS_CATALOG_PROPS = "iceberg.gcp.it.catalog-properties";
+
+  @Test
+  void testMetricsReporting() throws Exception {
+    String projectId = System.getProperty(SYS_PROJECT, ServiceOptions.getDefaultProjectId());
+    String catalogUri = System.getProperty(SYS_CATALOG_URI);
+    String warehouse = System.getProperty(SYS_WAREHOUSE);
+    String namespace = System.getProperty(SYS_NAMESPACE);
+    String tableName = System.getProperty(SYS_TABLE);
+    assumeThat(projectId).as("GCP project ID is not configured").isNotNull();
+    assumeThat(catalogUri).as("%s is not set", SYS_CATALOG_URI).isNotNull();
+    assumeThat(warehouse).as("%s is not set", SYS_WAREHOUSE).isNotNull();
+    assumeThat(namespace).as("%s is not set", SYS_NAMESPACE).isNotNull();
+    assumeThat(tableName).as("%s is not set", SYS_TABLE).isNotNull();
+
+    Map<String, String> properties = Maps.newHashMap();
+    properties.put(CatalogProperties.URI, catalogUri);
+    properties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
+    properties.put(CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO");
+    properties.put(
+        CatalogProperties.METRICS_REPORTER_IMPL, CloudMonitoringMetricsReporter.class.getName());
+    properties.put(GCPProperties.GCP_MONITORING_PROJECT_ID, projectId);
+    properties.putAll(parseExtraProperties(System.getProperty(SYS_CATALOG_PROPS, "")));
+
+    long startMillis = System.currentTimeMillis();
+    String catalogId = "iceberg_cloud_monitoring_it";
+
+    RESTCatalog catalog = new RESTCatalog();
+    catalog.initialize(catalogId, properties);
+
+    Table table = catalog.loadTable(TableIdentifier.of(namespace, tableName));
+    try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
+      tasks.forEach(t -> {});
+    }
+
+    String prefix =
+        properties.getOrDefault(
+            GCPProperties.GCP_MONITORING_METRIC_PREFIX,
+            GCPProperties.GCP_MONITORING_METRIC_PREFIX_DEFAULT);
+    String filter =
+        String.format(
+            "metric.type=\"%s/scan_metrics\" AND metric.label.\"metric_name\"=\"scan_operations\"",
+            prefix);
+
+    try (MetricServiceClient client = MetricServiceClient.create()) {
+      Awaitility.await("scan_operations metric appears in Cloud Monitoring")
+          .atMost(Duration.ofMinutes(3))
+          .pollInterval(Duration.ofSeconds(5))
+          .until(() -> findScanOperations(client, projectId, filter, startMillis));
+    }
+  }
+
+  private static boolean findScanOperations(
+      MetricServiceClient client, String projectId, String filter, long startMillis) {
+    TimeInterval interval =
+        TimeInterval.newBuilder()
+            .setStartTime(Timestamps.fromMillis(startMillis))
+            .setEndTime(Timestamps.fromMillis(System.currentTimeMillis()))
+            .build();
+    ListTimeSeriesRequest request =
+        ListTimeSeriesRequest.newBuilder()
+            .setName(ProjectName.of(projectId).toString())
+            .setFilter(filter)
+            .setInterval(interval)
+            .build();
+    ListTimeSeriesPagedResponse response = client.listTimeSeries(request);
+    for (TimeSeries ts : response.iterateAll()) {
+      assertThat(ts.getMetric().getLabelsMap()).containsKey("table");
+      assertThat(ts.getMetric().getLabelsMap()).containsKey("catalog");
+      assertThat(ts.getMetric().getLabelsMap()).doesNotContainKeys("filter", "queried_field_ids");
+      return true;
+    }
+    return false;
+  }
+
+  private static Map<String, String> parseExtraProperties(String raw) {
+    Map<String, String> out = Maps.newHashMap();
+    if (raw == null || raw.isBlank()) {
+      return out;
+    }
+    for (String pair : raw.split(",")) {
+      int eq = pair.indexOf('=');
+      if (eq > 0) {
+        out.put(pair.substring(0, eq).trim(), pair.substring(eq + 1).trim());
+      }
+    }
+    return out;
+  }
+}