add dd_tags into djm.tags spark.application span attribute

Author: pawel-big-lebowski

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

@@ -36,6 +36,7 @@
 import java.util.Properties;
 import java.util.UUID;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 import org.apache.spark.ExceptionFailure;
 import org.apache.spark.SparkConf;
 import org.apache.spark.TaskFailedReason;
@@ -244,9 +245,16 @@ private void initApplicationSpanIfNotInitialized() {
     AgentTracer.SpanBuilder builder = buildSparkSpan("spark.application", null);
 
     if (applicationStart != null) {
+      String ddTags =
+          Config.get().getGlobalTags().entrySet().stream()
+              .sorted(Map.Entry.comparingByKey())
+              .map(e -> e.getKey() + ":" + e.getValue())
+              .collect(Collectors.joining(","));
+
       builder
           .withStartTimestamp(applicationStart.time() * 1000)
           .withTag("application_name", applicationStart.appName())
+          .withTag("djm.tags", ddTags)
           .withTag("spark_user", applicationStart.sparkUser());
 
       if (applicationStart.appAttemptId().isDefined()) {

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

@@ -588,6 +588,26 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     .contains("_dd.ol_intake.process_tags:" + ProcessTags.getTagsForSerialization())
   }
 
+  def "test DD_TAGS are put into a single attributes of application span"() {
+    setup:
+    def ddTags = "tagKey1:tagKeyValue1,tagKey2:tagKeyValue2"
+    injectSysConfig("dd.tags", ddTags)
+    def listener = getTestDatadogSparkListener()
+    listener.onApplicationStart(applicationStartEvent(1000L))
+    listener.onApplicationEnd(new SparkListenerApplicationEnd(5000L))
+
+    expect:
+    assertTraces(1) {
+      trace(1) {
+        span {
+          operationName "spark.application"
+          spanType "spark"
+          assert span.tags["djm.tags"] == ddTags
+        }
+      }
+    }
+  }
+
   def "test setupOpenLineage fills circuit breaker config"(
     Boolean configEnabled,
     String sparkConfCircuitBreakerType,