[baggage-processor] Add BaggageLogRecordProcessor

baggage-processor/src/main/java/io/opentelemetry/contrib/baggage/processor/BaggageLogRecordProcessor.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.baggage.processor;
+
+import io.opentelemetry.api.baggage.Baggage;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.logs.LogRecordProcessor;
+import io.opentelemetry.sdk.logs.ReadWriteLogRecord;
+import java.util.function.Predicate;
+
+/**
+ * This log record processor copies attributes stored in {@link Baggage} into each newly created log
+ * record.
+ */
+public class BaggageLogRecordProcessor implements LogRecordProcessor {
+
+  /**
+   * Creates a new {@link BaggageLogRecordProcessor} that copies all baggage entries into the newly
+   * created log record.
+   */
+  public static BaggageLogRecordProcessor allowAllBaggageKeys() {
+    return new BaggageLogRecordProcessor(baggageKey -> true);
+  }
+
+  private final Predicate<String> baggageKeyPredicate;
+
+  /**
+   * Creates a new {@link BaggageLogRecordProcessor} that copies only baggage entries with keys that
+   * pass the provided filter into the newly created log record.
+   */
+  public BaggageLogRecordProcessor(Predicate<String> baggageKeyPredicate) {
+    this.baggageKeyPredicate = baggageKeyPredicate;
+  }
+
+  @Override
+  public void onEmit(Context context, ReadWriteLogRecord logRecord) {
+    Baggage.fromContext(context)
+        .forEach(
+            (s, baggageEntry) -> {
+              if (baggageKeyPredicate.test(s)) {
+                logRecord.setAttribute(AttributeKey.stringKey(s), baggageEntry.getValue());
+              }
+            });
+  }
+}

baggage-processor/src/main/java/io/opentelemetry/contrib/baggage/processor/BaggageProcessorCustomizer.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.baggage.processor;
+
+import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer;
+import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider;
+import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
+import io.opentelemetry.sdk.logs.SdkLoggerProviderBuilder;
+import io.opentelemetry.sdk.trace.SdkTracerProviderBuilder;
+import java.util.List;
+
+public class BaggageProcessorCustomizer implements AutoConfigurationCustomizerProvider {
+  @Override
+  public void customize(AutoConfigurationCustomizer autoConfigurationCustomizer) {
+    autoConfigurationCustomizer
+        .addTracerProviderCustomizer(
+            (sdkTracerProviderBuilder, config) -> {
+              addSpanProcessor(sdkTracerProviderBuilder, config);
+              return sdkTracerProviderBuilder;
+            })
+        .addLoggerProviderCustomizer(
+            (sdkLoggerProviderBuilder, config) -> {
+              addLogRecordProcessor(sdkLoggerProviderBuilder, config);
+              return sdkLoggerProviderBuilder;
+            });
+  }
+
+  private static void addSpanProcessor(
+      SdkTracerProviderBuilder sdkTracerProviderBuilder, ConfigProperties config) {
+    List<String> keys =
+        config.getList("otel.java.experimental.span-attributes.copy-from-baggage.include");
+
+    if (keys.isEmpty()) {
+      return;
+    }
+
+    sdkTracerProviderBuilder.addSpanProcessor(createBaggageSpanProcessor(keys));
+  }
+
+  static BaggageSpanProcessor createBaggageSpanProcessor(List<String> keys) {
+    if (keys.size() == 1 && keys.get(0).equals("*")) {
+      return BaggageSpanProcessor.allowAllBaggageKeys();
+    }
+    return new BaggageSpanProcessor(keys::contains);
+  }
+
+  private static void addLogRecordProcessor(
+      SdkLoggerProviderBuilder sdkLoggerProviderBuilder, ConfigProperties config) {
+    List<String> keys =
+        config.getList("otel.java.experimental.log-attributes.copy-from-baggage.include");
+
+    if (keys.isEmpty()) {
+      return;
+    }
+
+    sdkLoggerProviderBuilder.addLogRecordProcessor(createBaggageLogRecordProcessor(keys));
+  }
+
+  static BaggageLogRecordProcessor createBaggageLogRecordProcessor(List<String> keys) {
+    if (keys.size() == 1 && keys.get(0).equals("*")) {
+      return BaggageLogRecordProcessor.allowAllBaggageKeys();
+    }
+    return new BaggageLogRecordProcessor(keys::contains);
+  }
+}

baggage-processor/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider

@@ -1 +1 @@
-io.opentelemetry.contrib.baggage.processor.BaggageSpanProcessorCustomizer
+io.opentelemetry.contrib.baggage.processor.BaggageProcessorCustomizer

baggage-processor/src/test/java/io/opentelemetry/contrib/baggage/processor/BaggageSpanProcessorCustomizerTest.java → baggage-processor/src/test/java/io/opentelemetry/contrib/baggage/processor/BaggageProcessorCustomizerTest.java

@@ -5,6 +5,7 @@
 
 package io.opentelemetry.contrib.baggage.processor;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 import static org.mockito.Mockito.verify;
 
@@ -20,14 +21,20 @@
 import io.opentelemetry.sdk.autoconfigure.internal.ComponentLoader;
 import io.opentelemetry.sdk.autoconfigure.internal.SpiHelper;
 import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
+import io.opentelemetry.sdk.autoconfigure.spi.logs.ConfigurableLogRecordExporterProvider;
 import io.opentelemetry.sdk.autoconfigure.spi.traces.ConfigurableSpanExporterProvider;
+import io.opentelemetry.sdk.logs.ReadWriteLogRecord;
+import io.opentelemetry.sdk.logs.data.LogRecordData;
+import io.opentelemetry.sdk.logs.export.LogRecordExporter;
 import io.opentelemetry.sdk.testing.assertj.SpanDataAssert;
 import io.opentelemetry.sdk.testing.assertj.TracesAssert;
+import io.opentelemetry.sdk.testing.exporter.InMemoryLogRecordExporter;
 import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
 import io.opentelemetry.sdk.trace.ReadWriteSpan;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
 import java.time.Duration;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
@@ -38,57 +45,69 @@
 import org.mockito.junit.jupiter.MockitoExtension;
 
 @ExtendWith(MockitoExtension.class)
-class BaggageSpanProcessorCustomizerTest {
+class BaggageProcessorCustomizerTest {
 
   private static final String MEMORY_EXPORTER = "memory";
 
   @Test
   void test_customizer() {
     assertCustomizer(Collections.emptyMap(), span -> span.hasTotalAttributeCount(0));
-    assertCustomizer(
-        Collections.singletonMap(
-            "otel.java.experimental.span-attributes.copy-from-baggage.include", "key"),
-        span -> span.hasAttribute(AttributeKey.stringKey("key"), "value"));
+    Map<String, String> properties = new HashMap<>();
+    properties.put("otel.java.experimental.span-attributes.copy-from-baggage.include", "key");
+    properties.put("otel.java.experimental.log-attributes.copy-from-baggage.include", "key");
+    // TODO try use
+    //  AttributeAssertion attributeAssertion =
+    // OpenTelemetryAssertions.equalTo(AttributeKey.stringKey("key"), "value");
+    assertCustomizer(properties, span -> span.hasAttribute(AttributeKey.stringKey("key"), "value"));
   }
 
   private static void assertCustomizer(
       Map<String, String> properties, Consumer<SpanDataAssert> spanDataAssertConsumer) {
 
     InMemorySpanExporter spanExporter = InMemorySpanExporter.create();
+    InMemoryLogRecordExporter logExporter = InMemoryLogRecordExporter.create();
 
-    OpenTelemetrySdk sdk = getOpenTelemetrySdk(properties, spanExporter);
+    OpenTelemetrySdk sdk = getOpenTelemetrySdk(properties, spanExporter, logExporter);
     try (Scope ignore = Baggage.current().toBuilder().put("key", "value").build().makeCurrent()) {
       sdk.getTracer("test").spanBuilder("test").startSpan().end();
+      sdk.getLogsBridge().get("test").logRecordBuilder().setBody("test").emit();
     }
+    // TODO verify log record attributes
     await()
         .atMost(Duration.ofSeconds(1))
         .untilAsserted(
-            () ->
-                TracesAssert.assertThat(spanExporter.getFinishedSpanItems())
-                    .hasTracesSatisfyingExactly(
-                        trace -> trace.hasSpansSatisfyingExactly(spanDataAssertConsumer)));
+            () -> {
+              TracesAssert.assertThat(spanExporter.getFinishedSpanItems())
+                  .hasTracesSatisfyingExactly(
+                      trace -> trace.hasSpansSatisfyingExactly(spanDataAssertConsumer));
+              List<LogRecordData> finishedLogRecordItems = logExporter.getFinishedLogRecordItems();
+                  assertThat(finishedLogRecordItems).hasSize(1);
+            });
   }
 
   private static OpenTelemetrySdk getOpenTelemetrySdk(
-      Map<String, String> properties, InMemorySpanExporter spanExporter) {
-    SpiHelper spiHelper =
-        SpiHelper.create(BaggageSpanProcessorCustomizerTest.class.getClassLoader());
+      Map<String, String> properties,
+      InMemorySpanExporter spanExporter,
+      InMemoryLogRecordExporter logRecordExporter) {
+    SpiHelper spiHelper = SpiHelper.create(BaggageProcessorCustomizerTest.class.getClassLoader());
 
     AutoConfiguredOpenTelemetrySdkBuilder sdkBuilder =
         AutoConfiguredOpenTelemetrySdk.builder()
             .addPropertiesSupplier(
                 () ->
                     ImmutableMap.of(
-                        // We set the export interval of the spans to 100 ms. The default value is 5
+                        // We set the export interval of the spans to 10 ms. The default value is 5
                         // seconds.
-                        "otel.bsp.schedule.delay",
+                        "otel.bsp.schedule.delay", // span exporter
+                        "10",
+                        "otel.blrp.schedule.delay", // log exporter
                         "10",
                         "otel.traces.exporter",
                         MEMORY_EXPORTER,
                         "otel.metrics.exporter",
                         "none",
                         "otel.logs.exporter",
-                        "none"))
+                        MEMORY_EXPORTER))
             .addPropertiesSupplier(() -> properties);
     AutoConfigureUtil.setComponentLoader(
         sdkBuilder,
@@ -105,6 +124,20 @@ public SpanExporter createExporter(ConfigProperties configProperties) {
                           return spanExporter;
                         }
 
+                        @Override
+                        public String getName() {
+                          return MEMORY_EXPORTER;
+                        }
+                      });
+            } else if (spiClass == ConfigurableLogRecordExporterProvider.class) {
+              return Collections.singletonList(
+                  (T)
+                      new ConfigurableLogRecordExporterProvider() {
+                        @Override
+                        public LogRecordExporter createExporter(ConfigProperties configProperties) {
+                          return logRecordExporter;
+                        }
+
                         @Override
                         public String getName() {
                           return MEMORY_EXPORTER;
@@ -120,7 +153,7 @@ public String getName() {
   @Test
   public void test_baggageSpanProcessor_adds_attributes_to_spans(@Mock ReadWriteSpan span) {
     try (BaggageSpanProcessor processor =
-        BaggageSpanProcessorCustomizer.createProcessor(Collections.singletonList("*"))) {
+        BaggageProcessorCustomizer.createBaggageSpanProcessor(Collections.singletonList("*"))) {
       try (Scope ignore = Baggage.current().toBuilder().put("key", "value").build().makeCurrent()) {
         processor.onStart(Context.current(), span);
         verify(span).setAttribute("key", "value");
@@ -132,7 +165,7 @@ public void test_baggageSpanProcessor_adds_attributes_to_spans(@Mock ReadWriteSp
   public void test_baggageSpanProcessor_adds_attributes_to_spans_when_key_filter_matches(
       @Mock ReadWriteSpan span) {
     try (BaggageSpanProcessor processor =
-        BaggageSpanProcessorCustomizer.createProcessor(Collections.singletonList("key"))) {
+        BaggageProcessorCustomizer.createBaggageSpanProcessor(Collections.singletonList("key"))) {
       try (Scope ignore =
           Baggage.current().toBuilder()
               .put("key", "value")
@@ -145,4 +178,36 @@ public void test_baggageSpanProcessor_adds_attributes_to_spans_when_key_filter_m
       }
     }
   }
+
+  @Test
+  public void test_baggageLogRecordProcessor_adds_attributes_to_logRecord(
+      @Mock ReadWriteLogRecord logRecord) {
+    try (BaggageLogRecordProcessor processor =
+        BaggageProcessorCustomizer.createBaggageLogRecordProcessor(
+            Collections.singletonList("*"))) {
+      try (Scope ignore = Baggage.current().toBuilder().put("key", "value").build().makeCurrent()) {
+        processor.onEmit(Context.current(), logRecord);
+        verify(logRecord).setAttribute(AttributeKey.stringKey("key"), "value");
+      }
+    }
+  }
+
+  @Test
+  public void test_baggageLogRecordProcessor_adds_attributes_to_spans_when_key_filter_matches(
+      @Mock ReadWriteLogRecord logRecord) {
+    try (BaggageLogRecordProcessor processor =
+        BaggageProcessorCustomizer.createBaggageLogRecordProcessor(
+            Collections.singletonList("key"))) {
+      try (Scope ignore =
+          Baggage.current().toBuilder()
+              .put("key", "value")
+              .put("other", "value")
+              .build()
+              .makeCurrent()) {
+        processor.onEmit(Context.current(), logRecord);
+        verify(logRecord).setAttribute(AttributeKey.stringKey("key"), "value");
+        verify(logRecord, Mockito.never()).setAttribute(AttributeKey.stringKey("other"), "value");
+      }
+    }
+  }
 }