diff --git a/dd-java-agent/instrumentation/spark/spark-common/src/main/java/datadog/trace/instrumentation/spark/AbstractDatadogSparkListener.java b/dd-java-agent/instrumentation/spark/spark-common/src/main/java/datadog/trace/instrumentation/spark/AbstractDatadogSparkListener.java index 2f033394a53..02aeb77d4f6 100644 --- a/dd-java-agent/instrumentation/spark/spark-common/src/main/java/datadog/trace/instrumentation/spark/AbstractDatadogSparkListener.java +++ b/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; diff --git a/dd-java-agent/instrumentation/spark/spark-common/src/testFixtures/groovy/datadog/trace/instrumentation/spark/AbstractSparkListenerTest.groovy b/dd-java-agent/instrumentation/spark/spark-common/src/testFixtures/groovy/datadog/trace/instrumentation/spark/AbstractSparkListenerTest.groovy index a5ebb71915d..dcab8305102 100644 --- a/dd-java-agent/instrumentation/spark/spark-common/src/testFixtures/groovy/datadog/trace/instrumentation/spark/AbstractSparkListenerTest.groovy +++ b/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() diff --git a/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java b/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java index 1aba2cb276b..d07b0362cc6 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java +++ b/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; diff --git a/dd-trace-api/src/main/java/datadog/trace/api/config/TraceInstrumentationConfig.java b/dd-trace-api/src/main/java/datadog/trace/api/config/TraceInstrumentationConfig.java index 6d44cf8b249..b16433dd4d5 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/config/TraceInstrumentationConfig.java +++ b/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"; diff --git a/internal-api/src/main/java/datadog/trace/api/Config.java b/internal-api/src/main/java/datadog/trace/api/Config.java index 5ced61622ca..0cc17c4c198 100644 --- a/internal-api/src/main/java/datadog/trace/api/Config.java +++ b/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=" diff --git a/metadata/supported-configurations.json b/metadata/supported-configurations.json index 733f3736cbb..38491997e0d 100644 --- a/metadata/supported-configurations.json +++ b/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",