add Logs Kafka Exporter and Metrics Kafka Exporter

dependencyManagement/build.gradle.kts

@@ -33,6 +33,7 @@ val CORE_DEPENDENCIES = listOf(
   "com.google.errorprone:error_prone_core:${errorProneVersion}",
   "io.github.netmikey.logunit:logunit-jul:2.0.0",
   "io.opentelemetry.proto:opentelemetry-proto:1.0.0-alpha",
+  "io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi:1.33.0",
   "io.prometheus:simpleclient:${prometheusVersion}",
   "io.prometheus:simpleclient_common:${prometheusVersion}",
   "io.prometheus:simpleclient_httpserver:${prometheusVersion}",

kafka-exporter/build.gradle.kts

@@ -24,6 +24,7 @@ dependencies {
   implementation("com.google.protobuf:protobuf-java")
 
   testImplementation("io.opentelemetry:opentelemetry-api")
+  testImplementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi")
   testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
   testImplementation("com.google.guava:guava")
   testImplementation("org.testcontainers:junit-jupiter")

kafka-exporter/src/main/java/io/opentelemetry/contrib/kafka/KafkaLogRecordExporter.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.kafka;
+
+import io.opentelemetry.contrib.kafka.impl.KafkaExporter;
+import io.opentelemetry.contrib.kafka.impl.KafkaExporterBuilder;
+import io.opentelemetry.exporter.internal.otlp.logs.LogsRequestMarshaler;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.logs.data.LogRecordData;
+import io.opentelemetry.sdk.logs.export.LogRecordExporter;
+import java.util.Collection;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * A {@link LogRecordExporter} which writes {@linkplain LogRecordData logs} to Kafka in OLTP format.
+ */
+public final class KafkaLogRecordExporter implements LogRecordExporter {
+
+  private static final Logger logger = Logger.getLogger(KafkaLogRecordExporter.class.getName());
+
+  private final AtomicBoolean isShutdown = new AtomicBoolean();
+
+  private final KafkaExporterBuilder<LogsRequestMarshaler> builder;
+
+  private final KafkaExporter<LogsRequestMarshaler> delegate;
+
+  /**
+   * Returns a new {@link KafkaLogRecordExporter} using the default values.
+   *
+   * <p>To load configuration values from environment variables and system properties, use <a
+   * href="https://github.com/open-telemetry/opentelemetry-java/tree/main/sdk-extensions/autoconfigure">opentelemetry-sdk-extension-autoconfigure</a>.
+   *
+   * @return a new {@link KafkaLogRecordExporter} instance.
+   */
+  public static KafkaLogRecordExporter getDefault() {
+    return builder().build();
+  }
+
+  /**
+   * Returns a new builder instance for this exporter.
+   *
+   * @return a new builder instance for this exporter.
+   */
+  public static KafkaLogRecordExporterBuilder builder() {
+    return new KafkaLogRecordExporterBuilder();
+  }
+
+  KafkaLogRecordExporter(
+      KafkaExporterBuilder<LogsRequestMarshaler> builder,
+      KafkaExporter<LogsRequestMarshaler> delegate) {
+    this.builder = builder;
+    this.delegate = delegate;
+  }
+
+  public KafkaLogRecordExporterBuilder toBuilder() {
+    return new KafkaLogRecordExporterBuilder(builder.copy());
+  }
+
+  @Override
+  public CompletableResultCode export(Collection<LogRecordData> logs) {
+    if (isShutdown.get()) {
+      return CompletableResultCode.ofFailure();
+    }
+    try {
+      LogsRequestMarshaler exportRequest = LogsRequestMarshaler.create(logs);
+      return delegate.export(exportRequest, logs.size());
+    } catch (RuntimeException e) {
+      logger.log(Level.WARNING, "send log records to kafka failed.", e);
+      return CompletableResultCode.ofFailure();
+    }
+  }
+
+  @Override
+  public CompletableResultCode flush() {
+    // TODO
+    return CompletableResultCode.ofSuccess();
+  }
+
+  @Override
+  public CompletableResultCode shutdown() {
+    if (!isShutdown.compareAndSet(false, true)) {
+      logger.log(Level.INFO, "Calling shutdown() multiple times.");
+    }
+    return CompletableResultCode.ofSuccess();
+  }
+
+  @Override
+  public String toString() {
+    return "KafkaLogRecordExporter{" + builder.toString(false) + "}";
+  }
+}

kafka-exporter/src/main/java/io/opentelemetry/contrib/kafka/KafkaLogRecordExporterBuilder.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.kafka;
+
+import static io.opentelemetry.api.internal.Utils.checkArgument;
+import static java.util.Objects.requireNonNull;
+
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.contrib.kafka.impl.KafkaExporterBuilder;
+import io.opentelemetry.exporter.internal.otlp.logs.LogsRequestMarshaler;
+import io.opentelemetry.sdk.common.export.RetryPolicy;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+/**
+ * Builder utility for {@link KafkaLogRecordExporter}.
+ *
+ * @since 1.27.0
+ */
+public final class KafkaLogRecordExporterBuilder {
+
+  private final KafkaExporterBuilder<LogsRequestMarshaler> delegate;
+
+  KafkaLogRecordExporterBuilder(KafkaExporterBuilder<LogsRequestMarshaler> delegate) {
+    this.delegate = delegate;
+  }
+
+  KafkaLogRecordExporterBuilder() {
+    this(new KafkaExporterBuilder<>("kafka", "log"));
+  }
+
+  @CanIgnoreReturnValue
+  public KafkaLogRecordExporterBuilder setProtocolVersion(String protocolVersion) {
+    requireNonNull(protocolVersion, "protocolVersion");
+    delegate.setProtocolVersion(protocolVersion);
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public KafkaLogRecordExporterBuilder setBrokers(String brokers) {
+    requireNonNull(brokers, "brokers");
+    delegate.setBrokers(brokers);
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public KafkaLogRecordExporterBuilder setTopic(String topic) {
+    requireNonNull(topic, "topic");
+    delegate.setTopic(topic);
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public KafkaLogRecordExporterBuilder setTimeout(long timeout, TimeUnit unit) {
+    requireNonNull(unit, "unit");
+    checkArgument(timeout >= 0, "timeout must be non-negative");
+    delegate.setTimeout(timeout, unit);
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public KafkaLogRecordExporterBuilder setTimeout(Duration timeout) {
+    requireNonNull(timeout, "timeout");
+    return setTimeout(timeout.toNanos(), TimeUnit.NANOSECONDS);
+  }
+
+  @CanIgnoreReturnValue
+  public KafkaLogRecordExporterBuilder setRetryPolicy(RetryPolicy retryPolicy) {
+    requireNonNull(retryPolicy, "retryPolicy");
+    delegate.setRetryPolicy(retryPolicy);
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public KafkaLogRecordExporterBuilder setMeterProvider(MeterProvider meterProvider) {
+    requireNonNull(meterProvider, "meterProvider");
+    setMeterProvider(() -> meterProvider);
+    return this;
+  }
+
+  @CanIgnoreReturnValue
+  public KafkaLogRecordExporterBuilder setMeterProvider(
+      Supplier<MeterProvider> meterProviderSupplier) {
+    requireNonNull(meterProviderSupplier, "meterProviderSupplier");
+    delegate.setMeterProvider(meterProviderSupplier);
+    return this;
+  }
+
+  public KafkaLogRecordExporter build() {
+    return new KafkaLogRecordExporter(delegate, delegate.build());
+  }
+}

kafka-exporter/src/main/java/io/opentelemetry/contrib/kafka/KafkaMetricExporter.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.kafka;
+
+import io.opentelemetry.contrib.kafka.impl.KafkaExporter;
+import io.opentelemetry.contrib.kafka.impl.KafkaExporterBuilder;
+import io.opentelemetry.exporter.internal.otlp.metrics.MetricsRequestMarshaler;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.metrics.Aggregation;
+import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector;
+import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
+import io.opentelemetry.sdk.metrics.export.MetricExporter;
+import java.util.Collection;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/** A {@link MetricExporter} which writes {@linkplain MetricData spans} to Kafka in OTLP format. */
+public final class KafkaMetricExporter implements MetricExporter {
+
+  private static final Logger logger = Logger.getLogger(KafkaMetricExporter.class.getName());
+
+  private final AtomicBoolean isShutdown = new AtomicBoolean();
+
+  private final KafkaExporterBuilder<MetricsRequestMarshaler> builder;
+
+  private final KafkaExporter<MetricsRequestMarshaler> delegate;
+
+  private final AggregationTemporalitySelector aggregationTemporalitySelector;
+
+  private final DefaultAggregationSelector defaultAggregationSelector;
+
+  /**
+   * Returns a new {@link KafkaMetricExporter} using the default values.
+   *
+   * <p>To load configuration values from environment variables and system properties, use <a
+   * href="https://github.com/open-telemetry/opentelemetry-java/tree/main/sdk-extensions/autoconfigure">opentelemetry-sdk-extension-autoconfigure</a>.
+   *
+   * @return a new {@link KafkaMetricExporter} instance.
+   */
+  public static KafkaMetricExporter getDefault() {
+    return builder().build();
+  }
+
+  /**
+   * Returns a new builder instance for this exporter.
+   *
+   * @return a new builder instance for this exporter.
+   */
+  public static KafkaMetricExporterBuilder builder() {
+    return new KafkaMetricExporterBuilder();
+  }
+
+  KafkaMetricExporter(
+      KafkaExporterBuilder<MetricsRequestMarshaler> builder,
+      KafkaExporter<MetricsRequestMarshaler> delegate,
+      AggregationTemporalitySelector aggregationTemporalitySelector,
+      DefaultAggregationSelector defaultAggregationSelector) {
+    this.builder = builder;
+    this.delegate = delegate;
+    this.aggregationTemporalitySelector = aggregationTemporalitySelector;
+    this.defaultAggregationSelector = defaultAggregationSelector;
+  }
+
+  public KafkaMetricExporterBuilder toBuilder() {
+    return new KafkaMetricExporterBuilder(builder.copy());
+  }
+
+  @Override
+  public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) {
+    return aggregationTemporalitySelector.getAggregationTemporality(instrumentType);
+  }
+
+  @Override
+  public Aggregation getDefaultAggregation(InstrumentType instrumentType) {
+    return defaultAggregationSelector.getDefaultAggregation(instrumentType);
+  }
+
+  @Override
+  public CompletableResultCode export(Collection<MetricData> metrics) {
+    if (isShutdown.get()) {
+      return CompletableResultCode.ofFailure();
+    }
+    try {
+      MetricsRequestMarshaler exportRequest = MetricsRequestMarshaler.create(metrics);
+      return delegate.export(exportRequest, metrics.size());
+    } catch (RuntimeException e) {
+      logger.log(Level.WARNING, "send metrics to kafka failed.", e);
+      return CompletableResultCode.ofFailure();
+    }
+  }
+
+  @Override
+  public CompletableResultCode flush() {
+    // TODO
+    return CompletableResultCode.ofSuccess();
+  }
+
+  @Override
+  public CompletableResultCode shutdown() {
+    if (!isShutdown.compareAndSet(false, true)) {
+      logger.log(Level.INFO, "Calling shutdown() multiple times.");
+    }
+    return CompletableResultCode.ofSuccess();
+  }
+
+  @Override
+  public String toString() {
+    return "KafkaMetricExporter{" + builder.toString(false) + "}";
+  }
+}