[baggage-processor] Add BaggageLogRecordProcessor

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

@@ -0,0 +1,42 @@
+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,67 @@
+/*
+ * 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

@@ -20,14 +20,19 @@
 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.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,28 +43,35 @@
 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));
+    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(
-        Collections.singletonMap(
-            "otel.java.experimental.span-attributes.copy-from-baggage.include", "key"),
+        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(
@@ -70,9 +82,10 @@ private static void assertCustomizer(
   }
 
   private static OpenTelemetrySdk getOpenTelemetrySdk(
-      Map<String, String> properties, InMemorySpanExporter spanExporter) {
+      Map<String, String> properties, InMemorySpanExporter spanExporter,
+      InMemoryLogRecordExporter logRecordExporter) {
     SpiHelper spiHelper =
-        SpiHelper.create(BaggageSpanProcessorCustomizerTest.class.getClassLoader());
+        SpiHelper.create(BaggageProcessorCustomizerTest.class.getClassLoader());
 
     AutoConfiguredOpenTelemetrySdkBuilder sdkBuilder =
         AutoConfiguredOpenTelemetrySdk.builder()
@@ -88,7 +101,7 @@ private static OpenTelemetrySdk getOpenTelemetrySdk(
                         "otel.metrics.exporter",
                         "none",
                         "otel.logs.exporter",
-                        "none"))
+                        MEMORY_EXPORTER))
             .addPropertiesSupplier(() -> properties);
     AutoConfigureUtil.setComponentLoader(
         sdkBuilder,
@@ -105,6 +118,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 +147,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 +159,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 +172,33 @@ 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");
+      }
+    }
+  }
 }