Ability to instrument logs before OTel injection into OTel appenders

instrumentation/log4j/log4j-appender-2.17/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/log4j/appender/v2_17/Log4jHelper.java

@@ -27,10 +27,12 @@ public final class Log4jHelper {
 
   private static final LogEventMapper<Map<String, String>> mapper;
 
+  private static final boolean captureExperimentalAttributes;
+
   static {
     InstrumentationConfig config = InstrumentationConfig.get();
 
-    boolean captureExperimentalAttributes =
+    captureExperimentalAttributes =
         config.getBoolean("otel.instrumentation.log4j-appender.experimental-log-attributes", false);
     boolean captureMapMessageAttributes =
         config.getBoolean(
@@ -66,7 +68,21 @@ public static void capture(
             .build()
             .logRecordBuilder();
     Map<String, String> contextData = ThreadContext.getImmutableContext();
-    mapper.mapLogEvent(builder, message, level, marker, throwable, contextData);
+
+    if (captureExperimentalAttributes) {
+      Thread currentThread = Thread.currentThread();
+      mapper.mapLogEvent(
+          builder,
+          message,
+          level,
+          marker,
+          throwable,
+          contextData,
+          currentThread.getName(),
+          currentThread.getId());
+    } else {
+      mapper.mapLogEvent(builder, message, level, marker, throwable, contextData, null, -1);
+    }
     builder.setTimestamp(Instant.now());
     builder.emit();
   }

instrumentation/log4j/log4j-appender-2.17/library/README.md

@@ -93,11 +93,12 @@ Setting can be configured as XML attributes, for example:
 
 The available settings are:
 
-| XML Attribute                   | Type    | Default | Description                                                                                                           |
-| ------------------------------- | ------- | ------- | --------------------------------------------------------------------------------------------------------------------- |
-| `captureExperimentalAttributes` | Boolean | `false` | Enable the capture of experimental span attributes `thread.name` and `thread.id`.                                     |
-| `captureMapMessageAttributes`   | Boolean | `false` | Enable the capture of `MapMessage` attributes.                                                                        |
-| `captureMarkerAttribute;`       | Boolean | `false` | Enable the capture of Log4j markers as attributes.                                                                    |
-| `captureContextDataAttributes`  | String  |         | Comma separated list of context data attributes to capture. Use the wildcard character `*` to capture all attributes. |
+| XML Attribute                   | Type    | Default | Description                                                                                                                                                                                                |
+|---------------------------------|---------|---------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `captureExperimentalAttributes` | Boolean | `false` | Enable the capture of experimental span attributes `thread.name` and `thread.id`.                                                                                                                          |
+| `captureMapMessageAttributes`   | Boolean | `false` | Enable the capture of `MapMessage` attributes.                                                                                                                                                             |
+| `captureMarkerAttribute;`       | Boolean | `false` | Enable the capture of Log4j markers as attributes.                                                                                                                                                         |
+| `captureContextDataAttributes`  | String  |         | Comma separated list of context data attributes to capture. Use the wildcard character `*` to capture all attributes.                                                                                      |
+| `firstLogsCacheSize`            | Integer | 1000    | Log telemetry is emitted after the initialization of the OpenTelemetry Log4j appender with an OpenTelemetry object. This setting allows you to modify the size of the cache used to replay the first logs. |
 
 [source code attributes]: https://github.com/open-telemetry/semantic-conventions/blob/main/docs/general/attributes.md#source-code-attributes

instrumentation/log4j/log4j-appender-2.17/library/src/main/java/io/opentelemetry/instrumentation/log4j/appender/v2_17/LogEventToReplay.java

@@ -0,0 +1,209 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.log4j.appender.v2_17;
+
+import java.util.Map;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.Marker;
+import org.apache.logging.log4j.ThreadContext;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.impl.ThrowableProxy;
+import org.apache.logging.log4j.core.time.Instant;
+import org.apache.logging.log4j.message.Message;
+import org.apache.logging.log4j.message.StringMapMessage;
+import org.apache.logging.log4j.message.StructuredDataMessage;
+import org.apache.logging.log4j.util.ReadOnlyStringMap;
+
+class LogEventToReplay implements LogEvent {
+
+  private static final long serialVersionUID = -3639807661256104630L;
+
+  private final LogEvent logEvent;
+  private final String loggerName;
+  private final Message message;
+  private final Level level;
+  private final Instant instant;
+  private final Throwable thrown;
+  private final Marker marker;
+  private final ReadOnlyStringMap contextData;
+  private final String threadName;
+  private final long threadId;
+
+  // Log4j 2 seems to reuse Message objects
+  private static class MessageCopy implements Message {
+
+    private static final long serialVersionUID = 6921483958779195777L;
+    private final String formattedMessage;
+    private final String format;
+    private final Object[] parameters;
+    private final Throwable throwable;
+
+    public MessageCopy(Message message) {
+      this.formattedMessage = message.getFormattedMessage();
+      this.format = message.getFormat();
+      this.parameters = message.getParameters();
+      this.throwable = message.getThrowable();
+    }
+
+    @Override
+    public String getFormattedMessage() {
+      return formattedMessage;
+    }
+
+    @Override
+    public String getFormat() {
+      return format;
+    }
+
+    @Override
+    public Object[] getParameters() {
+      return parameters;
+    }
+
+    @Override
+    public Throwable getThrowable() {
+      return throwable;
+    }
+  }
+
+  LogEventToReplay(LogEvent logEvent) {
+    this.logEvent = logEvent;
+    this.loggerName = logEvent.getLoggerName();
+    Message messageOrigin = logEvent.getMessage();
+    if (messageOrigin instanceof StructuredDataMessage) {
+      StructuredDataMessage structuredDataMessage = (StructuredDataMessage) messageOrigin;
+      this.message =
+          new StructuredDataMessage(
+              structuredDataMessage.getId(),
+              structuredDataMessage.getFormat(),
+              structuredDataMessage.getType(),
+              structuredDataMessage.getData());
+    } else if (messageOrigin instanceof StringMapMessage) {
+      this.message = messageOrigin;
+    } else {
+      this.message = new MessageCopy(logEvent.getMessage());
+    }
+
+    this.level = logEvent.getLevel();
+    this.instant = logEvent.getInstant();
+    this.thrown = logEvent.getThrown();
+    this.marker = logEvent.getMarker();
+    this.contextData = logEvent.getContextData();
+    this.threadName = logEvent.getThreadName();
+    this.threadId = logEvent.getThreadId();
+  }
+
+  @Override
+  public LogEvent toImmutable() {
+    return logEvent.toImmutable();
+  }
+
+  @SuppressWarnings("deprecation") // Delegate
+  @Override
+  public Map<String, String> getContextMap() {
+    return logEvent.getContextMap();
+  }
+
+  @Override
+  public ReadOnlyStringMap getContextData() {
+    return contextData;
+  }
+
+  @Override
+  public ThreadContext.ContextStack getContextStack() {
+    return logEvent.getContextStack();
+  }
+
+  @Override
+  public String getLoggerFqcn() {
+    return logEvent.getLoggerFqcn();
+  }
+
+  @Override
+  public Level getLevel() {
+    return level;
+  }
+
+  @Override
+  public String getLoggerName() {
+    return loggerName;
+  }
+
+  @Override
+  public Marker getMarker() {
+    return marker;
+  }
+
+  @Override
+  public Message getMessage() {
+    return message;
+  }
+
+  @Override
+  public long getTimeMillis() {
+    return logEvent.getTimeMillis();
+  }
+
+  @Override
+  public Instant getInstant() {
+    return instant;
+  }
+
+  @Override
+  public StackTraceElement getSource() {
+    return logEvent.getSource();
+  }
+
+  @Override
+  public String getThreadName() {
+    return threadName;
+  }
+
+  @Override
+  public long getThreadId() {
+    return threadId;
+  }
+
+  @Override
+  public int getThreadPriority() {
+    return logEvent.getThreadPriority();
+  }
+
+  @Override
+  public Throwable getThrown() {
+    return thrown;
+  }
+
+  @Override
+  public ThrowableProxy getThrownProxy() {
+    return logEvent.getThrownProxy();
+  }
+
+  @Override
+  public boolean isEndOfBatch() {
+    return logEvent.isEndOfBatch();
+  }
+
+  @Override
+  public boolean isIncludeLocation() {
+    return logEvent.isIncludeLocation();
+  }
+
+  @Override
+  public void setEndOfBatch(boolean endOfBatch) {
+    logEvent.setEndOfBatch(endOfBatch);
+  }
+
+  @Override
+  public void setIncludeLocation(boolean locationRequired) {
+    logEvent.setIncludeLocation(locationRequired);
+  }
+
+  @Override
+  public long getNanoTime() {
+    return logEvent.getNanoTime();
+  }
+}

instrumentation/log4j/log4j-appender-2.17/library/src/main/java/io/opentelemetry/instrumentation/log4j/appender/v2_17/OpenTelemetryAppender.java

@@ -15,7 +15,10 @@
 import java.io.Serializable;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
@@ -48,6 +51,10 @@ public class OpenTelemetryAppender extends AbstractAppender {
   private final LogEventMapper<ReadOnlyStringMap> mapper;
   private volatile OpenTelemetry openTelemetry;
 
+  private BlockingQueue<LogEventToReplay> eventsToReplay = new ArrayBlockingQueue<>(1000);
+
+  private final AtomicBoolean logCacheWarningDisplayed = new AtomicBoolean();
+
   /**
    * Installs the {@code openTelemetry} instance on any {@link OpenTelemetryAppender}s identified in
    * the {@link LoggerContext}.
@@ -82,6 +89,7 @@ public static class Builder<B extends Builder<B>> extends AbstractAppender.Build
     @PluginBuilderAttribute private boolean captureMapMessageAttributes;
     @PluginBuilderAttribute private boolean captureMarkerAttribute;
     @PluginBuilderAttribute private String captureContextDataAttributes;
+    @PluginBuilderAttribute private int firstLogsCacheSize;
 
     @Nullable private OpenTelemetry openTelemetry;
 
@@ -121,6 +129,17 @@ public B setCaptureContextDataAttributes(String captureContextDataAttributes) {
       return asBuilder();
     }
 
+    /**
+     * Log telemetry is emitted after the initialization of the OpenTelemetry Logback appender with
+     * an {@link OpenTelemetry} object. This setting allows you to modify the size of the cache used
+     * to replay the first logs.
+     */
+    @CanIgnoreReturnValue
+    public B setFirstLogsCacheSize(int firstLogsCacheSize) {
+      this.firstLogsCacheSize = firstLogsCacheSize;
+      return asBuilder();
+    }
+
     /** Configures the {@link OpenTelemetry} used to append logs. */
     @CanIgnoreReturnValue
     public B setOpenTelemetry(OpenTelemetry openTelemetry) {
@@ -144,6 +163,7 @@ public OpenTelemetryAppender build() {
           captureMapMessageAttributes,
           captureMarkerAttribute,
           captureContextDataAttributes,
+          firstLogsCacheSize,
           openTelemetry);
     }
   }
@@ -158,17 +178,23 @@ private OpenTelemetryAppender(
       boolean captureMapMessageAttributes,
       boolean captureMarkerAttribute,
       String captureContextDataAttributes,
+      int firstLogsCacheSize,
       OpenTelemetry openTelemetry) {
 
     super(name, filter, layout, ignoreExceptions, properties);
+    List<String> captureContextDataAttributesAsList =
+        splitAndFilterBlanksAndNulls(captureContextDataAttributes);
     this.mapper =
         new LogEventMapper<>(
             ContextDataAccessorImpl.INSTANCE,
             captureExperimentalAttributes,
             captureMapMessageAttributes,
             captureMarkerAttribute,
-            splitAndFilterBlanksAndNulls(captureContextDataAttributes));
+            captureContextDataAttributesAsList);
     this.openTelemetry = openTelemetry;
+    if (firstLogsCacheSize != 0) {
+      this.eventsToReplay = new ArrayBlockingQueue<>(firstLogsCacheSize);
+    }
   }
 
   private static List<String> splitAndFilterBlanksAndNulls(String value) {
@@ -187,10 +213,30 @@ private static List<String> splitAndFilterBlanksAndNulls(String value) {
    */
   public void setOpenTelemetry(OpenTelemetry openTelemetry) {
     this.openTelemetry = openTelemetry;
+    while (!eventsToReplay.isEmpty()) {
+      try {
+        LogEvent eventToReplay = eventsToReplay.poll(10, TimeUnit.MILLISECONDS);
+        append(eventToReplay);
+      } catch (InterruptedException e) {
+        // Ignore
+      }
+    }
   }
 
+  @SuppressWarnings("SystemOut")
   @Override
   public void append(LogEvent event) {
+    if (openTelemetry == OpenTelemetry.noop()) {
+      if (eventsToReplay.remainingCapacity() > 0) {
+        LogEventToReplay logEventToReplay = new LogEventToReplay(event);
+        eventsToReplay.offer(logEventToReplay);
+      } else if (!logCacheWarningDisplayed.getAndSet(true)) {
+        System.err.println(
+            "Log cache size of the OpenTelemetry appender is too small. firstLogsCacheSize value has to be increased");
+      }
+      return;
+    }
+
     String instrumentationName = event.getLoggerName();
     if (instrumentationName == null || instrumentationName.isEmpty()) {
       instrumentationName = "ROOT";
@@ -209,7 +255,9 @@ public void append(LogEvent event) {
         event.getLevel(),
         event.getMarker(),
         event.getThrown(),
-        contextData);
+        contextData,
+        event.getThreadName(),
+        event.getThreadId());
 
     Instant timestamp = event.getInstant();
     if (timestamp != null) {