initial testing implementation

Author: ghareeb-falazi

instrumentation/iceberg-1.8/README.md

@@ -0,0 +1,42 @@
+# 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
+OpenTelemetry-based metrics for scans and commits. For example:
+
+```java
+OpenTelemetry openTelemetry = ...
+IcebergTelemetry icebergTelemetry = IcebergTelemetry.create(openTelemetry);
+TableScan tableScan = icebergTelemetry.wrapScan(table.newScan());
+
+try (CloseableIterable<FileScanTask> fileScanTasks = tableScan.planFiles()) {
+  // ...
+}
+```

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

@@ -1,7 +1,9 @@
 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"))
 }

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

@@ -5,8 +5,11 @@
 
 package io.opentelemetry.instrumentation.iceberg.v1_8;
 
+import org.apache.iceberg.Scan;
+import org.apache.iceberg.ScanTask;
+import org.apache.iceberg.ScanTaskGroup;
+
 import io.opentelemetry.api.OpenTelemetry;
-import org.apache.iceberg.TableScan;
 
 public class IcebergTelemetry {
   private final OpenTelemetry openTelemetry;
@@ -19,7 +22,7 @@ public static IcebergTelemetry create(OpenTelemetry openTelemetry) {
     this.openTelemetry = openTelemetry;
   }
 
-  public TableScan wrapTableScan(TableScan tableScan) {
-    return tableScan.metricsReporter(new IcebergMetricsReporter(openTelemetry));
+  public <ThisT, T extends ScanTask, G extends ScanTaskGroup<T>> ThisT wrapScan(Scan<ThisT, T, G> scan) {
+    return scan.metricsReporter(new IcebergMetricsReporter(openTelemetry));
   }
 }

instrumentation/iceberg-1.8/library/src/test/java/io/opentelemetry/instrumentation/iceberg/v1_8/IcebergLibraryTest.java

@@ -0,0 +1,25 @@
+package io.opentelemetry.instrumentation.iceberg.v1_8;
+
+import org.apache.iceberg.TableScan;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.LibraryInstrumentationExtension;
+
+public class IcebergLibraryTest extends AbstractIcebergTest {
+  @RegisterExtension
+  final InstrumentationExtension testing = LibraryInstrumentationExtension.create();
+
+  @Override
+  protected InstrumentationExtension testing() {
+    return this.testing;
+  }
+
+  @Override
+  protected TableScan configure(TableScan tableScan) {
+    OpenTelemetry openTelemetry = testing.getOpenTelemetry();
+    IcebergTelemetry icebergTelemetry = IcebergTelemetry.create(openTelemetry);
+    return icebergTelemetry.wrapScan(tableScan);
+  }
+}

instrumentation/iceberg-1.8/metadata.yaml

@@ -0,0 +1,2 @@
+description: This standalone instrumentation enables metrics for Apache Iceberg scans and commits.
+library_link: https://iceberg.apache.org/

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

@@ -0,0 +1,14 @@
+plugins {
+  id("otel.java-conventions")
+}
+
+dependencies {
+  implementation("org.apache.iceberg:iceberg-core:1.8.1")
+  implementation("org.apache.iceberg:iceberg-core:1.8.1") {
+    artifact {
+      classifier = "tests"
+    }
+  }
+  implementation("org.apache.commons:commons-compress:1.26.2")
+  api(project(":testing-common"))
+}

instrumentation/iceberg-1.8/testing/src/main/java/io/opentelemetry/instrumentation/iceberg/v1_8/AbstractIcebergTest.java

@@ -0,0 +1,93 @@
+package io.opentelemetry.instrumentation.iceberg.v1_8;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.TestTables.TestTable;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.metrics.MetricsReport;
+import org.apache.iceberg.metrics.MetricsReporter;
+import org.apache.iceberg.types.Types.IntegerType;
+import org.apache.iceberg.types.Types.NestedField;
+import org.apache.iceberg.types.Types.StringType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+
+abstract class AbstractIcebergTest {
+  protected abstract InstrumentationExtension testing();
+
+  protected abstract TableScan configure(TableScan tableScan);
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(AbstractIcebergTest.class);
+  protected static final int FORMAT_VERSION = 2;
+  protected static final Schema SCHEMA = new Schema(NestedField.required(3, "id", IntegerType.get()), NestedField.required(4, "data", StringType.get()));
+  protected static final int BUCKETS_NUMBER = 16;
+  protected static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA).bucket("data", 16).build();
+  protected static final DataFile FILE_1 = DataFiles.builder(SPEC).withPath("/path/to/data-a.parquet").withFileSizeInBytes(10L).withPartitionPath("data_bucket=0").withRecordCount(1L).build();
+  protected static final DataFile FILE_2 = DataFiles.builder(SPEC).withPath("/path/to/data-b.parquet").withFileSizeInBytes(10L).withPartitionPath("data_bucket=1").withRecordCount(1L).withSplitOffsets(Arrays.asList(1L)).build();
+
+  @TempDir
+  protected File tableDir = null;
+  protected TestTable table;
+
+  @BeforeEach
+  void init() {
+    this.table = TestTables.create(this.tableDir, "test", SCHEMA, SPEC, FORMAT_VERSION);
+    this.table.newFastAppend().appendFile(FILE_1).appendFile(FILE_2).commit();
+  }
+
+  @Test
+  void testCreateTelemetry() throws IOException {
+    SimpleReporter reporter = new SimpleReporter();
+
+    TableScan scan = table.newScan()
+      .select("id", "data");
+    assertNotNull(scan);
+    assertNull(reporter.report);
+      
+    Schema projection = scan.schema();
+    assertNotNull(projection);
+
+    try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+      assertNotNull(tasks);
+      int counter = 0;
+
+      for (FileScanTask fileTask : tasks) {
+        LOGGER.info(fileTask.file().location());
+        counter++;
+      }
+
+      assertEquals(2, counter);
+      assertNotNull(reporter.report);
+    }
+  }
+
+  static class SimpleReporter implements MetricsReporter {
+    MetricsReport report;
+
+    @Override
+    public void report(MetricsReport report) {
+      LOGGER.error("I am invoked!");
+      this.report = report;
+    }
+
+  }
+
+}

settings.gradle.kts

@@ -304,6 +304,7 @@ 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:iceberg-1.8:testing")
 include(":instrumentation:influxdb-2.4:javaagent")
 include(":instrumentation:internal:internal-application-logger:bootstrap")
 include(":instrumentation:internal:internal-application-logger:javaagent")