Adds minimal traceparent header support to Elasticsearch

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -434,7 +434,10 @@ public ActionModule(Settings settings, IndexNameExpressionResolver indexNameExpr
         destructiveOperations = new DestructiveOperations(settings, clusterSettings);
         Set<RestHeaderDefinition> headers = Stream.concat(
             actionPlugins.stream().flatMap(p -> p.getRestHeaders().stream()),
-            Stream.of(new RestHeaderDefinition(Task.X_OPAQUE_ID, false))
+            Stream.of(
+                new RestHeaderDefinition(Task.X_OPAQUE_ID, false),
+                new RestHeaderDefinition(Task.TRACE_PARENT, false)
+            )
         ).collect(Collectors.toSet());
         UnaryOperator<RestHandler> restWrapper = null;
         for (ActionPlugin plugin : actionPlugins) {

server/src/main/java/org/elasticsearch/common/logging/ECSJsonLayout.java

@@ -55,11 +55,12 @@ public EcsLayout build() {
         private KeyValuePair[] additionalFields() {
             return new KeyValuePair[] {
                 new KeyValuePair("event.dataset", dataset),
+                new KeyValuePair("trace.id", "%trace_id"),
                 new KeyValuePair("elasticsearch.cluster.uuid", "%cluster_id"),
                 new KeyValuePair("elasticsearch.node.id", "%node_id"),
                 new KeyValuePair("elasticsearch.node.name", "%ESnode_name"),
                 new KeyValuePair("elasticsearch.cluster.name", "${sys:es.logs.cluster_name}"), };
-        }
+    }
 
         public String getDataset() {
             return dataset;

server/src/main/java/org/elasticsearch/common/logging/TraceIdConverter.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.common.logging;
+
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.config.plugins.Plugin;
+import org.apache.logging.log4j.core.pattern.ConverterKeys;
+import org.apache.logging.log4j.core.pattern.LogEventPatternConverter;
+import org.apache.logging.log4j.core.pattern.PatternConverter;
+import org.elasticsearch.tasks.Task;
+
+import java.util.Objects;
+
+/**
+ * Pattern converter to format the trace id provided in the traceparent header into JSON fields <code>trace.id</code>.
+ */
+@Plugin(category = PatternConverter.CATEGORY, name = "TraceIdConverter")
+@ConverterKeys({"trace_id"})
+public final class TraceIdConverter extends LogEventPatternConverter {
+    /**
+     * Called by log4j2 to initialize this converter.
+     */
+    public static TraceIdConverter newInstance(@SuppressWarnings("unused") final String[] options) {
+        return new TraceIdConverter();
+    }
+
+    public TraceIdConverter() {
+        super("trace_id", "trace_id");
+    }
+
+    public static String getTraceId() {
+        return HeaderWarning.THREAD_CONTEXT.stream()
+            .map(t -> t.<String>getTransient(Task.TRACE_ID))
+            .filter(Objects::nonNull)
+            .findFirst()
+            .orElse(null);
+    }
+
+    /**
+     * Formats the trace.id into json fields.
+     *
+     * @param event - a log event is ignored in this method as it uses the clusterId value
+     *              from <code>NodeAndClusterIdStateListener</code> to format
+     */
+    @Override
+    public void format(LogEvent event, StringBuilder toAppendTo) {
+        String traceId = getTraceId();
+        if (traceId != null) {
+            toAppendTo.append(traceId);
+        }
+    }
+
+}

server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java

@@ -109,11 +109,19 @@ public StoredContext stashContext() {
          * This is needed so the DeprecationLogger in another thread can see the value of X-Opaque-ID provided by a user.
          * Otherwise when context is stash, it should be empty.
          */
-        if (context.requestHeaders.containsKey(Task.X_OPAQUE_ID)) {
-            ThreadContextStruct threadContextStruct =
-                DEFAULT_CONTEXT.putHeaders(Map.of(Task.X_OPAQUE_ID, context.requestHeaders.get(Task.X_OPAQUE_ID)));
+
+        if (context.requestHeaders.containsKey(Task.X_OPAQUE_ID) || context.requestHeaders.containsKey(Task.TRACE_PARENT)) {
+            Map<String, String> map = new HashMap<>(2, 1);
+            if (context.requestHeaders.containsKey(Task.X_OPAQUE_ID)) {
+                map.put(Task.X_OPAQUE_ID, context.requestHeaders.get(Task.X_OPAQUE_ID));
+            }
+            if (context.requestHeaders.containsKey(Task.TRACE_PARENT)) {
+                map.put(Task.TRACE_PARENT, context.requestHeaders.get(Task.TRACE_PARENT));
+            }
+            ThreadContextStruct threadContextStruct = DEFAULT_CONTEXT.putHeaders(map);
             threadLocal.set(threadContextStruct);
-        } else {
+        }
+        else {
             threadLocal.set(DEFAULT_CONTEXT);
         }
         return () -> {

server/src/main/java/org/elasticsearch/node/Node.java

@@ -574,7 +574,7 @@ protected Node(final Environment initialEnvironment,
             final Transport transport = networkModule.getTransportSupplier().get();
             Set<String> taskHeaders = Stream.concat(
                 pluginsService.filterPlugins(ActionPlugin.class).stream().flatMap(p -> p.getTaskHeaders().stream()),
-                Stream.of(Task.X_OPAQUE_ID)
+                Stream.of(Task.X_OPAQUE_ID, Task.TRACE_PARENT)
             ).collect(Collectors.toSet());
             final TransportService transportService = newTransportService(settings, transport, threadPool,
                 networkModule.getTransportInterceptor(), localNodeFactory, settingsModule.getClusterSettings(), taskHeaders);

server/src/main/java/org/elasticsearch/rest/RestController.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.http.HttpServerTransport;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.rest.RestHandler.Route;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.usage.UsageService;
 
 import java.io.ByteArrayOutputStream;
@@ -344,6 +345,17 @@ private void tryAllHandlers(final RestRequest request, final RestChannel channel
                         BytesRestResponse.
                             createSimpleErrorResponse(channel, BAD_REQUEST, "multiple values for single-valued header [" + name + "]."));
                     return;
+                } else if (name.equals(Task.TRACE_PARENT)) {
+                    String traceparent = distinctHeaderValues.get(0);
+                    // TODO rely on high performance `TraceContext` from apm-agent-java
+                    // This does little validation of the actual id values, since we only use them to persist in logging for now this
+                    // should be fine. Once we actually start instrumenting through apm-agent-java it will take care of validation.
+                    // Java has a fast path for single character regexes so this should not be as bad as its signature implies
+                    String[] tokens  = traceparent.split("-");
+                    if (tokens.length == 4) {
+                        threadContext.putTransient(Task.TRACE_ID, tokens[1]);
+                    }
+                    threadContext.putHeader(name, String.join(",", distinctHeaderValues));
                 } else {
                     threadContext.putHeader(name, String.join(",", distinctHeaderValues));
                 }

server/src/main/java/org/elasticsearch/tasks/Task.java

@@ -28,6 +28,10 @@ public class Task {
      */
     public static final String X_OPAQUE_ID = "X-Opaque-Id";
 
+    public static final String TRACE_PARENT = "traceparent";
+
+    public static final String TRACE_ID = "trace.id";
+
     private final long id;
 
     private final String type;