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/library/build.gradle.kts

@@ -0,0 +1,7 @@
+plugins {
+  id("otel.library-instrumentation")
+}
+
+dependencies {
+  implementation("org.apache.iceberg:iceberg-core:1.8.1")
+}

instrumentation/iceberg-1.8/library/src/main/java/opentlelemetry/instrumentation/iceberg/v1_6/IcebergMetricsReporter.java

@@ -0,0 +1,178 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package opentlelemetry.instrumentation.iceberg.v1_6;
+
+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 io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.metrics.DoubleGauge;
+
+public class IcebergMetricsReporter implements MetricsReporter {
+  private static final String INSTRUMENTATION_NAME = "io.opentelemetry.iceberg_1.6";
+
+  private final OpenTelemetry openTelemetry;
+
+  IcebergMetricsReporter(OpenTelemetry openTelemetry) {
+    this.openTelemetry = openTelemetry;
+  }
+
+  @Override
+  public void report(MetricsReport report) {
+    if (report instanceof ScanReport) {
+      reportScanMetrics((ScanReport) report);
+    } else if (report instanceof CommitReport) {
+      reportCommitMetrics((CommitReport) report);
+    }
+  }
+
+  void reportScanMetrics(ScanReport scanReport) {
+    final ScanMetricsResult metrics = scanReport.scanMetrics();
+    TimerResult duration = metrics.totalPlanningDuration();
+
+    if (duration != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.totalPlanningDuration(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(duration.totalDuration().toMillis());
+    }
+
+    CounterResult current = metrics.resultDataFiles();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.scannedDataFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.resultDeleteFiles();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.scannedDeleteFilesCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.scannedDataManifests();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.scannedDataManifestsCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.scannedDeleteManifests();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.scannedDeleteManifestsCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.totalDataManifests();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.totalDataManifestsCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.totalDeleteManifests();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.totalDeleteManifestsCount(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.totalFileSizeInBytes();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.scannedDataFilesSize(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.totalDeleteFileSizeInBytes();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.scannedDeleteFilesSize(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.skippedDataManifests();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.skippedDataManifests(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.skippedDeleteManifests();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.skippedDeleteManifests(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.skippedDataFiles();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.skippedDataFiles(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.skippedDeleteFiles();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.skippedDeleteFiles(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.indexedDeleteFiles();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.indexedDeleteFiles(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.equalityDeleteFiles();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.equalityDeleteFiles(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.positionalDeleteFiles();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.positionDeleteFiles(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+    current = metrics.dvs();
+
+    if (current != null) {
+      DoubleGauge metric =
+          ScanMetricsBuilder.deletionVectorFiles(openTelemetry.getMeter(INSTRUMENTATION_NAME));
+      metric.set(current.value());
+    }
+
+  }
+
+  void reportCommitMetrics(CommitReport commitReport) {}
+}

instrumentation/iceberg-1.8/library/src/main/java/opentlelemetry/instrumentation/iceberg/v1_6/IcebergTelemetry.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package opentlelemetry.instrumentation.iceberg.v1_6;
+
+import io.opentelemetry.api.OpenTelemetry;
+import org.apache.iceberg.TableScan;
+
+public class IcebergTelemetry {
+  private final OpenTelemetry openTelemetry;
+
+  public static IcebergTelemetry create(OpenTelemetry openTelemetry) {
+    return new IcebergTelemetry(openTelemetry);
+  }
+
+  IcebergTelemetry(OpenTelemetry openTelemetry) {
+    this.openTelemetry = openTelemetry;
+  }
+
+  public TableScan wrapTableScan(TableScan tableScan) {
+    return tableScan.metricsReporter(new IcebergMetricsReporter(openTelemetry));
+  }
+}

instrumentation/iceberg-1.8/library/src/main/java/opentlelemetry/instrumentation/iceberg/v1_6/ScanMetricsBuilder.java

@@ -0,0 +1,168 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package opentlelemetry.instrumentation.iceberg.v1_6;
+
+import io.opentelemetry.api.metrics.DoubleGauge;
+import io.opentelemetry.api.metrics.Meter;
+
+public class ScanMetricsBuilder {
+  private static final String ROOT = "iceberg.scan";
+  private static final String TOTAL_PLANNING_DURATION = ROOT + ".planning.duration";
+  private static final String RESULT_DATA_FILES = ROOT + ".scanned.data_files.count";
+  private static final String RESULT_DELETE_FILES = ROOT + ".scanned.delete_files.count";
+  private static final String SCANNED_DATA_MANIFESTS = ROOT + ".scanned.data_manifests.count";
+  private static final String SCANNED_DELETE_MANIFESTS = ROOT + ".scanned.delete_manifests.count";
+  private static final String TOTAL_DATA_MANIFESTS = ROOT + ".total.data_manifests.count";
+  private static final String TOTAL_DELETE_MANIFESTS = ROOT + ".total.delete_manifests.count";
+  private static final String TOTAL_FILE_SIZE_IN_BYTES = ROOT + ".scanned.data_files.size";
+  private static final String TOTAL_DELETE_FILE_SIZE_IN_BYTES = ROOT + "scanned.delete_files.size";
+  private static final String SKIPPED_DATA_MANIFESTS = ROOT + ".skipped.data_manifests.count";
+  private static final String SKIPPED_DELETE_MANIFESTS = ROOT + ".skipped.delete_manifests.count";
+  private static final String SKIPPED_DATA_FILES = ROOT + ".skipped.data_files.count";
+  private static final String SKIPPED_DELETE_FILES = ROOT + ".skipped.delete_files.count";
+  private static final String INDEXED_DELETE_FILES = ROOT + ".scanned.indexed_delete_files.count";
+  private static final String EQUALITY_DELETE_FILES = ROOT + ".scanned.equality_delete_files.count";
+  private static final String POSITIONAL_DELETE_FILES =
+      ROOT + ".scanned.positional_delete_files.count";
+  private static final String DVS = ROOT + ".scanned.dvs.count";
+
+  static DoubleGauge totalPlanningDuration(Meter meter) {
+    return meter
+        .gaugeBuilder(TOTAL_PLANNING_DURATION)
+        .setDescription("The total duration needed to plan the scan.")
+        .setUnit("ms")
+        .build();
+  }
+
+  static DoubleGauge scannedDataFilesCount(Meter meter) {
+    return meter
+        .gaugeBuilder(RESULT_DATA_FILES)
+        .setDescription("The number of scanned data files.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge scannedDeleteFilesCount(Meter meter) {
+    return meter
+        .gaugeBuilder(RESULT_DELETE_FILES)
+        .setDescription("The number of scanned delete files.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge scannedDataManifestsCount(Meter meter) {
+    return meter
+        .gaugeBuilder(SCANNED_DATA_MANIFESTS)
+        .setDescription("The number of scanned data manifests.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge scannedDeleteManifestsCount(Meter meter) {
+    return meter
+        .gaugeBuilder(SCANNED_DELETE_MANIFESTS)
+        .setDescription("The number of scanned delete manifests.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge totalDataManifestsCount(Meter meter) {
+    return meter
+        .gaugeBuilder(TOTAL_DATA_MANIFESTS)
+        .setDescription("The number of all data manifests.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge totalDeleteManifestsCount(Meter meter) {
+    return meter
+        .gaugeBuilder(TOTAL_DELETE_MANIFESTS)
+        .setDescription("The number of all delete manifests.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge scannedDataFilesSize(Meter meter) {
+    return meter
+        .gaugeBuilder(TOTAL_FILE_SIZE_IN_BYTES)
+        .setDescription("The total size of all scanned data files.")
+        .setUnit("byte")
+        .build();
+  }
+
+  static DoubleGauge scannedDeleteFilesSize(Meter meter) {
+    return meter
+        .gaugeBuilder(TOTAL_DELETE_FILE_SIZE_IN_BYTES)
+        .setDescription("The total size of all scanned delete files.")
+        .setUnit("byte")
+        .build();
+  }
+
+  static DoubleGauge skippedDataManifests(Meter meter) {
+    return meter
+        .gaugeBuilder(SKIPPED_DATA_MANIFESTS)
+        .setDescription("The number of data manifests that were skipped during the scan.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge skippedDeleteManifests(Meter meter) {
+    return meter
+        .gaugeBuilder(SKIPPED_DELETE_MANIFESTS)
+        .setDescription("The number of delete manifests that were skipped during the scan.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge skippedDataFiles(Meter meter) {
+    return meter
+        .gaugeBuilder(SKIPPED_DATA_FILES)
+        .setDescription("The number of data files that were skipped during the scan.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge skippedDeleteFiles(Meter meter) {
+    return meter
+        .gaugeBuilder(SKIPPED_DELETE_FILES)
+        .setDescription("The number of delete files that were skipped during the scan.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge indexedDeleteFiles(Meter meter) {
+    return meter
+        .gaugeBuilder(INDEXED_DELETE_FILES)
+        .setDescription(
+            "The number of delete files constituting the delete file index for this scan.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge equalityDeleteFiles(Meter meter) {
+    return meter
+        .gaugeBuilder(EQUALITY_DELETE_FILES)
+        .setDescription("The number of equality delete files relevant for the current scan.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge positionDeleteFiles(Meter meter) {
+    return meter
+        .gaugeBuilder(POSITIONAL_DELETE_FILES)
+        .setDescription("The number of position delete files relevant for the current scan.")
+        .setUnit("{file}")
+        .build();
+  }
+
+  static DoubleGauge deletionVectorFiles(Meter meter) {
+    return meter
+        .gaugeBuilder(DVS)
+        .setDescription("The number of deletion vector (DV) files relevant for the current scan.")
+        .setUnit("{file}")
+        .build();
+  }
+}

settings.gradle.kts

@@ -303,6 +303,7 @@ include(":instrumentation:hikaricp-3.0:library")
 include(":instrumentation:hikaricp-3.0:testing")
 include(":instrumentation:http-url-connection:javaagent")
 include(":instrumentation:hystrix-1.4:javaagent")
+include(":instrumentation:iceberg-1.8:library")
 include(":instrumentation:influxdb-2.4:javaagent")
 include(":instrumentation:internal:internal-application-logger:bootstrap")
 include(":instrumentation:internal:internal-application-logger:javaagent")