Add kafka client metrics to the javaagent instrumentation

instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/build.gradle.kts

@@ -28,6 +28,8 @@ tasks {
   withType<Test>().configureEach {
     usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
 
+    systemProperty("testLatestDeps", findProperty("testLatestDeps") as Boolean)
+
     // TODO run tests both with and without experimental span attributes
     jvmArgs("-Dotel.instrumentation.kafka.experimental-span-attributes=true")
   }

instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaConsumerInstrumentation.java

@@ -7,6 +7,8 @@
 
 import static io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge.currentContext;
 import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaSingletons.consumerReceiveInstrumenter;
+import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaSingletons.enhanceConfig;
+import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
 import static net.bytebuddy.matcher.ElementMatchers.isPublic;
 import static net.bytebuddy.matcher.ElementMatchers.named;
 import static net.bytebuddy.matcher.ElementMatchers.returns;
@@ -21,6 +23,8 @@
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
 import java.time.Duration;
+import java.util.Map;
+import java.util.Properties;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.matcher.ElementMatcher;
@@ -36,6 +40,12 @@ public ElementMatcher<TypeDescription> typeMatcher() {
 
   @Override
   public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        isConstructor().and(takesArgument(0, Map.class)),
+        this.getClass().getName() + "$ConstructorMapAdvice");
+    transformer.applyAdviceToMethod(
+        isConstructor().and(takesArgument(0, Properties.class)),
+        this.getClass().getName() + "$ConstructorPropertiesAdvice");
     transformer.applyAdviceToMethod(
         named("poll")
             .and(isPublic())
@@ -45,6 +55,24 @@ public void transform(TypeTransformer transformer) {
         this.getClass().getName() + "$PollAdvice");
   }
 
+  @SuppressWarnings("unused")
+  public static class ConstructorMapAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void onEnter(@Advice.Argument(0) Map<String, Object> config) {
+      enhanceConfig(config);
+    }
+  }
+
+  @SuppressWarnings("unused")
+  public static class ConstructorPropertiesAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void onEnter(@Advice.Argument(0) Properties config) {
+      enhanceConfig(config);
+    }
+  }
+
   @SuppressWarnings("unused")
   public static class PollAdvice {
     @Advice.OnMethodEnter(suppress = Throwable.class)

instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaProducerInstrumentation.java

@@ -5,7 +5,9 @@
 
 package io.opentelemetry.javaagent.instrumentation.kafkaclients;
 
+import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaSingletons.enhanceConfig;
 import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaSingletons.producerInstrumenter;
+import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
 import static net.bytebuddy.matcher.ElementMatchers.isMethod;
 import static net.bytebuddy.matcher.ElementMatchers.isPublic;
 import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -17,6 +19,8 @@
 import io.opentelemetry.javaagent.bootstrap.Java8BytecodeBridge;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
 import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
+import java.util.Map;
+import java.util.Properties;
 import net.bytebuddy.asm.Advice;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.matcher.ElementMatcher;
@@ -33,17 +37,41 @@ public ElementMatcher<TypeDescription> typeMatcher() {
 
   @Override
   public void transform(TypeTransformer transformer) {
+    transformer.applyAdviceToMethod(
+        isConstructor().and(takesArgument(0, Map.class)),
+        this.getClass().getName() + "$ConstructorMapAdvice");
+    transformer.applyAdviceToMethod(
+        isConstructor().and(takesArgument(0, Properties.class)),
+        this.getClass().getName() + "$ConstructorPropertiesAdvice");
     transformer.applyAdviceToMethod(
         isMethod()
             .and(isPublic())
             .and(named("send"))
             .and(takesArgument(0, named("org.apache.kafka.clients.producer.ProducerRecord")))
             .and(takesArgument(1, named("org.apache.kafka.clients.producer.Callback"))),
-        KafkaProducerInstrumentation.class.getName() + "$ProducerAdvice");
+        KafkaProducerInstrumentation.class.getName() + "$SendAdvice");
+  }
+
+  @SuppressWarnings("unused")
+  public static class ConstructorMapAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void onEnter(@Advice.Argument(0) Map<String, Object> config) {
+      enhanceConfig(config);
+    }
+  }
+
+  @SuppressWarnings("unused")
+  public static class ConstructorPropertiesAdvice {
+
+    @Advice.OnMethodEnter(suppress = Throwable.class)
+    public static void onEnter(@Advice.Argument(0) Properties config) {
+      enhanceConfig(config);
+    }
   }
 
   @SuppressWarnings("unused")
-  public static class ProducerAdvice {
+  public static class SendAdvice {
 
     @Advice.OnMethodEnter(suppress = Throwable.class)
     public static void onEnter(

instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaSingletons.java

@@ -8,8 +8,11 @@
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.kafka.internal.KafkaInstrumenterFactory;
+import io.opentelemetry.instrumentation.kafka.internal.OpenTelemetryMetricsReporter;
 import io.opentelemetry.javaagent.bootstrap.internal.ExperimentalConfig;
 import io.opentelemetry.javaagent.bootstrap.internal.InstrumentationConfig;
+import java.util.Map;
+import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -20,6 +23,9 @@ public final class KafkaSingletons {
   private static final boolean PROPAGATION_ENABLED =
       InstrumentationConfig.get()
           .getBoolean("otel.instrumentation.kafka.client-propagation.enabled", true);
+  private static final boolean METRICS_ENABLED =
+      InstrumentationConfig.get()
+          .getBoolean("otel.instrumentation.kafka.metric-reporter.enabled", true);
 
   private static final Instrumenter<ProducerRecord<?, ?>, Void> PRODUCER_INSTRUMENTER;
   private static final Instrumenter<ConsumerRecords<?, ?>, Void> CONSUMER_RECEIVE_INSTRUMENTER;
@@ -56,5 +62,20 @@ public static boolean isPropagationEnabled() {
     return CONSUMER_PROCESS_INSTRUMENTER;
   }
 
+  public static void enhanceConfig(Map<? super String, Object> config) {
+    if (!METRICS_ENABLED) {
+      return;
+    }
+    config.merge(
+        CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG,
+        OpenTelemetryMetricsReporter.class.getName(),
+        (class1, class2) -> class1 + "," + class2);
+    config.put(
+        OpenTelemetryMetricsReporter.CONFIG_KEY_OPENTELEMETRY_INSTANCE, GlobalOpenTelemetry.get());
+    config.put(
+        OpenTelemetryMetricsReporter.CONFIG_KEY_OPENTELEMETRY_INSTRUMENTATION_NAME,
+        INSTRUMENTATION_NAME);
+  }
+
   private KafkaSingletons() {}
 }

instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/test/java/io/opentelemetry/instrumentation/kafkaclients/OpenTelemetryMetricsReporterTest.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.kafkaclients;
+
+import static java.util.Collections.emptyMap;
+
+import io.opentelemetry.instrumentation.kafka.internal.AbstractOpenTelemetryMetricsReporterTest;
+import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
+import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
+import java.util.Map;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+@EnabledIfSystemProperty(
+    named = "testLatestDeps",
+    matches = "true",
+    disabledReason =
+        "kafka-clients 0.11 emits a significantly different set of metrics; it's probably fine to just test the latest version")
+class OpenTelemetryMetricsReporterTest extends AbstractOpenTelemetryMetricsReporterTest {
+
+  @RegisterExtension
+  static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
+
+  @Override
+  protected InstrumentationExtension testing() {
+    return testing;
+  }
+
+  @Override
+  protected Map<String, ?> additionalConfig() {
+    return emptyMap();
+  }
+}

instrumentation/kafka/kafka-clients/kafka-clients-0.11/testing/build.gradle.kts

@@ -10,4 +10,8 @@ dependencies {
   implementation(project(":instrumentation:kafka:kafka-clients:kafka-clients-common:library"))
 
   implementation("org.testcontainers:kafka")
+  implementation("org.testcontainers:junit-jupiter")
+
+  compileOnly("com.google.auto.value:auto-value-annotations")
+  annotationProcessor("com.google.auto.value:auto-value")
 }