put custom tags into a single defined attribute

Author: pawel-big-lebowski

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

@@ -244,6 +244,9 @@ private void initApplicationSpanIfNotInitialized() {
     AgentTracer.SpanBuilder builder = buildSparkSpan("spark.application", null);
 
     if (applicationStart != null) {
+      Config.get().getGlobalTags().entrySet().stream()
+          .forEach(e -> builder.withTag("dd.tags." + e.getKey(), e.getValue()));
+
       builder
           .withStartTimestamp(applicationStart.time() * 1000)
           .withTag("application_name", applicationStart.appName())

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

@@ -588,6 +588,27 @@ abstract class AbstractSparkListenerTest extends InstrumentationSpecification {
     .contains("_dd.ol_intake.process_tags:" + ProcessTags.getTagsForSerialization())
   }
 
+  def "test all custom tags get propagated into a single attributes of application span"() {
+    setup:
+    def customTags = "tagKey1:tagKeyValue1,tagKey2:tagKeyValue2"
+    injectSysConfig("dd.tags", customTags)
+    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["dd.tags.tagKey1"] == "tagKeyValue1"
+          assert span.tags["dd.tags.tagKey2"] == "tagKeyValue2"
+        }
+      }
+    }
+  }
+
   def "test setupOpenLineage fills circuit breaker config"(
     Boolean configEnabled,
     String sparkConfCircuitBreakerType,