Introduced trace post-processing

Author: ValentinZakharov

dd-java-agent/testing/src/main/groovy/datadog/trace/agent/test/AgentTestRunner.groovy

@@ -547,6 +547,12 @@ abstract class AgentTestRunner extends DDSpecification implements AgentBuilder.L
       check()
       return delegate.getDataCurrent(key)
     }
+
+    @Override
+    void setRequiresPostProcessing(boolean postProcessing) {
+      check()
+      delegate.setRequiresPostProcessing(postProcessing)
+    }
   }
 
   /** Override to clean up things after the agent is removed */

dd-trace-api/src/main/java/datadog/trace/api/config/TracerConfig.java

@@ -133,5 +133,9 @@ public final class TracerConfig {
 
   public static final String TRACE_FLUSH_INTERVAL = "trace.flush.interval";
 
+  public static final String TRACE_POST_PROCESSING_ENABLED = "trace.post-processing.enabled";
+
+  public static final String TRACE_POST_PROCESSING_TIMEOUT = "trace.post-processing.timeout";
+
   private TracerConfig() {}
 }

dd-trace-api/src/main/java/datadog/trace/api/internal/TraceSegment.java

@@ -75,6 +75,13 @@ default void setTagCurrent(String key, Object value) {
    */
   void setDataCurrent(String key, Object value);
 
+  /**
+   * Set flag to indicate the need for post-processing
+   *
+   * @param postProcessing flag to indicate the need for post-processing
+   */
+  void setRequiresPostProcessing(boolean postProcessing);
+
   /**
    * Gets the value of the tag from the current span in this {@code TraceSegment}.
    *
@@ -112,5 +119,8 @@ public void setDataCurrent(String key, Object value) {}
     public Object getDataCurrent(String key) {
       return null;
     }
+
+    @Override
+    public void setRequiresPostProcessing(boolean postProcessing) {}
   }
 }

dd-trace-core/src/main/java/datadog/trace/common/writer/TraceProcessingWorker.java

@@ -13,7 +13,9 @@
 import datadog.trace.common.writer.ddagent.PrioritizationStrategy;
 import datadog.trace.core.CoreSpan;
 import datadog.trace.core.DDSpan;
+import datadog.trace.core.DDSpanContext;
 import datadog.trace.core.monitor.HealthMetrics;
+import datadog.trace.util.AgentThreadFactory;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -235,6 +237,7 @@ public void onEvent(Object event) {
       try {
         if (event instanceof List) {
           List<DDSpan> trace = (List<DDSpan>) event;
+          maybePostProcessing(trace);
           // TODO populate `_sample_rate` metric in a way that accounts for lost/dropped traces
           payloadDispatcher.addTrace(trace);
         } else if (event instanceof FlushEvent) {
@@ -295,5 +298,40 @@ private void consumeBatch(MessagePassingQueue<Object> queue) {
     protected boolean queuesAreEmpty() {
       return primaryQueue.isEmpty() && secondaryQueue.isEmpty();
     }
+
+    private void maybePostProcessing(List<DDSpan> trace) {
+      if (!Config.get().isTracePostProcessingEnabled()) {
+        return;
+      }
+
+      if (trace == null || trace.isEmpty()) {
+        return;
+      }
+
+      DDSpanContext context = trace.get(0).context();
+      if (!context.isRequiresPostProcessing()) {
+        return;
+      }
+
+      Thread thread =
+          AgentThreadFactory.newAgentThread(
+              AgentThreadFactory.AgentThread.TRACE_POST_PROCESSOR,
+              () -> {
+                // DO A POST PROCESSING
+              });
+      thread.start();
+
+      try {
+        long timeout = Config.get().getTracePostProcessingTimeout();
+        thread.join(timeout);
+        if (thread.isAlive()) {
+          thread.interrupt();
+        }
+      } catch (InterruptedException e) {
+        if (log.isDebugEnabled()) {
+          log.debug("Trace post-processing is interrupted.", e);
+        }
+      }
+    }
   }
 }

dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java

@@ -140,6 +140,7 @@ public class DDSpanContext
   private final boolean injectBaggageAsTags;
   private volatile int encodedOperationName;
   private volatile int encodedResourceName;
+  private boolean requiresPostProcessing;
 
   public DDSpanContext(
       final DDTraceId traceId,
@@ -945,4 +946,13 @@ private String getTagName(String key) {
     // TODO is this decided?
     return "_dd." + key + ".json";
   }
+
+  @Override
+  public void setRequiresPostProcessing(boolean postProcessing) {
+    this.requiresPostProcessing = postProcessing;
+  }
+
+  public boolean isRequiresPostProcessing() {
+    return requiresPostProcessing;
+  }
 }

internal-api/src/main/java/datadog/trace/api/Config.java

@@ -434,6 +434,8 @@
 import static datadog.trace.api.config.TracerConfig.TRACE_PEER_SERVICE_COMPONENT_OVERRIDES;
 import static datadog.trace.api.config.TracerConfig.TRACE_PEER_SERVICE_DEFAULTS_ENABLED;
 import static datadog.trace.api.config.TracerConfig.TRACE_PEER_SERVICE_MAPPING;
+import static datadog.trace.api.config.TracerConfig.TRACE_POST_PROCESSING_ENABLED;
+import static datadog.trace.api.config.TracerConfig.TRACE_POST_PROCESSING_TIMEOUT;
 import static datadog.trace.api.config.TracerConfig.TRACE_PROPAGATION_EXTRACT_FIRST;
 import static datadog.trace.api.config.TracerConfig.TRACE_PROPAGATION_STYLE;
 import static datadog.trace.api.config.TracerConfig.TRACE_PROPAGATION_STYLE_EXTRACT;
@@ -924,6 +926,8 @@ static class HostNameHolder {
 
   private final boolean axisPromoteResourceName;
   private final float traceFlushIntervalSeconds;
+  private final boolean tracePostProcessingEnabled;
+  private final long tracePostProcessingTimeout;
 
   private final boolean telemetryDebugRequestsEnabled;
 
@@ -2036,6 +2040,11 @@ PROFILING_DATADOG_PROFILER_ENABLED, isDatadogProfilerSafeInCurrentEnvironment())
           "Agentless profiling activated but no api key provided. Profile uploading will likely fail");
     }
 
+    this.tracePostProcessingEnabled =
+        configProvider.getBoolean(TRACE_POST_PROCESSING_ENABLED, true);
+
+    this.tracePostProcessingTimeout = configProvider.getLong(TRACE_POST_PROCESSING_TIMEOUT, 1000);
+
     if (isCiVisibilityEnabled()
         && ciVisibilityAgentlessEnabled
         && (apiKey == null || apiKey.isEmpty())) {
@@ -2136,6 +2145,14 @@ public float getTraceFlushIntervalSeconds() {
     return traceFlushIntervalSeconds;
   }
 
+  public boolean isTracePostProcessingEnabled() {
+    return tracePostProcessingEnabled;
+  }
+
+  public long getTracePostProcessingTimeout() {
+    return tracePostProcessingTimeout;
+  }
+
   public boolean isIntegrationSynapseLegacyOperationName() {
     return integrationSynapseLegacyOperationName;
   }

internal-api/src/main/java/datadog/trace/util/AgentThreadFactory.java

@@ -16,6 +16,7 @@ public enum AgentThread {
     TRACE_STARTUP("dd-agent-startup-datadog-tracer"),
     TRACE_MONITOR("dd-trace-monitor"),
     TRACE_PROCESSOR("dd-trace-processor"),
+    TRACE_POST_PROCESSOR("dd-trace-post-processor"),
     SPAN_SAMPLING_PROCESSOR("dd-span-sampling-processor"),
     TRACE_CASSANDRA_ASYNC_SESSION("dd-cassandra-session-executor"),