From 671db0b4b6ea61948be1d6b960a5eb30b6b4b91b Mon Sep 17 00:00:00 2001 From: Talat Uyarer Date: Wed, 13 May 2026 16:51:18 -0700 Subject: [PATCH] Initial version of Cloud Monitoring --- build.gradle | 1 + docs/docs/metrics-reporting.md | 69 ++ gcp-bundle/build.gradle | 1 + ...dMonitoringMetricsReporterIntegration.java | 157 +++++ .../org/apache/iceberg/gcp/GCPProperties.java | 118 ++++ .../CloudMonitoringMetricsReporter.java | 601 ++++++++++++++++++ .../TestCloudMonitoringMetricsReporter.java | 562 ++++++++++++++++ 7 files changed, 1509 insertions(+) create mode 100644 gcp/src/integration/java/org/apache/iceberg/gcp/metrics/TestCloudMonitoringMetricsReporterIntegration.java create mode 100644 gcp/src/main/java/org/apache/iceberg/gcp/metrics/CloudMonitoringMetricsReporter.java create mode 100644 gcp/src/test/java/org/apache/iceberg/gcp/metrics/TestCloudMonitoringMetricsReporter.java diff --git a/build.gradle b/build.gradle index 261dfabf0412..4b1b24aca385 100644 --- a/build.gradle +++ b/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" diff --git a/docs/docs/metrics-reporting.md b/docs/docs/metrics-reporting.md index 4ca452b0d503..573b64ac4ef4 100644 --- a/docs/docs/metrics-reporting.md +++ b/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 (`/scan_metrics`, `/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: + +- `/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`). +- `/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. diff --git a/gcp-bundle/build.gradle b/gcp-bundle/build.gradle index d48763b5e366..c94949f26ebf 100644 --- a/gcp-bundle/build.gradle +++ b/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 } diff --git a/gcp/src/integration/java/org/apache/iceberg/gcp/metrics/TestCloudMonitoringMetricsReporterIntegration.java b/gcp/src/integration/java/org/apache/iceberg/gcp/metrics/TestCloudMonitoringMetricsReporterIntegration.java new file mode 100644 index 000000000000..4ffebbac66bd --- /dev/null +++ b/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. + * + *

Required configuration (system properties or environment variables): + * + *

    + *
  • {@code iceberg.gcp.project} (or default GCP credentials project) + *
  • {@code iceberg.gcp.it.catalog-uri} — REST catalog URI + *
  • {@code iceberg.gcp.it.warehouse} — warehouse location (e.g. {@code gs://...}) + *
  • {@code iceberg.gcp.it.namespace} and {@code iceberg.gcp.it.table} — table to scan + *
+ * + * 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 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 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 parseExtraProperties(String raw) { + Map 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; + } +} diff --git a/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java b/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java index 2702ce565d4e..98f803d8e40f 100644 --- a/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java +++ b/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java @@ -33,6 +33,17 @@ import org.apache.iceberg.rest.RESTUtil; import org.apache.iceberg.util.PropertyUtil; +/** + * Configuration properties for the {@code iceberg-gcp} module. + * + *

Property namespaces: + * + *

    + *
  • {@code gcs.*} — Google Cloud Storage (FileIO, KMS, OAuth/impersonation for GCS). + *
  • {@code gcp.*} — Cross-service GCP features that are not specific to GCS (e.g. Cloud + * Monitoring metric reporting). + *
+ */ public class GCPProperties implements Serializable { // Service Options public static final String GCS_PROJECT_ID = "gcs.project-id"; @@ -73,6 +84,63 @@ 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"; + /** + * Google Cloud project that owns the time series written by the Cloud Monitoring metrics + * reporter. Falls back to {@link #GCS_PROJECT_ID} when not set. + */ + public static final String GCP_MONITORING_PROJECT_ID = "gcp.monitoring.project-id"; + + /** + * Metric type prefix for series written by the Cloud Monitoring metrics reporter. Defaults to + * {@link #GCP_MONITORING_METRIC_PREFIX_DEFAULT}. + */ + 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"; + + /** + * Optional override for the Cloud Monitoring API endpoint (host:port). Intended for emulators, + * private endpoints and VPC-SC perimeters. When unset the default Google-managed endpoint is + * used. + */ + public static final String GCP_MONITORING_ENDPOINT = "gcp.monitoring.endpoint"; + + /** + * Optional billing/quota project used for Cloud Monitoring API calls. When unset the project + * associated with the active credentials is billed. + */ + public static final String GCP_MONITORING_QUOTA_PROJECT_ID = "gcp.monitoring.quota-project-id"; + + /** + * Optional {@code MonitoredResource} type used when writing time series (e.g. {@code + * k8s_container}, {@code gce_instance}). Defaults to {@code global} when unset. + */ + public static final String GCP_MONITORING_RESOURCE_TYPE = "gcp.monitoring.resource-type"; + + /** Default {@code MonitoredResource} type used by the Cloud Monitoring metrics reporter. */ + public static final String GCP_MONITORING_RESOURCE_TYPE_DEFAULT = "global"; + + /** + * Opt-in: when {@code true}, the Cloud Monitoring reporter additionally emits per-column + * predicate and projection time series ({@code /predicate_columns} and {@code + * /projection_columns}). These series enable workload-driven analyses such as + * clustering-key recommendation, at the cost of higher time-series cardinality. + */ + public static final String GCP_MONITORING_PREDICATE_TRACKING_ENABLED = + "gcp.monitoring.predicate-tracking-enabled"; + + /** + * Per-scan cap on the number of columns reported via the predicate and projection metric types. + * When the union of predicate columns and projected columns for a scan exceeds this cap, + * predicate and projection emission for that scan is skipped to bound time-series cardinality. + * The base scan/commit metrics are unaffected. + */ + public static final String GCP_MONITORING_COLUMN_CAP = "gcp.monitoring.column-cap"; + + /** Default per-scan column cap for predicate and projection emission. */ + public static final int GCP_MONITORING_COLUMN_CAP_DEFAULT = 32; + /** * 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 +167,14 @@ public class GCPProperties implements Serializable { private boolean gcsOauth2RefreshCredentialsEnabled; private boolean gcsAnalyticsCoreEnabled; + private String gcpMonitoringProjectId; + private String gcpMonitoringMetricPrefix; + private String gcpMonitoringEndpoint; + private String gcpMonitoringQuotaProjectId; + private String gcpMonitoringResourceType; + private boolean gcpMonitoringPredicateTrackingEnabled; + private int gcpMonitoringColumnCap = GCP_MONITORING_COLUMN_CAP_DEFAULT; + private String gcsImpersonateServiceAccount; private int gcsImpersonateLifetimeSeconds; private List gcsImpersonateDelegates; @@ -197,6 +273,20 @@ public GCPProperties(Map 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); + gcpMonitoringEndpoint = properties.get(GCP_MONITORING_ENDPOINT); + gcpMonitoringQuotaProjectId = properties.get(GCP_MONITORING_QUOTA_PROJECT_ID); + gcpMonitoringResourceType = + properties.getOrDefault(GCP_MONITORING_RESOURCE_TYPE, GCP_MONITORING_RESOURCE_TYPE_DEFAULT); + gcpMonitoringPredicateTrackingEnabled = + PropertyUtil.propertyAsBoolean( + properties, GCP_MONITORING_PREDICATE_TRACKING_ENABLED, false); + gcpMonitoringColumnCap = + PropertyUtil.propertyAsInt( + properties, GCP_MONITORING_COLUMN_CAP, GCP_MONITORING_COLUMN_CAP_DEFAULT); } public Optional channelReadChunkSize() { @@ -278,4 +368,32 @@ public Map properties() { public boolean isGcsAnalyticsCoreEnabled() { return gcsAnalyticsCoreEnabled; } + + public Optional gcpMonitoringProjectId() { + return Optional.ofNullable(gcpMonitoringProjectId); + } + + public String gcpMonitoringMetricPrefix() { + return gcpMonitoringMetricPrefix; + } + + public Optional gcpMonitoringEndpoint() { + return Optional.ofNullable(gcpMonitoringEndpoint); + } + + public Optional gcpMonitoringQuotaProjectId() { + return Optional.ofNullable(gcpMonitoringQuotaProjectId); + } + + public String gcpMonitoringResourceType() { + return gcpMonitoringResourceType; + } + + public boolean gcpMonitoringPredicateTrackingEnabled() { + return gcpMonitoringPredicateTrackingEnabled; + } + + public int gcpMonitoringColumnCap() { + return gcpMonitoringColumnCap; + } } diff --git a/gcp/src/main/java/org/apache/iceberg/gcp/metrics/CloudMonitoringMetricsReporter.java b/gcp/src/main/java/org/apache/iceberg/gcp/metrics/CloudMonitoringMetricsReporter.java new file mode 100644 index 000000000000..a083c9f503e7 --- /dev/null +++ b/gcp/src/main/java/org/apache/iceberg/gcp/metrics/CloudMonitoringMetricsReporter.java @@ -0,0 +1,601 @@ +/* + * 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 com.google.api.Metric; +import com.google.api.MonitoredResource; +import com.google.api.gax.core.NoCredentialsProvider; +import com.google.cloud.monitoring.v3.MetricServiceClient; +import com.google.cloud.monitoring.v3.MetricServiceSettings; +import com.google.monitoring.v3.CreateTimeSeriesRequest; +import com.google.monitoring.v3.Point; +import com.google.monitoring.v3.ProjectName; +import com.google.monitoring.v3.TimeInterval; +import com.google.monitoring.v3.TimeSeries; +import com.google.monitoring.v3.TypedValue; +import com.google.protobuf.util.Timestamps; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.function.Function; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.expressions.BoundPredicate; +import org.apache.iceberg.expressions.BoundReference; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.ExpressionUtil; +import org.apache.iceberg.expressions.ExpressionVisitors; +import org.apache.iceberg.expressions.UnboundPredicate; +import org.apache.iceberg.gcp.GCPProperties; +import org.apache.iceberg.metrics.CommitMetricsResult; +import org.apache.iceberg.metrics.CommitReport; +import org.apache.iceberg.metrics.CounterResult; +import org.apache.iceberg.metrics.MetricsReport; +import org.apache.iceberg.metrics.MetricsReporter; +import org.apache.iceberg.metrics.ScanMetricsResult; +import org.apache.iceberg.metrics.ScanReport; +import org.apache.iceberg.metrics.TimerResult; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.util.Tasks; +import org.apache.iceberg.util.ThreadPools; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link MetricsReporter} that exports {@link ScanReport} and {@link CommitReport} as time series + * to Google Cloud Monitoring. + * + *

Each report becomes a small batch of {@link TimeSeries} points: one "operation" series ({@code + * scan_operations} / {@code commit_operations}) plus one series per non-null counter or timer in + * the corresponding {@link ScanMetricsResult} / {@link CommitMetricsResult}. + * + *

Label policy

+ * + * Labels are intentionally fixed to bound time-series cardinality: + * + *
    + *
  • {@code table} — fully qualified table name from the report. + *
  • {@code catalog} — catalog impl name when available, otherwise omitted. + *
  • {@code metric_name} — the stable name of the underlying counter/timer (see {@link + * #SCAN_METRICS} / {@link #COMMIT_METRICS}). These names are independent of the Java method + * names on {@link ScanMetricsResult} / {@link CommitMetricsResult} so they survive upstream + * renames. + *
  • {@code operation} — commit operation (only on commit metrics). + *
+ * + * Query predicates, projected column names and projected field ids are not attached as + * labels. Each unique label combination is a billable time series in Cloud Monitoring; we keep the + * key set small by design. + * + *

Metric kinds

+ * + * All series are reported as {@code GAUGE} of {@code INT64}. Iceberg counters carry per-report + * totals (not running totals across reports), so each point is a point-in-time observation. + * Downstream aggregations (e.g. {@code sum}, {@code count}, {@code rate}) operate over time + * windows. + * + *

Thread model

+ * + * {@link #report(MetricsReport)} hands work off to a shared, JVM-managed daemon executor (see + * {@link ThreadPools#newExitingWorkerPool(String, int)}) and returns immediately. Failures are + * logged via {@link Tasks.Builder#suppressFailureWhenFinished()} and never propagate to the caller. + * + *

Lifecycle

+ * + * This reporter owns the underlying {@link MetricServiceClient}. {@link #close()} is invoked + * automatically by the catalog. + */ +public class CloudMonitoringMetricsReporter implements MetricsReporter { + private static final Logger LOG = LoggerFactory.getLogger(CloudMonitoringMetricsReporter.class); + + /** Cloud Monitoring limit on time series per CreateTimeSeries request. */ + private static final int MAX_TIME_SERIES_PER_REQUEST = 200; + + /** Label keys emitted on every scan time series. */ + @VisibleForTesting + static final Set SCAN_LABEL_KEYS = ImmutableSet.of("table", "catalog", "metric_name"); + + /** Label keys emitted on every commit time series. */ + @VisibleForTesting + static final Set COMMIT_LABEL_KEYS = + ImmutableSet.of("table", "catalog", "metric_name", "operation"); + + /** + * Label keys emitted on the predicate-column metric type. Only emitted when {@link + * GCPProperties#GCP_MONITORING_PREDICATE_TRACKING_ENABLED} is true. + */ + @VisibleForTesting + static final Set PREDICATE_LABEL_KEYS = + ImmutableSet.of("table", "catalog", "column", "op_class"); + + /** + * Label keys emitted on the projection-column metric type. Only emitted when {@link + * GCPProperties#GCP_MONITORING_PREDICATE_TRACKING_ENABLED} is true. + */ + @VisibleForTesting + static final Set PROJECTION_LABEL_KEYS = ImmutableSet.of("table", "catalog", "column"); + + /** + * Names are decoupled from upstream Java method names so a rename in {@code ScanMetricsResult} + * cannot silently change exported metric names. + */ + @VisibleForTesting + static final Map> SCAN_METRICS = + ImmutableMap.>builder() + .put("total_planning_duration_ms", ScanMetricsResult::totalPlanningDuration) + .put("result_data_files", ScanMetricsResult::resultDataFiles) + .put("result_delete_files", ScanMetricsResult::resultDeleteFiles) + .put("total_data_manifests", ScanMetricsResult::totalDataManifests) + .put("total_delete_manifests", ScanMetricsResult::totalDeleteManifests) + .put("scanned_data_manifests", ScanMetricsResult::scannedDataManifests) + .put("skipped_data_manifests", ScanMetricsResult::skippedDataManifests) + .put("total_file_size_bytes", ScanMetricsResult::totalFileSizeInBytes) + .put("total_delete_file_size_bytes", ScanMetricsResult::totalDeleteFileSizeInBytes) + .put("skipped_data_files", ScanMetricsResult::skippedDataFiles) + .put("skipped_delete_files", ScanMetricsResult::skippedDeleteFiles) + .put("scanned_delete_manifests", ScanMetricsResult::scannedDeleteManifests) + .put("skipped_delete_manifests", ScanMetricsResult::skippedDeleteManifests) + .put("indexed_delete_files", ScanMetricsResult::indexedDeleteFiles) + .put("equality_delete_files", ScanMetricsResult::equalityDeleteFiles) + .put("positional_delete_files", ScanMetricsResult::positionalDeleteFiles) + .put("dvs", ScanMetricsResult::dvs) + .build(); + + @VisibleForTesting + static final Map> COMMIT_METRICS = + ImmutableMap.>builder() + .put("total_duration_ms", CommitMetricsResult::totalDuration) + .put("attempts", CommitMetricsResult::attempts) + .put("added_data_files", CommitMetricsResult::addedDataFiles) + .put("removed_data_files", CommitMetricsResult::removedDataFiles) + .put("total_data_files", CommitMetricsResult::totalDataFiles) + .put("added_delete_files", CommitMetricsResult::addedDeleteFiles) + .put("added_equality_delete_files", CommitMetricsResult::addedEqualityDeleteFiles) + .put("added_positional_delete_files", CommitMetricsResult::addedPositionalDeleteFiles) + .put("added_dvs", CommitMetricsResult::addedDVs) + .put("removed_delete_files", CommitMetricsResult::removedDeleteFiles) + .put("removed_equality_delete_files", CommitMetricsResult::removedEqualityDeleteFiles) + .put("removed_positional_delete_files", CommitMetricsResult::removedPositionalDeleteFiles) + .put("removed_dvs", CommitMetricsResult::removedDVs) + .put("total_delete_files", CommitMetricsResult::totalDeleteFiles) + .put("added_records", CommitMetricsResult::addedRecords) + .put("removed_records", CommitMetricsResult::removedRecords) + .put("total_records", CommitMetricsResult::totalRecords) + .put("added_files_size_bytes", CommitMetricsResult::addedFilesSizeInBytes) + .put("removed_files_size_bytes", CommitMetricsResult::removedFilesSizeInBytes) + .put("total_files_size_bytes", CommitMetricsResult::totalFilesSizeInBytes) + .put("added_positional_deletes", CommitMetricsResult::addedPositionalDeletes) + .put("removed_positional_deletes", CommitMetricsResult::removedPositionalDeletes) + .put("total_positional_deletes", CommitMetricsResult::totalPositionalDeletes) + .put("added_equality_deletes", CommitMetricsResult::addedEqualityDeletes) + .put("removed_equality_deletes", CommitMetricsResult::removedEqualityDeletes) + .put("total_equality_deletes", CommitMetricsResult::totalEqualityDeletes) + .put("manifests_created", CommitMetricsResult::manifestsCreated) + .put("manifests_replaced", CommitMetricsResult::manifestsReplaced) + .put("manifests_kept", CommitMetricsResult::manifestsKept) + .put("manifest_entries_processed", CommitMetricsResult::manifestEntriesProcessed) + .build(); + + private static final ExecutorService DEFAULT_EXECUTOR = + ThreadPools.newExitingWorkerPool("gcp-cloud-monitoring-metrics-reporter", 1); + + private final ExecutorService executor; + + private MetricServiceClient client; + private ProjectName projectName; + private MonitoredResource resource; + private String catalogName; + private String scanMetricType; + private String commitMetricType; + private String predicateMetricType; + private String projectionMetricType; + private boolean predicateTrackingEnabled; + private int columnCap; + private volatile boolean closed = false; + + /** Required by {@link org.apache.iceberg.CatalogUtil#loadMetricsReporter}. */ + public CloudMonitoringMetricsReporter() { + this.executor = DEFAULT_EXECUTOR; + } + + @VisibleForTesting + CloudMonitoringMetricsReporter( + GCPProperties properties, MetricServiceClient client, ExecutorService executor) { + this.executor = executor; + bind(properties, client); + } + + @Override + public void initialize(Map properties) { + GCPProperties props = new GCPProperties(properties); + this.catalogName = properties.get(CatalogProperties.CATALOG_IMPL); + bind(props, createClient(props)); + } + + private void bind(GCPProperties properties, MetricServiceClient metricServiceClient) { + Preconditions.checkArgument( + properties.gcpMonitoringProjectId().isPresent(), + "Cannot initialize CloudMonitoringMetricsReporter: %s (or %s as fallback) must be set", + GCPProperties.GCP_MONITORING_PROJECT_ID, + GCPProperties.GCS_PROJECT_ID); + this.client = metricServiceClient; + this.projectName = ProjectName.of(properties.gcpMonitoringProjectId().get()); + this.resource = + MonitoredResource.newBuilder().setType(properties.gcpMonitoringResourceType()).build(); + String prefix = properties.gcpMonitoringMetricPrefix(); + this.scanMetricType = prefix + "/scan_metrics"; + this.commitMetricType = prefix + "/commit_metrics"; + this.predicateMetricType = prefix + "/predicate_columns"; + this.projectionMetricType = prefix + "/projection_columns"; + this.predicateTrackingEnabled = properties.gcpMonitoringPredicateTrackingEnabled(); + this.columnCap = properties.gcpMonitoringColumnCap(); + } + + private static MetricServiceClient createClient(GCPProperties props) { + try { + MetricServiceSettings.Builder builder = MetricServiceSettings.newBuilder(); + if (props.gcpMonitoringEndpoint().isPresent()) { + builder.setEndpoint(props.gcpMonitoringEndpoint().get()); + } + if (props.gcpMonitoringQuotaProjectId().isPresent()) { + builder.setQuotaProjectId(props.gcpMonitoringQuotaProjectId().get()); + } + if (props.noAuth()) { + builder.setCredentialsProvider(NoCredentialsProvider.create()); + } + return MetricServiceClient.create(builder.build()); + } catch (IOException e) { + throw new UncheckedIOException("Failed to create Cloud Monitoring MetricServiceClient", e); + } + } + + @Override + public void report(MetricsReport report) { + if (closed) { + LOG.warn("CloudMonitoringMetricsReporter is closed; dropping report"); + return; + } + if (report == null) { + LOG.warn("Received invalid metrics report: null"); + return; + } + + Tasks.range(1) + .executeWith(executor) + .suppressFailureWhenFinished() + .onFailure( + (item, exception) -> + LOG.warn("Failed to report metrics to Cloud Monitoring", exception)) + .run( + item -> { + if (report instanceof ScanReport scanReport) { + sendTimeSeries(buildScanSeries(scanReport)); + } else if (report instanceof CommitReport commitReport) { + sendTimeSeries(buildCommitSeries(commitReport)); + } else { + LOG.debug("Ignoring unsupported metrics report type: {}", report.getClass()); + } + }); + } + + @VisibleForTesting + List buildScanSeries(ScanReport report) { + List series = Lists.newArrayList(); + TimeInterval interval = currentInterval(); + + series.add( + timeSeries( + scanMetricType, scanLabels(report.tableName(), "scan_operations"), interval, 1L)); + + ScanMetricsResult metrics = report.scanMetrics(); + if (metrics != null) { + for (Map.Entry> entry : SCAN_METRICS.entrySet()) { + Long value = toLong(entry.getValue().apply(metrics)); + if (value != null) { + series.add( + timeSeries( + scanMetricType, scanLabels(report.tableName(), entry.getKey()), interval, value)); + } + } + } + + if (predicateTrackingEnabled) { + addPredicateAndProjectionSeries(report, interval, series); + } + return series; + } + + private void addPredicateAndProjectionSeries( + ScanReport report, TimeInterval interval, List series) { + Set predicates = + PredicateColumnExtractor.extract(report.filter()); + List projection = + report.projectedFieldNames() == null ? List.of() : report.projectedFieldNames(); + + Set uniqueColumns = Sets.newHashSet(); + predicates.forEach(e -> uniqueColumns.add(e.column())); + uniqueColumns.addAll(projection); + + if (uniqueColumns.size() > columnCap) { + LOG.debug( + "Skipping predicate/projection emission for table {}: {} unique columns exceeds cap {}", + report.tableName(), + uniqueColumns.size(), + columnCap); + return; + } + + for (PredicateColumnExtractor.Entry entry : predicates) { + series.add( + timeSeries( + predicateMetricType, + predicateLabels(report.tableName(), entry.column(), entry.opClass()), + interval, + 1L)); + } + for (String column : projection) { + series.add( + timeSeries( + projectionMetricType, projectionLabels(report.tableName(), column), interval, 1L)); + } + } + + @VisibleForTesting + List buildCommitSeries(CommitReport report) { + List series = Lists.newArrayList(); + TimeInterval interval = currentInterval(); + + series.add( + timeSeries( + commitMetricType, + commitLabels(report.tableName(), report.operation(), "commit_operations"), + interval, + 1L)); + + CommitMetricsResult metrics = report.commitMetrics(); + if (metrics != null) { + for (Map.Entry> entry : COMMIT_METRICS.entrySet()) { + Long value = toLong(entry.getValue().apply(metrics)); + if (value != null) { + series.add( + timeSeries( + commitMetricType, + commitLabels(report.tableName(), report.operation(), entry.getKey()), + interval, + value)); + } + } + } + return series; + } + + private Map scanLabels(String tableName, String metricName) { + return ImmutableMap.of( + "table", nullSafe(tableName), "catalog", nullSafe(catalogName), "metric_name", metricName); + } + + private Map predicateLabels(String tableName, String column, String opClass) { + return ImmutableMap.of( + "table", + nullSafe(tableName), + "catalog", + nullSafe(catalogName), + "column", + nullSafe(column), + "op_class", + opClass); + } + + private Map projectionLabels(String tableName, String column) { + return ImmutableMap.of( + "table", nullSafe(tableName), "catalog", nullSafe(catalogName), "column", nullSafe(column)); + } + + private Map commitLabels(String tableName, String operation, String metricName) { + return ImmutableMap.of( + "table", + nullSafe(tableName), + "catalog", + nullSafe(catalogName), + "metric_name", + metricName, + "operation", + nullSafe(operation)); + } + + private TimeSeries timeSeries( + String metricType, Map labels, TimeInterval interval, long value) { + Point point = + Point.newBuilder() + .setInterval(interval) + .setValue(TypedValue.newBuilder().setInt64Value(value).build()) + .build(); + return TimeSeries.newBuilder() + .setMetric(Metric.newBuilder().setType(metricType).putAllLabels(labels).build()) + .setResource(resource) + .addPoints(point) + .build(); + } + + private static TimeInterval currentInterval() { + return TimeInterval.newBuilder() + .setEndTime(Timestamps.fromMillis(System.currentTimeMillis())) + .build(); + } + + private static Long toLong(Object metricResult) { + if (metricResult instanceof CounterResult counter) { + return counter.value(); + } + if (metricResult instanceof TimerResult timer) { + return timer.totalDuration().toMillis(); + } + return null; + } + + private static String nullSafe(String value) { + return value == null ? "" : value; + } + + private void sendTimeSeries(List series) { + if (series.isEmpty()) { + return; + } + for (int start = 0; start < series.size(); start += MAX_TIME_SERIES_PER_REQUEST) { + List chunk = + series.subList(start, Math.min(start + MAX_TIME_SERIES_PER_REQUEST, series.size())); + CreateTimeSeriesRequest request = + CreateTimeSeriesRequest.newBuilder() + .setName(projectName.toString()) + .addAllTimeSeries(chunk) + .build(); + client.createTimeSeries(request); + } + } + + @Override + public void close() { + if (closed) { + return; + } + closed = true; + if (client != null) { + try { + client.close(); + } catch (Exception e) { + LOG.warn("Failed to close Cloud Monitoring MetricServiceClient", e); + } + } + } + + /** + * Walks an {@link Expression} and collects {@code (column, op_class)} pairs from leaf predicates. + * Op classes bucket Iceberg's {@link Expression.Operation} into clustering-relevant groups: + * {@code point} (eq, in), {@code range} (lt/lte/gt/gte), {@code null} (is_null/not_null and the + * nan variants), and {@code other} (everything else, including negations and starts_with). + * + *

Aggregate operations and structural nodes (and/or/not/true/false) are not emitted; the + * structure is irrelevant for column-frequency analytics. + */ + @VisibleForTesting + static final class PredicateColumnExtractor extends ExpressionVisitors.ExpressionVisitor { + + /** A single column reference with its bucketed operator class. */ + static final class Entry { + private final String column; + private final String opClass; + + Entry(String column, String opClass) { + this.column = column; + this.opClass = opClass; + } + + String column() { + return column; + } + + String opClass() { + return opClass; + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if (!(other instanceof Entry that)) { + return false; + } + return column.equals(that.column) && opClass.equals(that.opClass); + } + + @Override + public int hashCode() { + return java.util.Objects.hash(column, opClass); + } + } + + private final Set entries = Sets.newLinkedHashSet(); + + static Set extract(Expression expression) { + PredicateColumnExtractor visitor = new PredicateColumnExtractor(); + ExpressionVisitors.visit(expression, visitor); + return visitor.entries; + } + + @Override + public Void predicate(BoundPredicate pred) { + if (pred.term() instanceof BoundReference ref) { + entries.add(new Entry(ref.name(), opClass(pred.op()))); + } + return null; + } + + @Override + public Void predicate(UnboundPredicate pred) { + String name = ExpressionUtil.describe(pred.term()); + if (name != null) { + entries.add(new Entry(name, opClass(pred.op()))); + } + return null; + } + + @Override + public Void and(Void left, Void right) { + return null; + } + + @Override + public Void or(Void left, Void right) { + return null; + } + + @Override + public Void not(Void result) { + return null; + } + + @Override + public Void alwaysTrue() { + return null; + } + + @Override + public Void alwaysFalse() { + return null; + } + + @VisibleForTesting + static String opClass(Expression.Operation op) { + return switch (op) { + case EQ, IN -> "point"; + case LT, LT_EQ, GT, GT_EQ -> "range"; + case IS_NULL, NOT_NULL, IS_NAN, NOT_NAN -> "null"; + default -> "other"; + }; + } + } +} diff --git a/gcp/src/test/java/org/apache/iceberg/gcp/metrics/TestCloudMonitoringMetricsReporter.java b/gcp/src/test/java/org/apache/iceberg/gcp/metrics/TestCloudMonitoringMetricsReporter.java new file mode 100644 index 000000000000..c0234afc86e8 --- /dev/null +++ b/gcp/src/test/java/org/apache/iceberg/gcp/metrics/TestCloudMonitoringMetricsReporter.java @@ -0,0 +1,562 @@ +/* + * 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.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import com.google.cloud.monitoring.v3.MetricServiceClient; +import com.google.monitoring.v3.CreateTimeSeriesRequest; +import com.google.monitoring.v3.ProjectName; +import com.google.monitoring.v3.TimeSeries; +import java.time.Duration; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.gcp.GCPProperties; +import org.apache.iceberg.metrics.CommitMetricsResult; +import org.apache.iceberg.metrics.CommitReport; +import org.apache.iceberg.metrics.CounterResult; +import org.apache.iceberg.metrics.ImmutableCommitMetricsResult; +import org.apache.iceberg.metrics.ImmutableCommitReport; +import org.apache.iceberg.metrics.ImmutableScanMetricsResult; +import org.apache.iceberg.metrics.ImmutableScanReport; +import org.apache.iceberg.metrics.MetricsContext.Unit; +import org.apache.iceberg.metrics.ScanMetricsResult; +import org.apache.iceberg.metrics.ScanReport; +import org.apache.iceberg.metrics.TimerResult; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +class TestCloudMonitoringMetricsReporter { + + private static final String PROJECT_ID = "test-project"; + private static final String TABLE_NAME = "db.t"; + private static final String METRIC_PREFIX = "custom.googleapis.com/iceberg"; + + private static GCPProperties properties() { + return new GCPProperties(ImmutableMap.of(GCPProperties.GCP_MONITORING_PROJECT_ID, PROJECT_ID)); + } + + private static GCPProperties propertiesWithPredicateTracking() { + return new GCPProperties( + ImmutableMap.of( + GCPProperties.GCP_MONITORING_PROJECT_ID, + PROJECT_ID, + GCPProperties.GCP_MONITORING_PREDICATE_TRACKING_ENABLED, + "true")); + } + + private static GCPProperties propertiesWithColumnCap(int cap) { + return new GCPProperties( + ImmutableMap.of( + GCPProperties.GCP_MONITORING_PROJECT_ID, + PROJECT_ID, + GCPProperties.GCP_MONITORING_PREDICATE_TRACKING_ENABLED, + "true", + GCPProperties.GCP_MONITORING_COLUMN_CAP, + String.valueOf(cap))); + } + + private static CloudMonitoringMetricsReporter reporter(MetricServiceClient client) { + return reporter(client, MoreExecutors.newDirectExecutorService()); + } + + private static CloudMonitoringMetricsReporter reporter( + MetricServiceClient client, ExecutorService executor) { + return new CloudMonitoringMetricsReporter(properties(), client, executor); + } + + private static CloudMonitoringMetricsReporter reporter( + GCPProperties props, MetricServiceClient client) { + return new CloudMonitoringMetricsReporter( + props, client, MoreExecutors.newDirectExecutorService()); + } + + private static ScanReport scanReport(ScanMetricsResult metrics) { + return ImmutableScanReport.builder() + .tableName(TABLE_NAME) + .snapshotId(1L) + .filter(Expressions.alwaysTrue()) + .schemaId(0) + .scanMetrics(metrics) + .build(); + } + + private static CommitReport commitReport(CommitMetricsResult metrics) { + return ImmutableCommitReport.builder() + .tableName(TABLE_NAME) + .snapshotId(1L) + .sequenceNumber(1L) + .operation("append") + .commitMetrics(metrics) + .build(); + } + + private static ScanMetricsResult emptyScanMetrics() { + return ImmutableScanMetricsResult.builder().build(); + } + + private static CommitMetricsResult emptyCommitMetrics() { + return ImmutableCommitMetricsResult.builder().build(); + } + + private static CreateTimeSeriesRequest captureRequest(MetricServiceClient client) { + ArgumentCaptor captor = + ArgumentCaptor.forClass(CreateTimeSeriesRequest.class); + verify(client).createTimeSeries(captor.capture()); + return captor.getValue(); + } + + private static TimeSeries seriesByMetricName(CreateTimeSeriesRequest request, String metricName) { + return request.getTimeSeriesList().stream() + .filter(ts -> metricName.equals(ts.getMetric().getLabelsOrDefault("metric_name", ""))) + .findFirst() + .orElseThrow(() -> new AssertionError("metric_name=" + metricName + " not found")); + } + + private static Set metricNamesIn(CreateTimeSeriesRequest request) { + return request.getTimeSeriesList().stream() + .map(ts -> ts.getMetric().getLabelsOrDefault("metric_name", "")) + .collect(Collectors.toSet()); + } + + @Test + void reportScan_emitsExpectedSeriesWithFixedLabelSet() { + MetricServiceClient client = mock(MetricServiceClient.class); + + ScanMetricsResult metrics = + ImmutableScanMetricsResult.builder() + .skippedDataFiles(CounterResult.of(Unit.COUNT, 10L)) + .resultDataFiles(CounterResult.of(Unit.COUNT, 3L)) + .totalPlanningDuration(TimerResult.of(TimeUnit.MILLISECONDS, Duration.ofMillis(42), 1)) + .build(); + + reporter(client).report(scanReport(metrics)); + + CreateTimeSeriesRequest request = captureRequest(client); + assertThat(request.getName()).isEqualTo(ProjectName.of(PROJECT_ID).toString()); + + // Operation + 3 non-null metrics = 4 series total. + assertThat(request.getTimeSeriesCount()).isEqualTo(4); + assertThat(metricNamesIn(request)) + .containsExactlyInAnyOrder( + "scan_operations", + "skipped_data_files", + "result_data_files", + "total_planning_duration_ms"); + + TimeSeries scanOps = seriesByMetricName(request, "scan_operations"); + assertThat(scanOps.getMetric().getType()).isEqualTo(METRIC_PREFIX + "/scan_metrics"); + assertThat(scanOps.getMetric().getLabelsMap().keySet()) + .isEqualTo(CloudMonitoringMetricsReporter.SCAN_LABEL_KEYS); + assertThat(scanOps.getMetric().getLabelsOrThrow("table")).isEqualTo(TABLE_NAME); + assertThat(scanOps.getPointsList()).hasSize(1); + assertThat(scanOps.getPoints(0).getValue().getInt64Value()).isEqualTo(1L); + + assertThat( + seriesByMetricName(request, "skipped_data_files") + .getPoints(0) + .getValue() + .getInt64Value()) + .isEqualTo(10L); + assertThat( + seriesByMetricName(request, "result_data_files") + .getPoints(0) + .getValue() + .getInt64Value()) + .isEqualTo(3L); + assertThat( + seriesByMetricName(request, "total_planning_duration_ms") + .getPoints(0) + .getValue() + .getInt64Value()) + .isEqualTo(42L); + } + + @Test + void reportCommit_emitsExpectedSeriesWithOperationLabel() { + MetricServiceClient client = mock(MetricServiceClient.class); + + CommitMetricsResult metrics = + ImmutableCommitMetricsResult.builder() + .addedDataFiles(CounterResult.of(Unit.COUNT, 5L)) + .totalDuration(TimerResult.of(TimeUnit.MILLISECONDS, Duration.ofMillis(123), 1)) + .build(); + + reporter(client).report(commitReport(metrics)); + + CreateTimeSeriesRequest request = captureRequest(client); + + assertThat(metricNamesIn(request)) + .containsExactlyInAnyOrder("commit_operations", "added_data_files", "total_duration_ms"); + + TimeSeries commitOps = seriesByMetricName(request, "commit_operations"); + assertThat(commitOps.getMetric().getType()).isEqualTo(METRIC_PREFIX + "/commit_metrics"); + assertThat(commitOps.getMetric().getLabelsMap().keySet()) + .isEqualTo(CloudMonitoringMetricsReporter.COMMIT_LABEL_KEYS); + assertThat(commitOps.getMetric().getLabelsOrThrow("operation")).isEqualTo("append"); + assertThat( + seriesByMetricName(request, "added_data_files").getPoints(0).getValue().getInt64Value()) + .isEqualTo(5L); + assertThat( + seriesByMetricName(request, "total_duration_ms") + .getPoints(0) + .getValue() + .getInt64Value()) + .isEqualTo(123L); + } + + @Test + void reportScan_emitsOnlyOperationSeriesWhenAllMetricsAreNull() { + MetricServiceClient client = mock(MetricServiceClient.class); + + reporter(client).report(scanReport(emptyScanMetrics())); + + CreateTimeSeriesRequest request = captureRequest(client); + assertThat(request.getTimeSeriesCount()).isEqualTo(1); + assertThat(metricNamesIn(request)).containsExactly("scan_operations"); + } + + @Test + void reportCommit_emitsOnlyOperationSeriesWhenAllMetricsAreNull() { + MetricServiceClient client = mock(MetricServiceClient.class); + + reporter(client).report(commitReport(emptyCommitMetrics())); + + CreateTimeSeriesRequest request = captureRequest(client); + assertThat(request.getTimeSeriesCount()).isEqualTo(1); + assertThat(metricNamesIn(request)).containsExactly("commit_operations"); + } + + @Test + void labelSetIsFixed_noPredicateOrColumnLabels() { + MetricServiceClient client = mock(MetricServiceClient.class); + ScanReport report = + ImmutableScanReport.builder() + .tableName(TABLE_NAME) + .snapshotId(1L) + .filter(Expressions.equal("id", 1)) + .schemaId(0) + .addProjectedFieldIds(1, 2, 3) + .addProjectedFieldNames("a", "b", "c") + .scanMetrics(emptyScanMetrics()) + .build(); + + reporter(client).report(report); + + CreateTimeSeriesRequest request = captureRequest(client); + for (TimeSeries ts : request.getTimeSeriesList()) { + Map labels = ts.getMetric().getLabelsMap(); + assertThat(labels).doesNotContainKey("filter"); + assertThat(labels).doesNotContainKey("queried_field_ids"); + assertThat(labels).doesNotContainKey("queried_column_names"); + assertThat(labels).doesNotContainKey("projected_field_ids"); + assertThat(labels.keySet()).isEqualTo(CloudMonitoringMetricsReporter.SCAN_LABEL_KEYS); + } + } + + @Test + void report_nullReport_isNoOp() { + MetricServiceClient client = mock(MetricServiceClient.class); + + reporter(client).report(null); + + verify(client, never()).createTimeSeries(any(CreateTimeSeriesRequest.class)); + } + + @Test + void report_unsupportedReportType_isNoOp() { + MetricServiceClient client = mock(MetricServiceClient.class); + + reporter(client).report(new UnknownReport()); + + verify(client, never()).createTimeSeries(any(CreateTimeSeriesRequest.class)); + } + + @Test + void report_clientThrows_doesNotPropagate() { + MetricServiceClient client = mock(MetricServiceClient.class); + doThrow(new RuntimeException("boom")) + .when(client) + .createTimeSeries(any(CreateTimeSeriesRequest.class)); + + // Must not throw; Tasks.suppressFailureWhenFinished swallows and logs. + reporter(client).report(scanReport(emptyScanMetrics())); + } + + @Test + void close_closesClientAndIsIdempotent() { + MetricServiceClient client = mock(MetricServiceClient.class); + CloudMonitoringMetricsReporter underTest = reporter(client); + + underTest.close(); + underTest.close(); + + verify(client, times(1)).close(); + } + + @Test + void report_afterClose_isNoOp() { + MetricServiceClient client = mock(MetricServiceClient.class); + CloudMonitoringMetricsReporter underTest = reporter(client); + + underTest.close(); + underTest.report(scanReport(emptyScanMetrics())); + + verify(client, never()).createTimeSeries(any(CreateTimeSeriesRequest.class)); + } + + @Test + void initialize_missingProjectId_throws() { + GCPProperties bad = new GCPProperties(ImmutableMap.of()); + org.assertj.core.api.Assertions.assertThatThrownBy( + () -> + new CloudMonitoringMetricsReporter( + bad, mock(MetricServiceClient.class), MoreExecutors.newDirectExecutorService())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(GCPProperties.GCP_MONITORING_PROJECT_ID); + } + + @Test + void scanAndCommitMetricMaps_keysAreStableSnakeCase() { + // Snapshot the expected metric names to catch accidental renames/removals. + assertThat(CloudMonitoringMetricsReporter.SCAN_METRICS.keySet()) + .containsExactlyInAnyOrder( + "total_planning_duration_ms", + "result_data_files", + "result_delete_files", + "total_data_manifests", + "total_delete_manifests", + "scanned_data_manifests", + "skipped_data_manifests", + "total_file_size_bytes", + "total_delete_file_size_bytes", + "skipped_data_files", + "skipped_delete_files", + "scanned_delete_manifests", + "skipped_delete_manifests", + "indexed_delete_files", + "equality_delete_files", + "positional_delete_files", + "dvs"); + + assertThat(CloudMonitoringMetricsReporter.COMMIT_METRICS.keySet()) + .containsExactlyInAnyOrder( + "total_duration_ms", + "attempts", + "added_data_files", + "removed_data_files", + "total_data_files", + "added_delete_files", + "added_equality_delete_files", + "added_positional_delete_files", + "added_dvs", + "removed_delete_files", + "removed_equality_delete_files", + "removed_positional_delete_files", + "removed_dvs", + "total_delete_files", + "added_records", + "removed_records", + "total_records", + "added_files_size_bytes", + "removed_files_size_bytes", + "total_files_size_bytes", + "added_positional_deletes", + "removed_positional_deletes", + "total_positional_deletes", + "added_equality_deletes", + "removed_equality_deletes", + "total_equality_deletes", + "manifests_created", + "manifests_replaced", + "manifests_kept", + "manifest_entries_processed"); + } + + // --------------------------------------------------------------------------------------------- + // Predicate / projection metric tests + // --------------------------------------------------------------------------------------------- + + private static ScanReport scanReportWith( + org.apache.iceberg.expressions.Expression filter, java.util.List projection) { + ImmutableScanReport.Builder builder = + ImmutableScanReport.builder() + .tableName(TABLE_NAME) + .snapshotId(1L) + .filter(filter) + .schemaId(0) + .scanMetrics(ImmutableScanMetricsResult.builder().build()); + for (int i = 0; i < projection.size(); i++) { + builder.addProjectedFieldIds(i + 1); + builder.addProjectedFieldNames(projection.get(i)); + } + return builder.build(); + } + + private static java.util.List seriesByMetricType( + CreateTimeSeriesRequest request, String metricType) { + return request.getTimeSeriesList().stream() + .filter(ts -> metricType.equals(ts.getMetric().getType())) + .collect(Collectors.toList()); + } + + @Test + void predicateTrackingDisabled_emitsNoPredicateOrProjectionSeries() { + MetricServiceClient client = mock(MetricServiceClient.class); + + reporter(client) + .report( + scanReportWith( + org.apache.iceberg.expressions.Expressions.equal("user_id", 1), + java.util.List.of("user_id", "ts"))); + + CreateTimeSeriesRequest request = captureRequest(client); + assertThat(seriesByMetricType(request, METRIC_PREFIX + "/predicate_columns")).isEmpty(); + assertThat(seriesByMetricType(request, METRIC_PREFIX + "/projection_columns")).isEmpty(); + } + + @Test + void predicateTrackingEnabled_emitsPredicateAndProjectionSeriesWithFixedLabels() { + MetricServiceClient client = mock(MetricServiceClient.class); + + reporter(propertiesWithPredicateTracking(), client) + .report( + scanReportWith( + org.apache.iceberg.expressions.Expressions.and( + org.apache.iceberg.expressions.Expressions.equal("user_id", 1), + org.apache.iceberg.expressions.Expressions.greaterThan("ts", 100)), + java.util.List.of("user_id", "ts", "value"))); + + CreateTimeSeriesRequest request = captureRequest(client); + + java.util.List predicateSeries = + seriesByMetricType(request, METRIC_PREFIX + "/predicate_columns"); + java.util.List projectionSeries = + seriesByMetricType(request, METRIC_PREFIX + "/projection_columns"); + + assertThat(predicateSeries).hasSize(2); + for (TimeSeries ts : predicateSeries) { + assertThat(ts.getMetric().getLabelsMap().keySet()) + .isEqualTo(CloudMonitoringMetricsReporter.PREDICATE_LABEL_KEYS); + assertThat(ts.getPoints(0).getValue().getInt64Value()).isEqualTo(1L); + } + + assertThat(predicateColumnsByOpClass(predicateSeries)) + .containsExactlyInAnyOrderEntriesOf(ImmutableMap.of("user_id", "point", "ts", "range")); + + assertThat(projectionSeries).hasSize(3); + for (TimeSeries ts : projectionSeries) { + assertThat(ts.getMetric().getLabelsMap().keySet()) + .isEqualTo(CloudMonitoringMetricsReporter.PROJECTION_LABEL_KEYS); + assertThat(ts.getPoints(0).getValue().getInt64Value()).isEqualTo(1L); + } + assertThat( + projectionSeries.stream() + .map(ts -> ts.getMetric().getLabelsOrThrow("column")) + .collect(Collectors.toSet())) + .containsExactlyInAnyOrder("user_id", "ts", "value"); + } + + @Test + void predicateTracking_dropsEmissionWhenColumnCapExceeded() { + MetricServiceClient client = mock(MetricServiceClient.class); + + // 3 unique columns; cap = 2 → predicate/projection emission must be dropped. + reporter(propertiesWithColumnCap(2), client) + .report( + scanReportWith( + org.apache.iceberg.expressions.Expressions.equal("a", 1), + java.util.List.of("b", "c"))); + + CreateTimeSeriesRequest request = captureRequest(client); + assertThat(seriesByMetricType(request, METRIC_PREFIX + "/predicate_columns")).isEmpty(); + assertThat(seriesByMetricType(request, METRIC_PREFIX + "/projection_columns")).isEmpty(); + // Base scan series still present. + assertThat(seriesByMetricType(request, METRIC_PREFIX + "/scan_metrics")).isNotEmpty(); + } + + @Test + void opClassBuckets_coverAllRelevantOperations() { + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.EQ)) + .isEqualTo("point"); + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.IN)) + .isEqualTo("point"); + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.LT)) + .isEqualTo("range"); + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.LT_EQ)) + .isEqualTo("range"); + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.GT)) + .isEqualTo("range"); + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.GT_EQ)) + .isEqualTo("range"); + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.IS_NULL)) + .isEqualTo("null"); + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.NOT_NULL)) + .isEqualTo("null"); + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.NOT_EQ)) + .isEqualTo("other"); + assertThat( + CloudMonitoringMetricsReporter.PredicateColumnExtractor.opClass( + org.apache.iceberg.expressions.Expression.Operation.STARTS_WITH)) + .isEqualTo("other"); + } + + private static Map predicateColumnsByOpClass(java.util.List series) { + Map out = new java.util.HashMap<>(); + for (TimeSeries ts : series) { + out.put( + ts.getMetric().getLabelsOrThrow("column"), ts.getMetric().getLabelsOrThrow("op_class")); + } + return out; + } + + private static final class UnknownReport implements org.apache.iceberg.metrics.MetricsReport { + // Anonymous report type that is neither ScanReport nor CommitReport. + } +}