Implement Library Instrumentation for Apache Iceberg

.fossa.yml

@@ -157,6 +157,9 @@ targets:
     - type: gradle
       path: ./
       target: ':instrumentation:hystrix-1.4:javaagent'
+    - type: gradle
+      path: ./
+      target: ':instrumentation:iceberg-1.8:library'
     - type: gradle
       path: ./
       target: ':instrumentation:influxdb-2.4:javaagent'

docs/supported-libraries.md

@@ -32,6 +32,7 @@ These are the supported libraries and frameworks:
 | [Apache ElasticJob](https://shardingsphere.apache.org/elasticjob/)                                                                          | 3.0+                               | N/A                                                                                                                                                                                                                                                                                                                                                                                     | none                                                                                                                                              |
 | [Apache HttpAsyncClient](https://hc.apache.org/index.html)                                                                                  | 4.1+                               | N/A                                                                                                                                                                                                                                                                                                                                                                                     | [HTTP Client Spans], [HTTP Client Metrics]                                                                                                        |
 | [Apache HttpClient](https://hc.apache.org/index.html)                                                                                       | 2.0+                               | [opentelemetry-apache-httpclient-4.3](../instrumentation/apache-httpclient/apache-httpclient-4.3/library),<br>[opentelemetry-apache-httpclient-5.2](../instrumentation/apache-httpclient/apache-httpclient-5.2/library)                                                                                                                                                                 | [HTTP Client Spans], [HTTP Client Metrics]                                                                                                        |
+| [Apache Iceberg](https://iceberg.apache.org/)                                                                                               | N/A                                | [opentelemetry-iceberg-1.8](../instrumentation/iceberg-1.8/library/)                                                                                                                                                                                                                                                                                                                    | none                                                                                                                                              |
 | [Apache ShenYu](https://shenyu.apache.org/)                                                                                                 | 2.4+                               | N/A                                                                                                                                                                                                                                                                                                                                                                                     | Provides `http.route` [2]                                                                                                                         |
 | [Apache Kafka Connect API](https://kafka.apache.org/documentation/#connect)                                                                 | 2.6+                               | N/A                                                                                                                                                                                                                                                                                                                                                                                     | [Messaging Spans]                                                                                                                                 |
 | [Apache Kafka Producer/Consumer API](https://kafka.apache.org/documentation/#producerapi)                                                   | 0.11+                              | [opentelemetry-kafka-clients-2.6](../instrumentation/kafka/kafka-clients/kafka-clients-2.6/library)                                                                                                                                                                                                                                                                                     | [Messaging Spans]                                                                                                                                 |

instrumentation/iceberg-1.8/README.md

@@ -0,0 +1,43 @@
+# Library Instrumentation for Apache Iceberg Version 1.8 and Higher
+
+Provides OpenTelemetry instrumentation for [Apache Iceberg](https://iceberg.apache.org/).
+
+## Quickstart
+
+### Add These Dependencies to Your Project
+
+Replace `OPENTELEMETRY_VERSION` with the [latest release](https://central.sonatype.com/artifact/io.opentelemetry.instrumentation/opentelemetry-iceberg-1.8).
+
+For Maven, add to your `pom.xml` dependencies:
+
+```xml
+<dependencies>
+  <dependency>
+    <groupId>io.opentelemetry.instrumentation</groupId>
+    <artifactId>opentelemetry-iceberg-1.8</artifactId>
+    <version>OPENTELEMETRY_VERSION</version>
+  </dependency>
+</dependencies>
+```
+
+For Gradle, add to your dependencies:
+
+```groovy
+implementation("io.opentelemetry.instrumentation:opentelemetry-iceberg-1.8:OPENTELEMETRY_VERSION")
+```
+
+### Usage
+
+The instrumentation library allows creating instrumented `Scan` (e.g., `TableScan`) instances for collecting and reporting OpenTelemetry-based scan metrics. For example:
+
+```java
+OpenTelemetry openTelemetry = // ...
+IcebergTelemetry icebergTelemetry = IcebergTelemetry.create(openTelemetry);
+TableScan tableScan = icebergTelemetry.wrapScan(table.newScan());
+
+try (CloseableIterable<FileScanTask> fileScanTasks = tableScan.planFiles()) {
+  // Process the scan tasks
+}
+
+// The metrics will be reported after the scan tasks iterable is closed
+```

instrumentation/iceberg-1.8/library/build.gradle.kts

@@ -0,0 +1,14 @@
+plugins {
+  id("otel.library-instrumentation")
+  id("otel.nullaway-conventions")
+}
+
+dependencies {
+  library("org.apache.iceberg:iceberg-core:1.8.1")
+  testImplementation(project(":instrumentation:iceberg-1.8:testing"))
+  implementation("io.opentelemetry:opentelemetry-api-incubator")
+}
+
+otelJava {
+  minJavaVersionSupported.set(JavaVersion.VERSION_11)
+}

instrumentation/iceberg-1.8/library/src/main/java/io/opentelemetry/instrumentation/iceberg/v1_8/IcebergMetricsReporter.java

@@ -0,0 +1,279 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.iceberg.v1_8;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.incubator.metrics.ExtendedDoubleHistogramBuilder;
+import io.opentelemetry.api.incubator.metrics.ExtendedLongCounterBuilder;
+import io.opentelemetry.api.metrics.DoubleHistogram;
+import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
+import io.opentelemetry.api.metrics.LongCounter;
+import io.opentelemetry.api.metrics.LongCounterBuilder;
+import io.opentelemetry.api.metrics.Meter;
+import java.util.List;
+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;
+
+final class IcebergMetricsReporter implements MetricsReporter {
+  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.iceberg-1.8";
+  private static final AttributeKey<Long> SCHEMA_ID = AttributeKey.longKey("iceberg.schema.id");
+  private static final AttributeKey<String> TABLE_NAME =
+      AttributeKey.stringKey("iceberg.table.name");
+  private static final AttributeKey<Long> SNAPSHOT_ID = AttributeKey.longKey("iceberg.snapshot.id");
+  private static final AttributeKey<String> SCAN_STATE =
+      AttributeKey.stringKey("iceberg.scan.state");
+  private static final AttributeKey<String> DELETE_TYPE =
+      AttributeKey.stringKey("iceberg.delete_file.type");
+  private static final List<AttributeKey<?>> BASE_ADVICE = List.of(SCHEMA_ID, TABLE_NAME);
+  private static final List<AttributeKey<?>> ADVICE_FOR_DATA_AND_MANIFEST_FILE_COUNTS =
+      List.of(SCHEMA_ID, TABLE_NAME, SCAN_STATE);
+  private static final List<AttributeKey<?>> ADVICE_FOR_DELETE_FILE_COUNTS =
+      List.of(SCHEMA_ID, TABLE_NAME, SCAN_STATE, DELETE_TYPE);
+
+  private final DoubleHistogram planningDuration;
+  private final LongCounter dataFilesCount;
+  private final LongCounter dataFilesSize;
+  private final LongCounter deleteFilesCount;
+  private final LongCounter deleteFilesSize;
+  private final LongCounter dataManifestsCount;
+  private final LongCounter deleteManifestsCount;
+
+  IcebergMetricsReporter(OpenTelemetry openTelemetry) {
+    Meter meter = openTelemetry.getMeter(INSTRUMENTATION_NAME);
+
+    planningDuration =
+        applyAdvice(BASE_ADVICE, ScanMetricsBuilderFactory.totalPlanningDuration(meter, "s"))
+            .build();
+    dataFilesCount =
+        applyAdvice(
+                ADVICE_FOR_DATA_AND_MANIFEST_FILE_COUNTS,
+                ScanMetricsBuilderFactory.dataFilesCount(meter))
+            .build();
+    dataFilesSize =
+        applyAdvice(BASE_ADVICE, ScanMetricsBuilderFactory.dataFilesSize(meter)).build();
+    deleteFilesCount =
+        applyAdvice(
+                ADVICE_FOR_DELETE_FILE_COUNTS, ScanMetricsBuilderFactory.deleteFilesCount(meter))
+            .build();
+    deleteFilesSize =
+        applyAdvice(BASE_ADVICE, ScanMetricsBuilderFactory.deleteFilesSize(meter)).build();
+    dataManifestsCount =
+        applyAdvice(
+                ADVICE_FOR_DATA_AND_MANIFEST_FILE_COUNTS,
+                ScanMetricsBuilderFactory.dataManifestsCount(meter))
+            .build();
+    deleteManifestsCount =
+        applyAdvice(
+                ADVICE_FOR_DATA_AND_MANIFEST_FILE_COUNTS,
+                ScanMetricsBuilderFactory.deleteManifestsCount(meter))
+            .build();
+  }
+
+  @Override
+  public void report(MetricsReport report) {
+    if (report instanceof ScanReport) {
+      reportScanMetrics((ScanReport) report);
+    }
+  }
+
+  void reportScanMetrics(ScanReport scanReport) {
+    Attributes scanAttributes =
+        Attributes.of(
+            SCHEMA_ID,
+            (long) scanReport.schemaId(),
+            TABLE_NAME,
+            scanReport.tableName(),
+            SNAPSHOT_ID,
+            scanReport.snapshotId());
+    ScanMetricsResult metrics = scanReport.scanMetrics();
+    TimerResult duration = metrics.totalPlanningDuration();
+
+    if (duration != null) {
+      planningDuration.record(duration.totalDuration().toMillis() / 1000.0, scanAttributes);
+    }
+
+    // Data files metrics
+    CounterResult current = metrics.resultDataFiles();
+
+    if (current != null) {
+      addValueToLongCounter(dataFilesCount, current.value(), scanAttributes, SCAN_STATE, "scanned");
+    }
+
+    current = metrics.skippedDataFiles();
+
+    if (current != null) {
+      addValueToLongCounter(dataFilesCount, current.value(), scanAttributes, SCAN_STATE, "skipped");
+    }
+
+    current = metrics.totalFileSizeInBytes();
+
+    if (current != null) {
+      dataFilesSize.add(current.value(), scanAttributes);
+    }
+
+    // Delete files metrics
+    current = metrics.totalDeleteFileSizeInBytes();
+
+    if (current != null) {
+      deleteFilesSize.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.resultDeleteFiles();
+
+    if (current != null) {
+      addValueToLongCounter(
+          deleteFilesCount,
+          current.value(),
+          scanAttributes,
+          SCAN_STATE,
+          "scanned",
+          DELETE_TYPE,
+          "all");
+    }
+
+    current = metrics.skippedDeleteFiles();
+
+    if (current != null) {
+      addValueToLongCounter(
+          deleteFilesCount,
+          current.value(),
+          scanAttributes,
+          SCAN_STATE,
+          "skipped",
+          DELETE_TYPE,
+          "all");
+    }
+
+    current = metrics.indexedDeleteFiles();
+
+    if (current != null) {
+      addValueToLongCounter(
+          deleteFilesCount,
+          current.value(),
+          scanAttributes,
+          SCAN_STATE,
+          "scanned",
+          DELETE_TYPE,
+          "indexed");
+    }
+
+    current = metrics.equalityDeleteFiles();
+
+    if (current != null) {
+      addValueToLongCounter(
+          deleteFilesCount,
+          current.value(),
+          scanAttributes,
+          SCAN_STATE,
+          "scanned",
+          DELETE_TYPE,
+          "equality");
+    }
+
+    current = metrics.positionalDeleteFiles();
+
+    if (current != null) {
+      addValueToLongCounter(
+          deleteFilesCount,
+          current.value(),
+          scanAttributes,
+          SCAN_STATE,
+          "scanned",
+          DELETE_TYPE,
+          "position");
+    }
+
+    current = metrics.dvs();
+
+    if (current != null) {
+      addValueToLongCounter(
+          deleteFilesCount,
+          current.value(),
+          scanAttributes,
+          SCAN_STATE,
+          "scanned",
+          DELETE_TYPE,
+          "dvs");
+    }
+
+    // Data manifests metrics
+    current = metrics.scannedDataManifests();
+
+    if (current != null) {
+      addValueToLongCounter(
+          dataManifestsCount, current.value(), scanAttributes, SCAN_STATE, "scanned");
+    }
+
+    current = metrics.skippedDataManifests();
+
+    if (current != null) {
+      addValueToLongCounter(
+          dataManifestsCount, current.value(), scanAttributes, SCAN_STATE, "skipped");
+    }
+
+    // Delete manifests metrics
+    current = metrics.scannedDeleteManifests();
+
+    if (current != null) {
+      addValueToLongCounter(
+          deleteManifestsCount, current.value(), scanAttributes, SCAN_STATE, "scanned");
+    }
+
+    current = metrics.skippedDeleteManifests();
+
+    if (current != null) {
+      addValueToLongCounter(
+          deleteManifestsCount, current.value(), scanAttributes, SCAN_STATE, "skipped");
+    }
+  }
+
+  private static void addValueToLongCounter(
+      LongCounter metric,
+      long measurement,
+      Attributes attributes,
+      AttributeKey<String> att1Key,
+      String att1Value) {
+    Attributes newAttributes = attributes.toBuilder().put(att1Key, att1Value).build();
+    metric.add(measurement, newAttributes);
+  }
+
+  private static void addValueToLongCounter(
+      LongCounter metric,
+      long measurement,
+      Attributes attributes,
+      AttributeKey<String> att1Key,
+      String att1Value,
+      AttributeKey<String> att2Key,
+      String att2Value) {
+    Attributes newAttributes =
+        attributes.toBuilder().put(att1Key, att1Value).put(att2Key, att2Value).build();
+    metric.add(measurement, newAttributes);
+  }
+
+  private static LongCounterBuilder applyAdvice(
+      List<AttributeKey<?>> attributeKeys, LongCounterBuilder builder) {
+    if (builder instanceof ExtendedLongCounterBuilder) {
+      return ((ExtendedLongCounterBuilder) builder).setAttributesAdvice(attributeKeys);
+    }
+
+    return builder;
+  }
+
+  private static DoubleHistogramBuilder applyAdvice(
+      List<AttributeKey<?>> attributeKeys, DoubleHistogramBuilder builder) {
+    if (builder instanceof ExtendedDoubleHistogramBuilder) {
+      return ((ExtendedDoubleHistogramBuilder) builder).setAttributesAdvice(attributeKeys);
+    }
+
+    return builder;
+  }
+}

instrumentation/iceberg-1.8/library/src/main/java/io/opentelemetry/instrumentation/iceberg/v1_8/IcebergTelemetry.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.iceberg.v1_8;
+
+import io.opentelemetry.api.OpenTelemetry;
+import org.apache.iceberg.Scan;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.ScanTaskGroup;
+
+/** Entrypoint for instrumenting Apache Iceberg scan metrics */
+public final class IcebergTelemetry {
+  private final OpenTelemetry openTelemetry;
+
+  /** Returns a new {@link IcebergTelemetry} configured with the given {@link OpenTelemetry}. */
+  public static IcebergTelemetry create(OpenTelemetry openTelemetry) {
+    return new IcebergTelemetry(openTelemetry);
+  }
+
+  IcebergTelemetry(OpenTelemetry openTelemetry) {
+    this.openTelemetry = openTelemetry;
+  }
+
+  /**
+   * Creates a new {@link Scan} instance based on an existing {@link Scan} instance. The new
+   * instance is associated with a custom {@link org.apache.iceberg.metrics.MetricsReporter} that
+   * reports scan metrics using the configured {@link OpenTelemetry} instance.
+   *
+   * @param <T1> the child class, returned by method chaining, e.g., {@link
+   *     Scan#project(org.apache.iceberg.Schema)}
+   * @param <T2> the type of tasks produced by this scan
+   * @param <T3> the type of task groups produced by this scan
+   * @param scan the original scan instance that will be instrumented
+   * @return an instrumented {@link Scan} instance based on the provided instance
+   */
+  public <T1, T2 extends ScanTask, T3 extends ScanTaskGroup<T2>> T1 wrapScan(
+      Scan<T1, T2, T3> scan) {
+    return scan.metricsReporter(new IcebergMetricsReporter(openTelemetry));
+  }
+}