Support agentless log submission for JBoss LogManager

dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/log/AgentlessLogSubmission.java

@@ -0,0 +1,147 @@
+package datadog.trace.bootstrap.instrumentation.log;
+
+import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.traceConfig;
+
+import datadog.trace.api.Config;
+import datadog.trace.api.CorrelationIdentifier;
+import datadog.trace.api.logging.intake.LogsIntake;
+import datadog.trace.bootstrap.instrumentation.api.Tags;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * Submits events captured by logging backend instrumentations to the agentless logs intake.
+ *
+ * <p>Events use the payload shape of Log4j's JSON layout, which the logs backend already parses.
+ * Callers must capture events on the producer thread, so that the correlation identifiers match the
+ * active span.
+ */
+public final class AgentlessLogSubmission {
+
+  public static final String TRACE = "TRACE";
+  public static final String DEBUG = "DEBUG";
+  public static final String INFO = "INFO";
+  public static final String WARN = "WARN";
+  public static final String ERROR = "ERROR";
+  public static final String FATAL = "FATAL";
+
+  private static final int MAX_STACKTRACE_STRING_LENGTH = 16 * 1_024;
+
+  private static final int SUBMISSION_SEVERITY =
+      severity(Config.get().getAgentlessLogSubmissionLevel());
+
+  private AgentlessLogSubmission() {}
+
+  /** Returns whether events of the given level reach the configured submission level. */
+  public static boolean isLevelEnabled(String level) {
+    return severity(level) >= SUBMISSION_SEVERITY;
+  }
+
+  /** Converts a {@link java.util.logging.Level} integer value to a level name of this class. */
+  public static String julLevel(int value) {
+    if (value >= 1100) {
+      return FATAL;
+    } else if (value >= 1000) {
+      return ERROR;
+    } else if (value >= 900) {
+      return WARN;
+    } else if (value >= 800) {
+      return INFO;
+    } else if (value >= 500) {
+      return DEBUG;
+    }
+    return TRACE;
+  }
+
+  /**
+   * Submits an event.
+   *
+   * @param level a level name of this class
+   * @param context the user context of the event, such as the MDC, or {@code null}
+   */
+  public static void submit(
+      String thread,
+      String level,
+      String loggerName,
+      String message,
+      Throwable thrown,
+      long timestamp,
+      Map<String, String> context) {
+    Map<String, Object> log = new HashMap<>();
+    log.put("thread", thread);
+    log.put("level", level);
+    log.put("loggerName", loggerName);
+    log.put("message", message);
+    log.put("timestamp", timestamp);
+    if (thrown != null) {
+      log.put("thrown", thrown(thrown));
+    }
+    Map<String, String> contextMap = new HashMap<>();
+    if (traceConfig().isLogsInjectionEnabled()) {
+      addCorrelationValues(contextMap);
+    }
+    if (context != null) {
+      contextMap.putAll(context);
+    }
+    log.put("contextMap", contextMap);
+    LogsIntake.log(log);
+  }
+
+  private static Map<String, Object> thrown(Throwable thrown) {
+    Map<String, Object> thrownLog = new HashMap<>();
+    thrownLog.put("message", thrown.getMessage());
+    thrownLog.put("name", thrown.getClass().getCanonicalName());
+    StringWriter stringWriter = new StringWriter();
+    thrown.printStackTrace(new PrintWriter(stringWriter));
+    StringBuffer stackTrace = stringWriter.getBuffer();
+    thrownLog.put(
+        "extendedStackTrace",
+        stackTrace.substring(0, Math.min(stackTrace.length(), MAX_STACKTRACE_STRING_LENGTH)));
+    return thrownLog;
+  }
+
+  private static void addCorrelationValues(Map<String, String> context) {
+    Config config = Config.get();
+    putIfNotEmpty(context, Tags.DD_SERVICE, config.getServiceName());
+    putIfNotEmpty(context, Tags.DD_ENV, config.getEnv());
+    putIfNotEmpty(context, Tags.DD_VERSION, config.getVersion());
+    String traceId = CorrelationIdentifier.getTraceId();
+    if (traceId != null && !traceId.equals("0")) {
+      context.put(CorrelationIdentifier.getTraceIdKey(), traceId);
+    }
+    String spanId = CorrelationIdentifier.getSpanId();
+    if (spanId != null && !spanId.equals("0")) {
+      context.put(CorrelationIdentifier.getSpanIdKey(), spanId);
+    }
+  }
+
+  private static void putIfNotEmpty(Map<String, String> context, String key, String value) {
+    if (value != null && !value.isEmpty()) {
+      context.put(key, value);
+    }
+  }
+
+  static int severity(String level) {
+    switch (level == null ? INFO : level.toUpperCase(Locale.ROOT)) {
+      case "ALL":
+        return 0;
+      case TRACE:
+        return 1;
+      case DEBUG:
+        return 2;
+      case WARN:
+        return 4;
+      case ERROR:
+        return 5;
+      case FATAL:
+        return 6;
+      case "OFF":
+        return 7;
+      default:
+        return 3;
+    }
+  }
+}

dd-java-agent/agent-bootstrap/src/test/java/datadog/trace/bootstrap/instrumentation/log/AgentlessLogSubmissionTest.java

@@ -0,0 +1,122 @@
+package datadog.trace.bootstrap.instrumentation.log;
+
+import static datadog.trace.bootstrap.instrumentation.log.AgentlessLogSubmission.isLevelEnabled;
+import static datadog.trace.bootstrap.instrumentation.log.AgentlessLogSubmission.julLevel;
+import static datadog.trace.bootstrap.instrumentation.log.AgentlessLogSubmission.severity;
+import static datadog.trace.bootstrap.instrumentation.log.AgentlessLogSubmission.submit;
+import static java.util.Collections.singletonMap;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import datadog.trace.api.logging.intake.LogsIntake;
+import datadog.trace.api.logging.intake.LogsWriter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.tabletest.junit.TableTest;
+
+class AgentlessLogSubmissionTest {
+
+  private final List<Map<String, Object>> messages = new ArrayList<>();
+
+  @BeforeEach
+  void registerWriter() {
+    LogsIntake.registerWriter(
+        new LogsWriter() {
+          @Override
+          public void log(Map<String, Object> message) {
+            messages.add(message);
+          }
+
+          @Override
+          public void start() {}
+
+          @Override
+          public void shutdown() {}
+        });
+  }
+
+  @AfterEach
+  void unregisterWriter() {
+    LogsIntake.registerWriter(null);
+  }
+
+  @TableTest({
+    "scenario    | value | level",
+    "OFF         | 2000  | FATAL",
+    "JBoss FATAL | 1100  | FATAL",
+    "SEVERE      | 1000  | ERROR",
+    "WARNING     | 900   | WARN ",
+    "INFO        | 800   | INFO ",
+    "CONFIG      | 700   | DEBUG",
+    "FINE        | 500   | DEBUG",
+    "FINER       | 400   | TRACE",
+    "FINEST      | 300   | TRACE"
+  })
+  void mapsJulLevels(int value, String level) {
+    assertEquals(level, julLevel(value));
+  }
+
+  @TableTest({
+    "scenario      | lower | higher",
+    "ALL and TRACE | ALL   | TRACE ",
+    "TRACE, DEBUG  | trace | DEBUG ",
+    "DEBUG, INFO   | DEBUG | INFO  ",
+    "INFO, WARN    | INFO  | WARN  ",
+    "WARN, ERROR   | WARN  | ERROR ",
+    "ERROR, FATAL  | ERROR | FATAL ",
+    "FATAL, OFF    | FATAL | OFF   "
+  })
+  void ordersLevels(String lower, String higher) {
+    assertTrue(severity(lower) < severity(higher));
+  }
+
+  @Test
+  void defaultsToInfo() {
+    assertFalse(isLevelEnabled("DEBUG"));
+    assertTrue(isLevelEnabled("INFO"));
+    assertTrue(isLevelEnabled("unknown"));
+  }
+
+  @Test
+  void submitsEvent() {
+    submit(
+        "main",
+        "ERROR",
+        "logger",
+        "message",
+        new IllegalStateException("failure"),
+        42L,
+        singletonMap("custom", "value"));
+
+    assertEquals(1, messages.size());
+    Map<String, Object> message = messages.get(0);
+    assertEquals("main", message.get("thread"));
+    assertEquals("ERROR", message.get("level"));
+    assertEquals("logger", message.get("loggerName"));
+    assertEquals("message", message.get("message"));
+    assertEquals(42L, message.get("timestamp"));
+    assertEquals(singletonMap("custom", "value"), message.get("contextMap"));
+    Map<?, ?> thrown = (Map<?, ?>) message.get("thrown");
+    assertEquals(IllegalStateException.class.getName(), thrown.get("name"));
+    assertEquals("failure", thrown.get("message"));
+    assertTrue(
+        thrown
+            .get("extendedStackTrace")
+            .toString()
+            .startsWith("java.lang.IllegalStateException: failure"));
+  }
+
+  @Test
+  void submitsEventWithoutOptionalFields() {
+    submit("main", "INFO", "logger", "message", null, 42L, null);
+
+    assertEquals(1, messages.size());
+    assertFalse(messages.get(0).containsKey("thrown"));
+    assertTrue(((Map<?, ?>) messages.get(0).get("contextMap")).isEmpty());
+  }
+}

dd-java-agent/agent-logs-intake/src/main/java/datadog/trace/logging/intake/LogsWriterImpl.java

@@ -89,14 +89,16 @@ public void log(Map<String, Object> message) {
   }
 
   private void logPollingLoop() {
-    BackendApi backendApi = apiFactory.createBackendApi(intake);
-    LogsDispatcher logsDispatcher = new LogsDispatcher(backendApi);
+    LogsDispatcher logsDispatcher = null;
 
     while (!Thread.currentThread().isInterrupted()) {
       try {
         List<Map<String, Object>> batch = new ArrayList<>();
         batch.add(messageQueue.take());
         messageQueue.drainTo(batch);
+        if (logsDispatcher == null) {
+          logsDispatcher = createDispatcher();
+        }
         logsDispatcher.dispatch(batch);
 
       } catch (InterruptedException e) {
@@ -107,7 +109,19 @@ private void logPollingLoop() {
     List<Map<String, Object>> batch = new ArrayList<>();
     messageQueue.drainTo(batch);
     if (!batch.isEmpty()) {
+      if (logsDispatcher == null) {
+        logsDispatcher = createDispatcher();
+      }
       logsDispatcher.dispatch(batch);
     }
   }
+
+  /**
+   * Creating the HTTP client initializes TLS, which can initialize the JUL {@code LogManager}. It
+   * is created for the first log, after the application has configured its own log manager.
+   */
+  private LogsDispatcher createDispatcher() {
+    BackendApi backendApi = apiFactory.createBackendApi(intake);
+    return new LogsDispatcher(backendApi);
+  }
 }

dd-java-agent/instrumentation/jboss/jboss-logmanager-1.1/build.gradle

@@ -11,6 +11,22 @@ muzzle {
   }
 }
 
+addTestSuiteForDir('latestDepTest', 'test')
+addTestSuiteExtendingForDir('latestDepForkedTest', 'latestDepTest', 'test')
+
+["latestDepTest", "latestDepForkedTest"].each {
+  tasks.named(it, Test) {
+    testJvmConstraints {
+      // jboss-logmanager 3.2 requires Java 17
+      minJavaVersion = JavaVersion.VERSION_17
+    }
+  }
+}
+
 dependencies {
   compileOnly group: 'org.jboss.logmanager', name: 'jboss-logmanager', version: '1.1.0.GA'
+
+  // LoggerNode.publish(ExtLogRecord), used by the instrumentation, was introduced in 1.2.0.GA
+  testImplementation group: 'org.jboss.logmanager', name: 'jboss-logmanager', version: '1.2.0.GA'
+  latestDepTestImplementation group: 'org.jboss.logmanager', name: 'jboss-logmanager', version: '+'
 }