Add kafka client metrics

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")
 }

instrumentation/kafka/kafka-clients/kafka-clients-0.11/testing/src/main/groovy/io/opentelemetry/instrumentation/kafkaclients/KafkaClientBaseTest.groovy

@@ -6,6 +6,8 @@
 package io.opentelemetry.instrumentation.kafkaclients
 
 import io.opentelemetry.instrumentation.test.InstrumentationSpecification
+import org.testcontainers.utility.DockerImageName
+
 import java.time.Duration
 import org.apache.kafka.clients.admin.AdminClient
 import org.apache.kafka.clients.admin.NewTopic
@@ -46,7 +48,7 @@ abstract class KafkaClientBaseTest extends InstrumentationSpecification {
   static TopicPartition topicPartition = new TopicPartition(SHARED_TOPIC, 0)
 
   def setupSpec() {
-    kafka = new KafkaContainer()
+    kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"))
       .withLogConsumer(new Slf4jLogConsumer(logger))
       .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
       .withStartupTimeout(Duration.ofMinutes(1))

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/test/groovy/io/opentelemetry/instrumentation/kafkaclients/MetricsTest.java

@@ -0,0 +1,8 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.kafkaclients;
+
+public class MetricsTest extends OpenTelemetryKafkaMetricsTest {}

instrumentation/kafka/kafka-clients/kafka-clients-common/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/InstrumentDescriptor.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.kafkaclients;
+
+import com.google.auto.value.AutoValue;
+
+/** A description of an OpenTelemetry metric instrument. */
+@AutoValue
+abstract class InstrumentDescriptor {
+
+  static final String INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE = "DOUBLE_OBSERVABLE_GAUGE";
+  static final String INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_COUNTER = "DOUBLE_OBSERVABLE_COUNTER";
+
+  abstract String getName();
+
+  abstract String getDescription();
+
+  abstract String getInstrumentType();
+
+  static InstrumentDescriptor createDoubleGauge(String name, String description) {
+    return new AutoValue_InstrumentDescriptor(
+        name, description, INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE);
+  }
+
+  static InstrumentDescriptor createDoubleCounter(String name, String description) {
+    return new AutoValue_InstrumentDescriptor(
+        name, description, INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_COUNTER);
+  }
+}

instrumentation/kafka/kafka-clients/kafka-clients-common/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/KafkaMetricRegistry.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.kafkaclients;
+
+import static io.opentelemetry.instrumentation.kafkaclients.InstrumentDescriptor.INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_COUNTER;
+import static io.opentelemetry.instrumentation.kafkaclients.InstrumentDescriptor.INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.api.metrics.ObservableDoubleMeasurement;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+import javax.annotation.Nullable;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.KafkaMetric;
+import org.apache.kafka.common.metrics.Measurable;
+
+/** A registry mapping kafka metrics to corresponding OpenTelemetry metric definitions. */
+class KafkaMetricRegistry {
+
+  private static final Set<String> groups = new HashSet<>(Arrays.asList("consumer", "producer"));
+  private static final Map<Class<?>, String> measureableToInstrumentType = new HashMap<>();
+  private static final Map<String, String> descriptionCache = new ConcurrentHashMap<>();
+
+  static {
+    Map<String, String> classNameToType = new HashMap<>();
+    classNameToType.put(
+        "org.apache.kafka.common.metrics.stats.Rate", INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE);
+    classNameToType.put(
+        "org.apache.kafka.common.metrics.stats.Avg", INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE);
+    classNameToType.put(
+        "org.apache.kafka.common.metrics.stats.Max", INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE);
+    classNameToType.put(
+        "org.apache.kafka.common.metrics.stats.Value", INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE);
+    classNameToType.put(
+        "org.apache.kafka.common.metrics.stats.CumulativeSum",
+        INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_COUNTER);
+    classNameToType.put(
+        "org.apache.kafka.common.metrics.stats.CumulativeCount",
+        INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_COUNTER);
+
+    for (Map.Entry<String, String> entry : classNameToType.entrySet()) {
+      try {
+        measureableToInstrumentType.put(Class.forName(entry.getKey()), entry.getValue());
+      } catch (ClassNotFoundException e) {
+        // Class doesn't exist in this version of kafka client - skip
+      }
+    }
+  }
+
+  @Nullable
+  static RegisteredObservable getRegisteredObservable(Meter meter, KafkaMetric kafkaMetric) {
+    // If metric is not a Measureable, we can't map it to an instrument
+    Class<? extends Measurable> measurable = getMeasurable(kafkaMetric);
+    if (measurable == null) {
+      return null;
+    }
+    MetricName metricName = kafkaMetric.metricName();
+    Optional<String> matchingGroup =
+        groups.stream().filter(group -> metricName.group().contains(group)).findFirst();
+    // Only map metrics that have a matching group
+    if (!matchingGroup.isPresent()) {
+      return null;
+    }
+    String instrumentName =
+        "kafka." + matchingGroup.get() + "." + metricName.name().replace("-", "_");
+    String instrumentDescription =
+        descriptionCache.computeIfAbsent(instrumentName, s -> metricName.description());
+    String instrumentType =
+        measureableToInstrumentType.getOrDefault(
+            measurable, INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE);
+
+    InstrumentDescriptor instrumentDescriptor =
+        toInstrumentDescriptor(instrumentType, instrumentName, instrumentDescription);
+    Attributes attributes = toAttributes(metricName.tags());
+    AutoCloseable observable =
+        createObservable(meter, attributes, instrumentDescriptor, kafkaMetric);
+    return RegisteredObservable.create(metricName, instrumentDescriptor, attributes, observable);
+  }
+
+  @Nullable
+  private static Class<? extends Measurable> getMeasurable(KafkaMetric kafkaMetric) {
+    try {
+      return kafkaMetric.measurable().getClass();
+    } catch (IllegalStateException e) {
+      return null;
+    }
+  }
+
+  private static InstrumentDescriptor toInstrumentDescriptor(
+      String instrumentType, String instrumentName, String instrumentDescription) {
+    switch (instrumentType) {
+      case INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE:
+        return InstrumentDescriptor.createDoubleGauge(instrumentName, instrumentDescription);
+      case INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_COUNTER:
+        return InstrumentDescriptor.createDoubleCounter(instrumentName, instrumentDescription);
+      default: // Continue below to throw
+    }
+    throw new IllegalStateException("Unrecognized instrument type. This is a bug.");
+  }
+
+  private static Attributes toAttributes(Map<String, String> tags) {
+    AttributesBuilder attributesBuilder = Attributes.builder();
+    tags.forEach(attributesBuilder::put);
+    return attributesBuilder.build();
+  }
+
+  private static AutoCloseable createObservable(
+      Meter meter,
+      Attributes attributes,
+      InstrumentDescriptor instrumentDescriptor,
+      KafkaMetric kafkaMetric) {
+    Consumer<ObservableDoubleMeasurement> callback =
+        observableMeasurement -> observableMeasurement.record(kafkaMetric.value(), attributes);
+    switch (instrumentDescriptor.getInstrumentType()) {
+      case INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE:
+        return meter
+            .gaugeBuilder(instrumentDescriptor.getName())
+            .setDescription(instrumentDescriptor.getDescription())
+            .buildWithCallback(callback);
+      case INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_COUNTER:
+        return meter
+            .counterBuilder(instrumentDescriptor.getName())
+            .setDescription(instrumentDescriptor.getDescription())
+            .ofDoubles()
+            .buildWithCallback(callback);
+      default: // Continue below to throw
+    }
+    // TODO: add support for other instrument types and value types as needed for new instruments.
+    // This should not happen.
+    throw new IllegalStateException("Unrecognized instrument type. This is a bug.");
+  }
+
+  private KafkaMetricRegistry() {}
+}

instrumentation/kafka/kafka-clients/kafka-clients-common/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/OpenTelemetryKafkaMetrics.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.kafkaclients;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.instrumentation.api.internal.GuardedBy;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.apache.kafka.common.metrics.KafkaMetric;
+import org.apache.kafka.common.metrics.MetricsReporter;
+
+/**
+ * A {@link MetricsReporter} which bridges Kafka metrics to OpenTelemetry metrics.
+ *
+ * <p>To use, configure {@link GlobalOpenTelemetry} instance via {@link
+ * GlobalOpenTelemetry#set(OpenTelemetry)}, and include a reference to this class in kafka producer
+ * or consumer configuration, i.e.:
+ *
+ * <pre>{@code
+ * //    Map<String, Object> config = new HashMap<>();
+ * //    // Register OpenTelemetryKafkaMetrics as reporter
+ * //    config.put(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, OpenTelemetryKafkaMetrics.class.getName());
+ * //    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
+ * //    ...
+ * //    try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(config)) { ... }
+ * }</pre>
+ */
+public class OpenTelemetryKafkaMetrics implements MetricsReporter {
+
+  private static final Logger logger = Logger.getLogger(OpenTelemetryKafkaMetrics.class.getName());
+  private static volatile Meter meter =
+      GlobalOpenTelemetry.getMeter("io.opentelemetry.kafka-clients");
+
+  private static final Object lock = new Object();
+
+  @GuardedBy("lock")
+  private static final List<RegisteredObservable> registeredObservables = new ArrayList<>();
+
+  /**
+   * Reset for test by reseting the {@link #meter} to {@code null} and closing all registered
+   * instruments.
+   */
+  static void resetForTest() {
+    meter = null;
+    closeAllInstruments();
+  }
+
+  // Visible for test
+  static List<RegisteredObservable> getRegisteredObservables() {
+    synchronized (lock) {
+      return new ArrayList<>(registeredObservables);
+    }
+  }
+
+  @Override
+  public void init(List<KafkaMetric> metrics) {
+    metrics.forEach(this::metricChange);
+  }
+
+  @Override
+  public void metricChange(KafkaMetric metric) {
+    RegisteredObservable registeredObservable =
+        KafkaMetricRegistry.getRegisteredObservable(meter, metric);
+    if (registeredObservable == null) {
+      logger.log(
+          Level.FINEST, "Metric changed but cannot map to instrument: {0}", metric.metricName());
+      return;
+    }
+
+    Set<AttributeKey<?>> attributeKeys = registeredObservable.getAttributes().asMap().keySet();
+    synchronized (lock) {
+      for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
+        RegisteredObservable curRegisteredObservable = it.next();
+        Set<AttributeKey<?>> curAttributeKeys =
+            curRegisteredObservable.getAttributes().asMap().keySet();
+        if (curRegisteredObservable.getKafkaMetricName().equals(metric.metricName())) {
+          logger.log(Level.FINEST, "Replacing instrument: {0}", curRegisteredObservable);
+          closeInstrument(curRegisteredObservable.getObservable());
+          it.remove();
+        } else if (curRegisteredObservable
+                .getInstrumentDescriptor()
+                .equals(registeredObservable.getInstrumentDescriptor())
+            && attributeKeys.size() > curAttributeKeys.size()
+            && attributeKeys.containsAll(curAttributeKeys)) {
+          logger.log(
+              Level.FINEST,
+              "Replacing instrument with higher dimension version: {0}",
+              curRegisteredObservable);
+          closeInstrument(curRegisteredObservable.getObservable());
+          it.remove();
+        }
+      }
+
+      registeredObservables.add(registeredObservable);
+    }
+  }
+
+  @Override
+  public void metricRemoval(KafkaMetric metric) {
+    logger.log(Level.FINEST, "Metric removed: {0}", metric.metricName());
+    synchronized (lock) {
+      for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
+        RegisteredObservable current = it.next();
+        if (current.getKafkaMetricName().equals(metric.metricName())) {
+          closeInstrument(current.getObservable());
+          it.remove();
+        }
+      }
+    }
+  }
+
+  @Override
+  public void close() {
+    closeAllInstruments();
+  }
+
+  static void closeAllInstruments() {
+    synchronized (lock) {
+      for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
+        closeInstrument(it.next().getObservable());
+        it.remove();
+      }
+    }
+  }
+
+  private static void closeInstrument(AutoCloseable observable) {
+    try {
+      observable.close();
+    } catch (Exception e) {
+      throw new IllegalStateException("Error occurred closing instrument", e);
+    }
+  }
+
+  @Override
+  public void configure(Map<String, ?> configs) {}
+}

instrumentation/kafka/kafka-clients/kafka-clients-common/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/RegisteredObservable.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.kafkaclients;
+
+import com.google.auto.value.AutoValue;
+import io.opentelemetry.api.common.Attributes;
+import org.apache.kafka.common.MetricName;
+
+@AutoValue
+abstract class RegisteredObservable {
+
+  abstract MetricName getKafkaMetricName();
+
+  abstract InstrumentDescriptor getInstrumentDescriptor();
+
+  abstract Attributes getAttributes();
+
+  abstract AutoCloseable getObservable();
+
+  static RegisteredObservable create(
+      MetricName metricName,
+      InstrumentDescriptor instrumentDescriptor,
+      Attributes attributes,
+      AutoCloseable observable) {
+    return new AutoValue_RegisteredObservable(
+        metricName, instrumentDescriptor, attributes, observable);
+  }
+}