feat: add feature flag to disable spark.task span on failure

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

@@ -761,6 +761,10 @@ public void onTaskEnd(SparkListenerTaskEnd taskEnd) {
       return;
     }
 
+    if (!Config.get().isSparkTaskSpanOnFailureEnabled()) {
+      return;
+    }
+
     AgentSpan stageSpan = stageSpans.get(stageSpanKey);
     if (stageSpan == null) {
       return;

dd-java-agent/instrumentation/spark/spark-common/src/testFixtures/groovy/datadog/trace/instrumentation/spark/AbstractSparkListenerTest.groovy

@@ -6,6 +6,7 @@ import com.datadoghq.sketch.ddsketch.store.CollapsingLowestDenseStore
 import datadog.trace.agent.test.InstrumentationSpecification
 import datadog.trace.api.DDTraceId
 import datadog.trace.api.ProcessTags
+import org.apache.spark.ExceptionFailure
 import org.apache.spark.SparkConf
 import org.apache.spark.Success$
 import org.apache.spark.executor.TaskMetrics
@@ -184,6 +185,52 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
       )
   }
 
+  protected failedTaskEndEvent(Integer stageId, Long launchTime) {
+    def taskInfo = new TaskInfo(
+      0,
+      0,
+      0,
+      launchTime,
+      "some_executor_id",
+      "some_host",
+      TaskLocality.PROCESS_LOCAL(),
+      false
+      )
+    taskInfo.markFinished(org.apache.spark.TaskState.FAILED(), launchTime + 100)
+
+    def taskMetrics = new TaskMetrics()
+
+    def reason = new ExceptionFailure(
+      "java.lang.NullPointerException",
+      "null",
+      "java.lang.NullPointerException: null\n\tat SomeClass.method(SomeClass.java:1)",
+      "java.lang.NullPointerException: null\n\tat SomeClass.method(SomeClass.java:1)",
+      Option.empty(),
+      JavaConverters.asScalaBuffer([]).toSeq()
+      )
+
+    if (TestSparkComputation.getSparkVersion() < "3") {
+      return new SparkListenerTaskEnd(
+        stageId,
+        0,
+        "task_type",
+        reason,
+        taskInfo,
+        taskMetrics
+        )
+    }
+
+    return new SparkListenerTaskEnd(
+      stageId,
+      0,
+      "task_type",
+      reason,
+      taskInfo,
+      null,
+      taskMetrics
+      )
+  }
+
   protected executorAddedEvent(time=0L, executorId="executor-0", totalCores=0) {
     new SparkListenerExecutorAdded(
       time,
@@ -407,6 +454,80 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     }
   }
 
+  def "failing task emits spark.task span by default"() {
+    setup:
+    def listener = getTestDatadogSparkListener()
+
+    listener.onApplicationStart(applicationStartEvent(1000L))
+    listener.onJobStart(jobStartEvent(1, 1900L, [1]))
+    listener.onStageSubmitted(stageSubmittedEvent(1, 1900L))
+    listener.onTaskEnd(failedTaskEndEvent(1, 1900L))
+    listener.onStageCompleted(stageCompletedEvent(1, 2200L))
+    listener.onJobEnd(jobEndEvent(1, 2200L))
+    listener.onApplicationEnd(new SparkListenerApplicationEnd(3100L))
+
+    expect:
+    assertTraces(1) {
+      trace(4) {
+        span {
+          operationName "spark.application"
+          spanType "spark"
+        }
+        span {
+          operationName "spark.job"
+          spanType "spark"
+          childOf(span(0))
+        }
+        span {
+          operationName "spark.stage"
+          spanType "spark"
+          childOf(span(1))
+        }
+        span {
+          operationName "spark.task"
+          spanType "spark"
+          errored true
+          childOf(span(2))
+          assert span.tags["error.type"] == "Spark Task Failed"
+        }
+      }
+    }
+  }
+
+  def "feature flag disables spark.task span on failure"() {
+    setup:
+    injectSysConfig("spark.task-span-on-failure.enabled", "false")
+    def listener = getTestDatadogSparkListener()
+
+    listener.onApplicationStart(applicationStartEvent(1000L))
+    listener.onJobStart(jobStartEvent(1, 1900L, [1]))
+    listener.onStageSubmitted(stageSubmittedEvent(1, 1900L))
+    listener.onTaskEnd(failedTaskEndEvent(1, 1900L))
+    listener.onStageCompleted(stageCompletedEvent(1, 2200L))
+    listener.onJobEnd(jobEndEvent(1, 2200L))
+    listener.onApplicationEnd(new SparkListenerApplicationEnd(3100L))
+
+    expect:
+    assertTraces(1) {
+      trace(3) {
+        span {
+          operationName "spark.application"
+          spanType "spark"
+        }
+        span {
+          operationName "spark.job"
+          spanType "spark"
+          childOf(span(0))
+        }
+        span {
+          operationName "spark.stage"
+          spanType "spark"
+          childOf(span(1))
+        }
+      }
+    }
+  }
+
   def "compute task metrics histograms"() {
     setup:
     def listener = getTestDatadogSparkListener()

dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java

@@ -320,6 +320,7 @@ public final class ConfigDefaults {
   static final boolean DEFAULT_ELASTICSEARCH_BODY_AND_PARAMS_ENABLED = false;
 
   static final boolean DEFAULT_SPARK_TASK_HISTOGRAM_ENABLED = true;
+  static final boolean DEFAULT_SPARK_TASK_SPAN_ON_FAILURE_ENABLED = true;
   static final boolean DEFAULT_SPARK_APP_NAME_AS_SERVICE = false;
   static final boolean DEFAULT_JAX_RS_EXCEPTION_AS_ERROR_ENABLED = true;
   static final boolean DEFAULT_TELEMETRY_DEBUG_REQUESTS_ENABLED = false;

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

@@ -180,6 +180,8 @@ public final class TraceInstrumentationConfig {
       "trace.elasticsearch.body-and-params.enabled";
 
   public static final String SPARK_TASK_HISTOGRAM_ENABLED = "spark.task-histogram.enabled";
+  public static final String SPARK_TASK_SPAN_ON_FAILURE_ENABLED =
+      "spark.task-span-on-failure.enabled";
   public static final String SPARK_APP_NAME_AS_SERVICE = "spark.app-name-as-service";
 
   public static final String TRACE_WEBSOCKET_MESSAGES_ENABLED = "trace.websocket.messages.enabled";

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

@@ -153,6 +153,7 @@
 import static datadog.trace.api.ConfigDefaults.DEFAULT_SITE;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_SPARK_APP_NAME_AS_SERVICE;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_SPARK_TASK_HISTOGRAM_ENABLED;
+import static datadog.trace.api.ConfigDefaults.DEFAULT_SPARK_TASK_SPAN_ON_FAILURE_ENABLED;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_SSI_INJECTION_FORCE;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_STARTUP_LOGS_ENABLED;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_SYMBOL_DATABASE_COMPRESSED;
@@ -621,6 +622,7 @@
 import static datadog.trace.api.config.TraceInstrumentationConfig.SERVLET_ROOT_CONTEXT_SERVICE_NAME;
 import static datadog.trace.api.config.TraceInstrumentationConfig.SPARK_APP_NAME_AS_SERVICE;
 import static datadog.trace.api.config.TraceInstrumentationConfig.SPARK_TASK_HISTOGRAM_ENABLED;
+import static datadog.trace.api.config.TraceInstrumentationConfig.SPARK_TASK_SPAN_ON_FAILURE_ENABLED;
 import static datadog.trace.api.config.TraceInstrumentationConfig.SPRING_DATA_REPOSITORY_INTERFACE_RESOURCE_NAME;
 import static datadog.trace.api.config.TraceInstrumentationConfig.SQS_BODY_PROPAGATION_ENABLED;
 import static datadog.trace.api.config.TraceInstrumentationConfig.TRACE_128_BIT_TRACEID_LOGGING_ENABLED;
@@ -1379,6 +1381,7 @@ public static String getHostName() {
   private final boolean elasticsearchParamsEnabled;
   private final boolean elasticsearchBodyAndParamsEnabled;
   private final boolean sparkTaskHistogramEnabled;
+  private final boolean sparkTaskSpanOnFailureEnabled;
   private final boolean sparkAppNameAsService;
   private final boolean jaxRsExceptionAsErrorsEnabled;
   private final boolean websocketMessagesInheritSampling;
@@ -3152,6 +3155,10 @@ PROFILING_DATADOG_PROFILER_ENABLED, isDatadogProfilerSafeInCurrentEnvironment())
         configProvider.getBoolean(
             SPARK_TASK_HISTOGRAM_ENABLED, DEFAULT_SPARK_TASK_HISTOGRAM_ENABLED);
 
+    this.sparkTaskSpanOnFailureEnabled =
+        configProvider.getBoolean(
+            SPARK_TASK_SPAN_ON_FAILURE_ENABLED, DEFAULT_SPARK_TASK_SPAN_ON_FAILURE_ENABLED);
+
     this.sparkAppNameAsService =
         configProvider.getBoolean(SPARK_APP_NAME_AS_SERVICE, DEFAULT_SPARK_APP_NAME_AS_SERVICE);
 
@@ -5075,6 +5082,10 @@ public boolean isSparkTaskHistogramEnabled() {
     return sparkTaskHistogramEnabled;
   }
 
+  public boolean isSparkTaskSpanOnFailureEnabled() {
+    return sparkTaskSpanOnFailureEnabled;
+  }
+
   public boolean useSparkAppNameAsService() {
     return sparkAppNameAsService;
   }
@@ -6650,6 +6661,8 @@ public String toString() {
         + appLogsCollectionEnabled
         + ", sparkTaskHistogramEnabled="
         + sparkTaskHistogramEnabled
+        + ", sparkTaskSpanOnFailureEnabled="
+        + sparkTaskSpanOnFailureEnabled
         + ", sparkAppNameAsService="
         + sparkAppNameAsService
         + ", jaxRsExceptionAsErrorsEnabled="

metadata/supported-configurations.json

@@ -3865,6 +3865,14 @@
         "aliases": []
       }
     ],
+    "DD_SPARK_TASK_SPAN_ON_FAILURE_ENABLED": [
+      {
+        "version": "A",
+        "type": "boolean",
+        "default": "true",
+        "aliases": []
+      }
+    ],
     "DD_SPRING_DATA_REPOSITORY_INTERFACE_RESOURCE_NAME": [
       {
         "version": "A",