Add kafka client metrics

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

@@ -10,4 +10,7 @@ dependencies {
   implementation(project(":instrumentation:kafka:kafka-clients:kafka-clients-common:library"))
 
   implementation("org.testcontainers:kafka")
+  implementation("org.testcontainers:junit-jupiter")
+
+  runtimeOnly("org.apache.kafka:kafka_2.13:2.8.1")
 }

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/KafkaMetricId.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.kafkaclients;
+
+import com.google.auto.value.AutoValue;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.kafka.common.metrics.KafkaMetric;
+import org.apache.kafka.common.metrics.Measurable;
+
+/** A value class collecting the identifying fields of a kafka {@link KafkaMetric}. */
+@AutoValue
+abstract class KafkaMetricId {
+
+  abstract String getGroup();
+
+  abstract String getName();
+
+  abstract String getDescription();
+
+  @Nullable
+  abstract Class<? extends Measurable> getMeasureable();
+
+  abstract Set<String> getAttributeKeys();
+
+  static KafkaMetricId create(KafkaMetric kafkaMetric) {
+    Class<? extends Measurable> measureable;
+    try {
+      measureable = kafkaMetric.measurable().getClass();
+    } catch (IllegalStateException e) {
+      measureable = null;
+    }
+    return new AutoValue_KafkaMetricId(
+        kafkaMetric.metricName().group(),
+        kafkaMetric.metricName().name(),
+        kafkaMetric.metricName().description(),
+        measureable,
+        kafkaMetric.metricName().tags().keySet());
+  }
+}

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

@@ -0,0 +1,133 @@
+/*
+ * 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.metrics.KafkaMetric;
+
+/** 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, KafkaMetricId kafkaMetricId, KafkaMetric kafkaMetric) {
+    // If metric is not a Measureable, we can't map it to an instrument
+    if (kafkaMetricId.getMeasureable() == null) {
+      return null;
+    }
+    Optional<String> matchingGroup =
+        groups.stream().filter(group -> kafkaMetricId.getGroup().contains(group)).findFirst();
+    // Only map metrics that have a matching group
+    if (!matchingGroup.isPresent()) {
+      return null;
+    }
+    String instrumentName =
+        "kafka." + matchingGroup.get() + "." + kafkaMetricId.getName().replace("-", "_");
+    String instrumentDescription =
+        descriptionCache.computeIfAbsent(instrumentName, s -> kafkaMetricId.getDescription());
+    String instrumentType =
+        measureableToInstrumentType.getOrDefault(
+            kafkaMetricId.getMeasureable(), INSTRUMENT_TYPE_DOUBLE_OBSERVABLE_GAUGE);
+
+    InstrumentDescriptor instrumentDescriptor =
+        toInstrumentDescriptor(instrumentType, instrumentName, instrumentDescription);
+    Attributes attributes = toAttributes(kafkaMetric);
+    AutoCloseable observable =
+        createObservable(meter, attributes, instrumentDescriptor, kafkaMetric);
+    return RegisteredObservable.create(kafkaMetricId, instrumentDescriptor, attributes, observable);
+  }
+
+  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.");
+  }
+
+  static Attributes toAttributes(KafkaMetric kafkaMetric) {
+    AttributesBuilder attributesBuilder = Attributes.builder();
+    kafkaMetric.metricName().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,150 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.kafkaclients;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.Meter;
+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 javax.annotation.Nullable;
+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 OpenTelemetry instance via {@link #setOpenTelemetry(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());
+  @Nullable private static volatile Meter meter;
+
+  private static final List<RegisteredObservable> registeredObservables = new ArrayList<>();
+
+  /**
+   * Setup OpenTelemetry. This should be called as early in the application lifecycle as possible.
+   * Kafka metrics that are observed before this is called may not be bridged.
+   */
+  public static void setOpenTelemetry(OpenTelemetry openTelemetry) {
+    meter = openTelemetry.getMeter("io.opentelemetry.kafka-clients");
+  }
+
+  /**
+   * 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() {
+    return registeredObservables;
+  }
+
+  @Override
+  public void init(List<KafkaMetric> metrics) {
+    metrics.forEach(this::metricChange);
+  }
+
+  @Override
+  public synchronized void metricChange(KafkaMetric metric) {
+    KafkaMetricId kafkaMetricId = KafkaMetricId.create(metric);
+    Meter currentMeter = meter;
+    if (currentMeter == null) {
+      logger.log(Level.FINEST, "Metric changed but meter not set: {0}", kafkaMetricId);
+      return;
+    }
+
+    RegisteredObservable registeredObservable =
+        KafkaMetricRegistry.getRegisteredObservable(currentMeter, kafkaMetricId, metric);
+    if (registeredObservable == null) {
+      logger.log(Level.FINEST, "Metric changed but cannot map to instrument: {0}", kafkaMetricId);
+      return;
+    }
+
+    InstrumentDescriptor instrumentDescriptor = registeredObservable.getInstrumentDescriptor();
+    Attributes attributes = registeredObservable.getAttributes();
+    Set<AttributeKey<?>> attributeKeys = attributes.asMap().keySet();
+
+    for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
+      RegisteredObservable curRegisteredObservable = it.next();
+      Set<AttributeKey<?>> curAttributeKeys =
+          curRegisteredObservable.getAttributes().asMap().keySet();
+      if (curRegisteredObservable.equals(registeredObservable)) {
+        logger.log(Level.FINEST, "Replacing instrument: {0}", curRegisteredObservable);
+        closeInstrument(curRegisteredObservable.getObservable());
+        it.remove();
+      } else if (curRegisteredObservable.getInstrumentDescriptor().equals(instrumentDescriptor)
+          && 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) {
+    KafkaMetricId kafkaMetricId = KafkaMetricId.create(metric);
+    logger.log(Level.FINEST, "Metric removed: {0}", kafkaMetricId);
+    for (Iterator<RegisteredObservable> it = registeredObservables.iterator(); it.hasNext(); ) {
+      RegisteredObservable current = it.next();
+      if (current.matches(metric)) {
+        closeInstrument(current.getObservable());
+        it.remove();
+      }
+    }
+  }
+
+  @Override
+  public void close() {
+    closeAllInstruments();
+  }
+
+  static void closeAllInstruments() {
+    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) {}
+}