Add LogRecordProcessor to record event log records as span events (#1551)

Author: jack-berg

.github/component_owners.yml

@@ -56,6 +56,7 @@ components:
   processors:
     - LikeTheSalad
     - breedx-splk
+    - jack-berg
   prometheus-collector:
     - jkwatson
   resource-providers:

processors/README.md

@@ -1,10 +1,38 @@
 # Processors
 
-This module provides tools to intercept and process signals globally.
+## Interceptable exporters
+
+This module provides tools to intercept and process signals globally:
+
+* `InterceptableSpanExporter`
+* `InterceptableMetricExporter`
+* `InterceptableLogRecordExporter`
+
+## Event to SpanEvent Bridge
+
+`EventToSpanEventBridge` is a `LogRecordProcessor` which records events (i.e. log records with an `event.name` attribute) as span events for the current span if:
+
+* The log record has a valid span context
+* `Span.current()` returns a span where `Span.isRecording()` is true
+
+For details of how the event log record is translated to span event, see [EventToSpanEventBridge Javadoc](./src/main/java/io/opentelemetry/contrib/eventbridge/EventToSpanEventBridge.java).
+
+`EventToSpanEventBridge` can be referenced in [declarative configuration](https://opentelemetry.io/docs/languages/java/configuration/#declarative-configuration) as follows:
+
+```yaml
+# Configure tracer provider as usual, omitted for brevity
+tracer_provider: ...
+
+logger_provider:
+  processors:
+      # TODO(jack-berg): remove "{}" after releasing [opentelemetry-java#6891](https://github.com/open-telemetry/opentelemetry-java/pull/6891/files)
+      - event_to_span_event_bridge: {}
+```
 
 ## Component owners
 
 - [Cesar Munoz](https://github.com/LikeTheSalad), Elastic
+- [Jack Berg](https://github.com/jack-berg), New Relic
 - [Jason Plumb](https://github.com/breedx-splk), Splunk
 
 Learn more about component owners in [component_owners.yml](../.github/component_owners.yml).

processors/build.gradle.kts

@@ -13,5 +13,13 @@ java {
 
 dependencies {
   api("io.opentelemetry:opentelemetry-sdk")
+  compileOnly("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi")
+
+  // For EventToSpanEventBridge
+  implementation("io.opentelemetry:opentelemetry-exporter-otlp-common")
+  implementation("com.fasterxml.jackson.core:jackson-core")
+
   testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
+  testImplementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure")
+  testImplementation("io.opentelemetry:opentelemetry-sdk-extension-incubator")
 }

processors/src/main/java/io/opentelemetry/contrib/eventbridge/EventToSpanEventBridge.java

@@ -0,0 +1,136 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.eventbridge;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.common.Value;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.exporter.internal.marshal.MarshalerWithSize;
+import io.opentelemetry.exporter.internal.otlp.AnyValueMarshaler;
+import io.opentelemetry.sdk.logs.LogRecordProcessor;
+import io.opentelemetry.sdk.logs.ReadWriteLogRecord;
+import io.opentelemetry.sdk.logs.data.LogRecordData;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * A processor that records events (i.e. log records with an {@code event.name} attribute) as span
+ * events for the current span if:
+ *
+ * <ul>
+ *   <li>The log record has a valid span context
+ *   <li>{@link Span#current()} returns a span where {@link Span#isRecording()} is true
+ *   <li>The log record's span context is the same as {@link Span#current()}
+ * </ul>
+ *
+ * <p>The event {@link LogRecordData} is converted to a span event as follows:
+ *
+ * <ul>
+ *   <li>{@code event.name} attribute is mapped to span event name
+ *   <li>{@link LogRecordData#getTimestampEpochNanos()} is mapped to span event timestamp
+ *   <li>{@link LogRecordData#getAttributes()} are mapped to span event attributes, excluding {@code
+ *       event.name}
+ *   <li>{@link LogRecordData#getObservedTimestampEpochNanos()} is mapped to span event attribute
+ *       with key {@code log.record.observed_timestamp}
+ *   <li>{@link LogRecordData#getSeverity()} is mapped to span event attribute with key {@code
+ *       log.record.severity_number}
+ *   <li>{@link LogRecordData#getBodyValue()} is mapped to span event attribute with key {@code
+ *       log.record.body}, as an escaped JSON string following the standard protobuf JSON encoding
+ *   <li>{@link LogRecordData#getTotalAttributeCount()} - {@link
+ *       LogRecordData#getAttributes()}.size() is mapped to span event attribute with key {@code
+ *       log.record.dropped_attributes_count}
+ * </ul>
+ */
+public final class EventToSpanEventBridge implements LogRecordProcessor {
+
+  private static final Logger logger = Logger.getLogger(EventToSpanEventBridge.class.getName());
+
+  private static final AttributeKey<String> EVENT_NAME = AttributeKey.stringKey("event.name");
+  private static final AttributeKey<Long> LOG_RECORD_OBSERVED_TIME_UNIX_NANO =
+      AttributeKey.longKey("log.record.observed_time_unix_nano");
+  private static final AttributeKey<Long> LOG_RECORD_SEVERITY_NUMBER =
+      AttributeKey.longKey("log.record.severity_number");
+  private static final AttributeKey<String> LOG_RECORD_BODY =
+      AttributeKey.stringKey("log.record.body");
+  private static final AttributeKey<Long> LOG_RECORD_DROPPED_ATTRIBUTES_COUNT =
+      AttributeKey.longKey("log.record.dropped_attributes_count");
+
+  private EventToSpanEventBridge() {}
+
+  /** Create an instance. */
+  public static EventToSpanEventBridge create() {
+    return new EventToSpanEventBridge();
+  }
+
+  @Override
+  public void onEmit(Context context, ReadWriteLogRecord logRecord) {
+    LogRecordData logRecordData = logRecord.toLogRecordData();
+    String eventName = logRecordData.getAttributes().get(EVENT_NAME);
+    if (eventName == null) {
+      return;
+    }
+    SpanContext logSpanContext = logRecordData.getSpanContext();
+    if (!logSpanContext.isValid()) {
+      return;
+    }
+    Span currentSpan = Span.current();
+    if (!currentSpan.isRecording()) {
+      return;
+    }
+    if (!currentSpan.getSpanContext().equals(logSpanContext)) {
+      return;
+    }
+    currentSpan.addEvent(
+        eventName,
+        toSpanEventAttributes(logRecordData),
+        logRecordData.getTimestampEpochNanos(),
+        TimeUnit.NANOSECONDS);
+  }
+
+  private static Attributes toSpanEventAttributes(LogRecordData logRecord) {
+    AttributesBuilder builder =
+        logRecord.getAttributes().toBuilder().removeIf(key -> key.equals(EVENT_NAME));
+
+    builder.put(LOG_RECORD_OBSERVED_TIME_UNIX_NANO, logRecord.getObservedTimestampEpochNanos());
+
+    builder.put(LOG_RECORD_SEVERITY_NUMBER, logRecord.getSeverity().getSeverityNumber());
+
+    // Add bridging for logRecord.getSeverityText() if EventBuilder adds severity text setter
+
+    Value<?> body = logRecord.getBodyValue();
+    if (body != null) {
+      MarshalerWithSize marshaler = AnyValueMarshaler.create(body);
+      ByteArrayOutputStream out = new ByteArrayOutputStream();
+      try {
+        marshaler.writeJsonTo(out);
+        builder.put(LOG_RECORD_BODY, out.toString(StandardCharsets.UTF_8.name()));
+      } catch (IOException e) {
+        logger.log(Level.WARNING, "Error converting log record body to JSON", e);
+      }
+    }
+
+    int droppedAttributesCount =
+        logRecord.getTotalAttributeCount() - logRecord.getAttributes().size();
+    if (droppedAttributesCount > 0) {
+      builder.put(LOG_RECORD_DROPPED_ATTRIBUTES_COUNT, droppedAttributesCount);
+    }
+
+    return builder.build();
+  }
+
+  @Override
+  public String toString() {
+    return "EventToSpanEventBridge{}";
+  }
+}

processors/src/main/java/io/opentelemetry/contrib/eventbridge/internal/EventToSpanEventBridgeComponentProvider.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.eventbridge.internal;
+
+import io.opentelemetry.contrib.eventbridge.EventToSpanEventBridge;
+import io.opentelemetry.sdk.autoconfigure.spi.internal.ComponentProvider;
+import io.opentelemetry.sdk.autoconfigure.spi.internal.StructuredConfigProperties;
+import io.opentelemetry.sdk.logs.LogRecordProcessor;
+
+/**
+ * Declarative configuration SPI implementation for {@link EventToSpanEventBridge}.
+ *
+ * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
+ * at any time.
+ */
+public class EventToSpanEventBridgeComponentProvider
+    implements ComponentProvider<LogRecordProcessor> {
+
+  @Override
+  public Class<LogRecordProcessor> getType() {
+    return LogRecordProcessor.class;
+  }
+
+  @Override
+  public String getName() {
+    return "event_to_span_event_bridge";
+  }
+
+  @Override
+  public LogRecordProcessor create(StructuredConfigProperties config) {
+    return EventToSpanEventBridge.create();
+  }
+}

processors/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.internal.ComponentProvider

@@ -0,0 +1 @@
+io.opentelemetry.contrib.eventbridge.internal.EventToSpanEventBridgeComponentProvider