fix span lifecycle in the launcher listener

Author: aboitreaud

dd-java-agent/instrumentation/spark/spark-common/src/main/java/datadog/trace/instrumentation/spark/SparkLauncherListener.java

@@ -11,6 +11,12 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * Listener for SparkLauncher spans. Tracks the lifecycle of a Spark application submitted via
+ * SparkLauncher.startApplication(). Only a single launcher span can be active at a time. Subsequent
+ * calls to startApplication() from the same or different launcher instances will not create spans;
+ * only the first launch in the JVM is traced
+ */
 public class SparkLauncherListener implements SparkAppHandle.Listener {
 
   private static final Logger log = LoggerFactory.getLogger(SparkLauncherListener.class);
@@ -81,46 +87,51 @@ public static synchronized void finishSpanWithThrowable(Throwable throwable) {
 
   @Override
   public void stateChanged(SparkAppHandle handle) {
-    SparkAppHandle.State state = handle.getState();
-    AgentSpan span = launcherSpan;
-    if (span != null) {
-      span.setTag("spark.launcher.app_state", state.toString());
-
-      String appId = handle.getAppId();
-      if (appId != null) {
-        span.setTag("spark.app_id", appId);
-        span.setTag("app_id", appId);
-      }
+    synchronized (SparkLauncherListener.class) {
+      SparkAppHandle.State state = handle.getState();
+      AgentSpan span = launcherSpan;
+      if (span != null) {
+        span.setTag("spark.launcher.app_state", state.toString());
+
+        String appId = handle.getAppId();
+        if (appId != null) {
+          span.setTag("spark.app_id", appId);
+          span.setTag("app_id", appId);
+        }
 
-      if (state.isFinal()) {
-        if (state == SparkAppHandle.State.FAILED
-            || state == SparkAppHandle.State.KILLED
-            || state == SparkAppHandle.State.LOST) {
-          // Set error tags but don't finish yet — RunMainAdvice may add the throwable
-          // with the full stack trace. The span will be finished by RunMainAdvice or
-          // the shutdown hook.
-          span.setError(true);
-          span.setTag(DDTags.ERROR_TYPE, "Spark Launcher Failed");
-          span.setTag(DDTags.ERROR_MSG, "Application " + state);
-        } else {
-          finishSpan(false, null);
+        if (state.isFinal()) {
+          if (state == SparkAppHandle.State.FAILED
+              || state == SparkAppHandle.State.KILLED
+              || state == SparkAppHandle.State.LOST) {
+            finishSpan(true, "Application " + state);
+          } else {
+            finishSpan(false, null);
+          }
         }
       }
     }
   }
 
   @Override
   public void infoChanged(SparkAppHandle handle) {
-    AgentSpan span = launcherSpan;
-    if (span != null) {
-      String appId = handle.getAppId();
-      if (appId != null) {
-        span.setTag("spark.app_id", appId);
-        span.setTag("app_id", appId);
+    synchronized (SparkLauncherListener.class) {
+      AgentSpan span = launcherSpan;
+      if (span != null) {
+        String appId = handle.getAppId();
+        if (appId != null) {
+          span.setTag("spark.app_id", appId);
+          span.setTag("app_id", appId);
+        }
       }
     }
   }
 
+  /**
+   * Extract launcher configuration via reflection and set as span tags. Secret redaction uses the
+   * default pattern only (not spark.redaction.regex) because the SparkLauncher conf map is a plain
+   * Map, not a SparkConf, so there is no way to read the user's custom redaction regex at this
+   * point.
+   */
   private static void setLauncherConfigTags(AgentSpan span, Object launcher) {
     try {
       Field builderField = launcher.getClass().getSuperclass().getDeclaredField("builder");

dd-java-agent/instrumentation/spark/spark-common/src/test/groovy/datadog/trace/instrumentation/spark/SparkLauncherTest.groovy

@@ -130,7 +130,7 @@ class SparkLauncherTest extends InstrumentationSpecification {
     }
   }
 
-  def "SparkLauncherListener sets error tags on FAILED state but does not finish span"() {
+  def "SparkLauncherListener finishes span with error on FAILED state"() {
     setup:
     SparkLauncherListener.launcherSpan = null
     def tracer = AgentTracer.get()
@@ -151,18 +151,22 @@ class SparkLauncherTest extends InstrumentationSpecification {
     listener.stateChanged(handle)
 
     then:
-    // Span stays open so RunMainAdvice can add the throwable
-    SparkLauncherListener.launcherSpan != null
-    SparkLauncherListener.launcherSpan.isError()
-    SparkLauncherListener.launcherSpan.getTags()["error.type"] == "Spark Launcher Failed"
-    SparkLauncherListener.launcherSpan.getTags()["error.message"] == "Application FAILED"
-    SparkLauncherListener.launcherSpan.getTags()["spark.app_id"] == "app-456"
-
-    cleanup:
-    SparkLauncherListener.finishSpan(false, null)
+    SparkLauncherListener.launcherSpan == null
+    assertTraces(1) {
+      trace(1) {
+        span {
+          operationName "spark.launcher.launch"
+          spanType "spark"
+          errored true
+          assert span.tags["error.type"] == "Spark Launcher Failed"
+          assert span.tags["error.message"] == "Application FAILED"
+          assert span.tags["spark.app_id"] == "app-456"
+        }
+      }
+    }
   }
 
-  def "finishSpanWithThrowable adds stack trace after FAILED state"() {
+  def "finishSpanWithThrowable is no-op after span already finished"() {
     setup:
     SparkLauncherListener.launcherSpan = null
     def tracer = AgentTracer.get()
@@ -178,10 +182,11 @@ class SparkLauncherTest extends InstrumentationSpecification {
     def handle = Mock(SparkAppHandle)
 
     when:
-    // Simulate: listener sets error tags, then RunMainAdvice finishes with throwable
+    // Listener finishes the span on FAILED, then RunMainAdvice tries to finish again
     handle.getState() >> SparkAppHandle.State.FAILED
     handle.getAppId() >> "app-456"
     listener.stateChanged(handle)
+    // This should be a no-op since the span is already finished
     SparkLauncherListener.finishSpanWithThrowable(new RuntimeException("job crashed"))
 
     then:
@@ -192,10 +197,8 @@ class SparkLauncherTest extends InstrumentationSpecification {
           operationName "spark.launcher.launch"
           spanType "spark"
           errored true
-          assert span.tags["error.type"] == "java.lang.RuntimeException"
-          assert span.tags["error.message"] == "job crashed"
-          assert span.tags["error.stack"] != null
-          assert span.tags["spark.app_id"] == "app-456"
+          assert span.tags["error.type"] == "Spark Launcher Failed"
+          assert span.tags["error.message"] == "Application FAILED"
         }
       }
     }