Demonstrate span-event-to-event, event-to-span-event bridging

docs/apidiffs/current_vs_latest/opentelemetry-sdk-logs.txt

@@ -1,2 +1,10 @@
 Comparing source compatibility of opentelemetry-sdk-logs-1.42.0-SNAPSHOT.jar against opentelemetry-sdk-logs-1.41.0.jar
-No changes.
++++  NEW CLASS: PUBLIC(+) io.opentelemetry.sdk.logs.EventToSpanEventBridge  (not serializable)
+	+++  CLASS FILE FORMAT VERSION: 52.0 <- n.a.
+	+++  NEW INTERFACE: io.opentelemetry.sdk.logs.LogRecordProcessor
+	+++  NEW INTERFACE: java.io.Closeable
+	+++  NEW INTERFACE: java.lang.AutoCloseable
+	+++  NEW SUPERCLASS: java.lang.Object
+	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.logs.EventToSpanEventBridge create(io.opentelemetry.api.metrics.MeterProvider)
+	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.logs.EventToSpanEventBridge create()
+	+++  NEW METHOD: PUBLIC(+) void onEmit(io.opentelemetry.context.Context, io.opentelemetry.sdk.logs.ReadWriteLogRecord)

docs/apidiffs/current_vs_latest/opentelemetry-sdk-trace.txt

@@ -1,2 +1,13 @@
 Comparing source compatibility of opentelemetry-sdk-trace-1.42.0-SNAPSHOT.jar against opentelemetry-sdk-trace-1.41.0.jar
-No changes.
++++  NEW CLASS: PUBLIC(+) io.opentelemetry.sdk.trace.SpanEventToEventBridge  (not serializable)
+	+++  CLASS FILE FORMAT VERSION: 52.0 <- n.a.
+	+++  NEW INTERFACE: java.io.Closeable
+	+++  NEW INTERFACE: java.lang.AutoCloseable
+	+++  NEW INTERFACE: io.opentelemetry.sdk.trace.SpanProcessor
+	+++  NEW SUPERCLASS: java.lang.Object
+	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.trace.SpanEventToEventBridge create(io.opentelemetry.api.incubator.events.EventLoggerProvider, io.opentelemetry.api.metrics.MeterProvider)
+	+++  NEW METHOD: PUBLIC(+) STATIC(+) io.opentelemetry.sdk.trace.SpanEventToEventBridge create(io.opentelemetry.api.incubator.events.EventLoggerProvider)
+	+++  NEW METHOD: PUBLIC(+) boolean isEndRequired()
+	+++  NEW METHOD: PUBLIC(+) boolean isStartRequired()
+	+++  NEW METHOD: PUBLIC(+) void onEnd(io.opentelemetry.sdk.trace.ReadableSpan)
+	+++  NEW METHOD: PUBLIC(+) void onStart(io.opentelemetry.context.Context, io.opentelemetry.sdk.trace.ReadWriteSpan)

sdk/logs/src/main/java/io/opentelemetry/sdk/logs/EventToSpanEventBridge.java

@@ -0,0 +1,105 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.logs;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.metrics.LongCounter;
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.logs.data.LogRecordData;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A processor that records events (i.e. log records with an attribute {@code event.name}) to the
+ * current span, if it is valid and {@link Span#isRecording()} is true.
+ */
+public class EventToSpanEventBridge implements LogRecordProcessor {
+
+  private static final AttributeKey<String> EVENT_NAME = AttributeKey.stringKey("event.name");
+  private static final AttributeKey<String> RESULT =
+      AttributeKey.stringKey("event_span_event_bridge.result");
+  private static final Attributes NOT_EVENT = Attributes.of(RESULT, "skipped_not_event");
+  private static final Attributes SPAN_INVALID = Attributes.of(RESULT, "skipped_span_invalid");
+  private static final Attributes SPAN_NOT_RECORDING =
+      Attributes.of(RESULT, "skipped_span_not_recording");
+  private static final Attributes BRIDGED = Attributes.of(RESULT, "success");
+
+  private final LongCounter processedCounter;
+
+  private EventToSpanEventBridge(MeterProvider meterProvider) {
+    this.processedCounter =
+        meterProvider
+            .meterBuilder(EventToSpanEventBridge.class.getName())
+            .build()
+            .counterBuilder("event_span_event_bridge.processed")
+            .build();
+  }
+
+  public static EventToSpanEventBridge create(MeterProvider meterProvider) {
+    return new EventToSpanEventBridge(meterProvider);
+  }
+
+  public static EventToSpanEventBridge create() {
+    return new EventToSpanEventBridge(MeterProvider.noop());
+  }
+
+  @Override
+  public void onEmit(Context context, ReadWriteLogRecord logRecord) {
+    LogRecordData logRecordData = logRecord.toLogRecordData();
+    String eventName = logRecordData.getAttributes().get(EVENT_NAME);
+    if (eventName == null) {
+      processedCounter.add(1, NOT_EVENT);
+      return;
+    }
+    if (!logRecordData.getSpanContext().isValid()) {
+      processedCounter.add(1, SPAN_INVALID);
+      return;
+    }
+    Span currentSpan = Span.current();
+    if (!currentSpan.isRecording()) {
+      processedCounter.add(1, SPAN_NOT_RECORDING);
+      return;
+    }
+    currentSpan.addEvent(
+        eventName,
+        toSpanEventAttributes(logRecordData),
+        logRecordData.getTimestampEpochNanos(),
+        TimeUnit.NANOSECONDS);
+    processedCounter.add(1, BRIDGED);
+  }
+
+  @SuppressWarnings("unchecked")
+  private static Attributes toSpanEventAttributes(LogRecordData logRecordData) {
+    AttributesBuilder builder = Attributes.builder();
+    logRecordData
+        .getAttributes()
+        .forEach(
+            (key, value) -> {
+              if (key.equals(EVENT_NAME)) {
+                return;
+              }
+              putInBuilder(builder, (AttributeKey<? super Object>) key, value);
+            });
+    builder.put("severity", logRecordData.getSeverity().name());
+    if (logRecordData.getSeverityText() != null) {
+      builder.put("severity_text", logRecordData.getSeverityText());
+    }
+    int droppedAttributesCount =
+        logRecordData.getTotalAttributeCount() - logRecordData.getAttributes().size();
+    if (droppedAttributesCount > 0) {
+      builder.put("dropped_attributes_count", droppedAttributesCount);
+    }
+    builder.put("body", logRecordData.getBody().asString());
+    return builder.build();
+  }
+
+  private static <T> void putInBuilder(AttributesBuilder builder, AttributeKey<T> key, T value) {
+    builder.put(key, value);
+  }
+}

sdk/logs/src/test/java/io/opentelemetry/sdk/logs/EventToSpanEventBridgeTest.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.logs;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.incubator.events.EventLogger;
+import io.opentelemetry.api.incubator.events.EventLoggerProvider;
+import io.opentelemetry.api.logs.Severity;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.api.trace.TracerProvider;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.sdk.logs.internal.SdkEventLoggerProvider;
+import io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import org.junit.jupiter.api.Test;
+
+class EventToSpanEventBridgeTest {
+
+  @Test
+  void demo() {
+    InMemorySpanExporter spanExporter = InMemorySpanExporter.create();
+    TracerProvider tracerProvider =
+        SdkTracerProvider.builder()
+            .addSpanProcessor(SimpleSpanProcessor.create(spanExporter))
+            .build();
+
+    EventLoggerProvider eventLoggerProvider =
+        SdkEventLoggerProvider.create(
+            SdkLoggerProvider.builder()
+                .addLogRecordProcessor(EventToSpanEventBridge.create())
+                .build());
+
+    Tracer tracer = tracerProvider.get("tracer");
+    EventLogger eventLogger = eventLoggerProvider.get("event-logger");
+
+    // Emit an event when a span is being recorded. This should be bridged to a span event.
+    Span span = tracer.spanBuilder("span").startSpan();
+    try (Scope unused = span.makeCurrent()) {
+      eventLogger
+          .builder("my.event-name")
+          .setSeverity(Severity.DEBUG)
+          .put("foo", "bar")
+          .put("number", 1)
+          .setAttributes(Attributes.builder().put("color", "red").build())
+          .emit();
+    } finally {
+      span.end();
+    }
+
+    // Emit an event when a span is not being recorded. This should be dropped.
+    eventLogger
+        .builder("my.event-name")
+        .setSeverity(Severity.DEBUG)
+        .put("foo", "baz")
+        .put("number", 2)
+        .setAttributes(Attributes.builder().put("color", "red").build())
+        .emit();
+
+    // Assert that first emitted event was bridged to the span
+    OpenTelemetryAssertions.assertThat(spanExporter.getFinishedSpanItems())
+        .satisfiesExactly(
+            spanData ->
+                OpenTelemetryAssertions.assertThat(spanData)
+                    .hasName("span")
+                    .hasEventsSatisfyingExactly(
+                        spanEvent ->
+                            spanEvent
+                                .hasName("my.event-name")
+                                .hasAttributes(
+                                    Attributes.builder()
+                                        // event body should be bridged to span event field with
+                                        // type AnyValue, but that doesn't exist (yet) so we string
+                                        // encode it.
+                                        .put("body", "[number=1, foo=bar]")
+                                        .put("severity", "DEBUG")
+                                        .put("color", "red")
+                                        .build())));
+  }
+}

sdk/trace/src/main/java/io/opentelemetry/sdk/trace/SpanEventToEventBridge.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.trace;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.incubator.events.EventBuilder;
+import io.opentelemetry.api.incubator.events.EventLogger;
+import io.opentelemetry.api.incubator.events.EventLoggerProvider;
+import io.opentelemetry.api.metrics.LongCounter;
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.trace.data.EventData;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/** A processor that records span events as events. */
+public class SpanEventToEventBridge implements SpanProcessor {
+
+  private final EventLoggerProvider eventLoggerProvider;
+  private final LongCounter processedCounter;
+
+  private SpanEventToEventBridge(
+      EventLoggerProvider eventLoggerProvider, MeterProvider meterProvider) {
+    this.eventLoggerProvider = eventLoggerProvider;
+    this.processedCounter =
+        meterProvider
+            .meterBuilder(SpanEventToEventBridge.class.getName())
+            .build()
+            .counterBuilder("span_event_event_bridge.processed")
+            .build();
+  }
+
+  public static SpanEventToEventBridge create(
+      EventLoggerProvider eventLoggerProvider, MeterProvider meterProvider) {
+    return new SpanEventToEventBridge(eventLoggerProvider, meterProvider);
+  }
+
+  public static SpanEventToEventBridge create(EventLoggerProvider eventLoggerProvider) {
+    return create(eventLoggerProvider, MeterProvider.noop());
+  }
+
+  @Override
+  public void onStart(Context parentContext, ReadWriteSpan span) {}
+
+  @Override
+  public boolean isStartRequired() {
+    return false;
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public void onEnd(ReadableSpan span) {
+    SpanData spanData = span.toSpanData();
+    List<EventData> spanEvents = spanData.getEvents();
+    if (spanEvents.size() == 0) {
+      return;
+    }
+    // TODO: scopeVersion, scopeSchemaUrl
+    EventLogger eventLogger =
+        eventLoggerProvider.get(spanData.getInstrumentationScopeInfo().getName());
+    for (EventData spanEvent : spanEvents) {
+      EventBuilder builder =
+          eventLogger
+              .builder("span-event." + spanEvent.getName())
+              .setTimestamp(spanEvent.getEpochNanos(), TimeUnit.NANOSECONDS);
+      spanEvent
+          .getAttributes()
+          .forEach(
+              (key, value) -> putInBuilder(builder, (AttributeKey<? super Object>) key, value));
+      builder.emit();
+    }
+    processedCounter.add(spanEvents.size());
+  }
+
+  @Override
+  public boolean isEndRequired() {
+    return true;
+  }
+
+  private static <T> void putInBuilder(EventBuilder builder, AttributeKey<T> key, T value) {
+    builder.put(key, value);
+  }
+}