Add spark.openlineage.appName tag to spark.application spans (#11246)

Add spark.openlineage.appName tag to spark.application spans

Test spark.openlineage.appName tag on spark.application spans

Merge branch 'master' into adrien.boitreaud/ol-jobname

Co-authored-by: adrien.boitreaud <adrien.boitreaud@datadoghq.com>

Author: aboitreaud

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

@@ -284,6 +284,7 @@ private void initApplicationSpanIfNotInitialized() {
 
     captureApplicationParameters(builder);
     captureEmrStepId(builder);
+    captureOpenlineageJobInfo(builder);
 
     Optional<OpenlineageParentContext> openlineageParentContext =
         OpenlineageParentContext.from(sparkConf);
@@ -299,6 +300,13 @@ private void initApplicationSpanIfNotInitialized() {
     applicationSpan.setMeasured(true);
   }
 
+  private void captureOpenlineageJobInfo(AgentTracer.SpanBuilder builder) {
+    String olAppName = sparkConf.get("spark.openlineage.appName", null);
+    if (olAppName != null) {
+      builder.withTag("spark.openlineage.appName", olAppName);
+    }
+  }
+
   private void captureOpenlineageContextIfPresent(
       AgentTracer.SpanBuilder builder, OpenlineageParentContext context) {
     builder.asChildOf(context);

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

@@ -721,6 +721,48 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     .getOption("spark.openlineage.circuitBreaker.timeoutInSeconds") == Option.apply("120")
   }
 
+  def "sets spark.openlineage.appName tag when configured"() {
+    setup:
+    def conf = new SparkConf()
+    conf.set("spark.openlineage.appName", "my-ol-app-name")
+    def listener = getTestDatadogSparkListener(conf)
+
+    when:
+    listener.onApplicationStart(applicationStartEvent(1000L))
+    listener.onApplicationEnd(new SparkListenerApplicationEnd(2000L))
+
+    then:
+    assertTraces(1) {
+      trace(1) {
+        span {
+          operationName "spark.application"
+          spanType "spark"
+          assert span.tags["spark.openlineage.appName"] == "my-ol-app-name"
+        }
+      }
+    }
+  }
+
+  def "does not set spark.openlineage.appName tag when not configured"() {
+    setup:
+    def listener = getTestDatadogSparkListener()
+
+    when:
+    listener.onApplicationStart(applicationStartEvent(1000L))
+    listener.onApplicationEnd(new SparkListenerApplicationEnd(2000L))
+
+    then:
+    assertTraces(1) {
+      trace(1) {
+        span {
+          operationName "spark.application"
+          spanType "spark"
+          assert !span.tags.containsKey("spark.openlineage.appName")
+        }
+      }
+    }
+  }
+
   protected validateRelativeError(double value, double expected, double relativeAccuracy) {
     double relativeError = Math.abs(value - expected) / expected
     assert relativeError < relativeAccuracy