Implement Library Instrumentation for Apache Iceberg

.fossa.yml

@@ -154,6 +154,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'

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 sacn 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,13 @@
+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"))
+}
+
+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,194 @@
+/*
+ * 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.metrics.LongCounter;
+import io.opentelemetry.api.metrics.LongGauge;
+import java.util.Locale;
+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;
+
+public 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> SNAPHSOT_ID = AttributeKey.longKey("iceberg.snapshot.id");
+
+  private final OpenTelemetry openTelemetry;
+
+  IcebergMetricsReporter(OpenTelemetry openTelemetry) {
+    this.openTelemetry = openTelemetry;
+  }
+
+  @Override
+  public void report(MetricsReport report) {
+    if (report instanceof ScanReport) {
+      reportScanMetrics((ScanReport) report);
+    }
+  }
+
+  void reportScanMetrics(ScanReport scanReport) {
+    Attributes scanAttributes =
+        Attributes.of(
+            SCHEMA_ID,
+            Long.valueOf(scanReport.schemaId()),
+            TABLE_NAME,
+            scanReport.tableName(),
+            SNAPHSOT_ID,
+            scanReport.snapshotId());
+    ScanMetricsResult metrics = scanReport.scanMetrics();
+    TimerResult duration = metrics.totalPlanningDuration();
+
+    if (duration != null) {
+      LongGauge metric =
+          ScanMetricsBuilder.totalPlanningDuration(
+              openTelemetry.getMeter(INSTRUMENTATION_NAME),
+              duration.timeUnit().name().toLowerCase(Locale.getDefault()));
+      metric.set(duration.totalDuration().toMillis(), scanAttributes);
+    }
+
+    CounterResult current = metrics.resultDataFiles();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.scannedDataFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.resultDeleteFiles();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.scannedDeleteFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.scannedDataManifests();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.scannedDataManifestsCount(
+              openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.scannedDeleteManifests();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.scannedDeleteManifestsCount(
+              openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.totalDataManifests();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.totalDataManifestsCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.totalDeleteManifests();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.totalDeleteManifestsCount(
+              openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.totalFileSizeInBytes();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.scannedDataFilesSize(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.totalDeleteFileSizeInBytes();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.scannedDeleteFilesSize(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.skippedDataManifests();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.skippedDataManifestsCount(
+              openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.skippedDeleteManifests();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.skippedDeleteManifestsCount(
+              openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.skippedDataFiles();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.skippedDataFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.skippedDeleteFiles();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.skippedDeleteFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.indexedDeleteFiles();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.indexedDeleteFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.equalityDeleteFiles();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.equalityDeleteFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.positionalDeleteFiles();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.positionDeleteFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+
+    current = metrics.dvs();
+
+    if (current != null) {
+      LongCounter metric =
+          ScanMetricsBuilder.deletionVectorFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.add(current.value(), scanAttributes);
+    }
+  }
+}

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

@@ -0,0 +1,28 @@
+/*
+ * 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;
+
+public class IcebergTelemetry {
+  private final OpenTelemetry openTelemetry;
+
+  public static IcebergTelemetry create(OpenTelemetry openTelemetry) {
+    return new IcebergTelemetry(openTelemetry);
+  }
+
+  IcebergTelemetry(OpenTelemetry openTelemetry) {
+    this.openTelemetry = openTelemetry;
+  }
+
+  public <T1, T2 extends ScanTask, T3 extends ScanTaskGroup<T2>> T1 wrapScan(
+      Scan<T1, T2, T3> scan) {
+    return scan.metricsReporter(new IcebergMetricsReporter(openTelemetry));
+  }
+}