Fixed more tests

Author: kr-igor

dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultPathwayContext.java

@@ -109,19 +109,12 @@ public synchronized void setCheckpoint(
     }
 
     // generate node hash
-    long nodeHash = hashOfKnownTags;
-    if (serviceNameOverride != null) {
-      nodeHash = FNV64Hash.continueHash(nodeHash, serviceNameOverride, FNV64Hash.Version.v1);
-    }
-    nodeHash =
-        FNV64Hash.continueHash(
-            nodeHash, DataStreamsTags.longToBytes(context.tags().getHash()), FNV64Hash.Version.v1);
-
+    long nodeHash = context.tags().getHash();
     // loop protection - a node should not be chosen as parent
     // for a sequential node with the same direction, as this
     // will cause a `cardinality explosion` for hash / parentHash tag values
     DataStreamsTags.Direction direction = context.tags().getDirectionValue();
-    if (direction == previousDirection) {
+    if (direction == previousDirection && previousDirection != null) {
       hash = closestOppositeDirectionHash;
     } else {
       previousDirection = direction;

dd-trace-core/src/test/groovy/datadog/trace/core/datastreams/DefaultPathwayContextTest.groovy

@@ -7,13 +7,15 @@ import datadog.trace.api.ProcessTags
 import datadog.trace.api.TagMap
 import datadog.trace.api.TraceConfig
 import datadog.trace.api.WellKnownTags
+import datadog.trace.api.config.GeneralConfig
 import datadog.trace.api.datastreams.DataStreamsTags
 import datadog.trace.api.datastreams.StatsPoint
 import datadog.trace.api.time.ControllableTimeSource
 import datadog.trace.bootstrap.instrumentation.api.AgentPropagation
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan
 import datadog.trace.bootstrap.instrumentation.api.AgentTracer
 import datadog.trace.common.metrics.Sink
+import datadog.trace.core.CoreTracer
 import datadog.trace.core.propagation.ExtractedContext
 import datadog.trace.core.test.DDCoreSpecification
 
@@ -502,9 +504,10 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
 
     def dataStreams = new DefaultDataStreamsMonitoring(sink, features, timeSource, { globalTraceConfig }, wellKnownTags, payloadWriter, DEFAULT_BUCKET_DURATION_NANOS)
 
+    DataStreamsTags.setGlobalBaseHash(baseHash)
     def context = new DefaultPathwayContext(timeSource, baseHash, null)
     timeSource.advance(MILLISECONDS.toNanos(50))
-    context.setCheckpoint(fromTags(DataStreamsTags.create("itnernal", DataStreamsTags.Direction.Inbound)), pointConsumer)
+    context.setCheckpoint(fromTags(DataStreamsTags.create("internal", DataStreamsTags.Direction.Inbound)), pointConsumer)
     def encoded = context.encode()
     Map<String, String> carrier = [
       (PROPAGATION_KEY_BASE64): encoded,
@@ -518,6 +521,7 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
     def extractedSpan = AgentSpan.fromContext(extractedContext)
 
     then:
+    encoded == "L+lDG/Pa9hRkZA=="
     !dynamicConfigEnabled || extractedSpan != null
     if (dynamicConfigEnabled) {
       def extracted = extractedSpan.context()
@@ -546,8 +550,15 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
       isDataStreamsEnabled() >> { return globalDsmEnabled }
     }
 
+    def tracerApi = Mock(AgentTracer.TracerAPI) {
+      captureTraceConfig() >> globalTraceConfig
+    }
+    AgentTracer.TracerAPI originalTracer = AgentTracer.get()
+    AgentTracer.forceRegister(tracerApi)
+
     def dataStreams = new DefaultDataStreamsMonitoring(sink, features, timeSource, { globalTraceConfig }, wellKnownTags, payloadWriter, DEFAULT_BUCKET_DURATION_NANOS)
 
+    DataStreamsTags.setGlobalBaseHash(baseHash)
     def context = new DefaultPathwayContext(timeSource, baseHash, null)
     timeSource.advance(MILLISECONDS.toNanos(50))
     context.setCheckpoint(fromTags(DataStreamsTags.create("internal", DataStreamsTags.Direction.Inbound)), pointConsumer)
@@ -562,6 +573,7 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
     def extractedSpan = AgentSpan.fromContext(extractedContext)
 
     then:
+    encoded == "L+lDG/Pa9hRkZA=="
     if (globalDsmEnabled) {
       extractedSpan != null
       def extracted = extractedSpan.context()
@@ -572,6 +584,9 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
       extractedSpan == null
     }
 
+    cleanup:
+    AgentTracer.forceRegister(originalTracer)
+
     where:
     globalDsmEnabled << [true, false]
   }
@@ -589,19 +604,27 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
       isDataStreamsEnabled() >> { return globalDsmEnabled }
     }
 
-    def dataStreams = new DefaultDataStreamsMonitoring(sink, features, timeSource, { globalTraceConfig }, wellKnownTags, payloadWriter, DEFAULT_BUCKET_DURATION_NANOS)
+    def tracerApi = Mock(AgentTracer.TracerAPI) {
+      captureTraceConfig() >> globalTraceConfig
+    }
+    AgentTracer.TracerAPI originalTracer = AgentTracer.get()
+    AgentTracer.forceRegister(tracerApi)
+
+    def dataStreams = new DefaultDataStreamsMonitoring(sink, features, timeSource, { globalTraceConfig },
+      wellKnownTags, payloadWriter, DEFAULT_BUCKET_DURATION_NANOS)
 
+    DataStreamsTags.setGlobalBaseHash(baseHash)
     def context = new DefaultPathwayContext(timeSource, baseHash, null)
     timeSource.advance(MILLISECONDS.toNanos(50))
-    context.setCheckpoint(fromTags(DataStreamsTags.create("internal", null)), pointConsumer)
+    context.setCheckpoint(fromTags(DataStreamsTags.create("internal", DataStreamsTags.Direction.Inbound)), pointConsumer)
     def encoded = context.encode()
     Map<String, String> carrier = [(PROPAGATION_KEY_BASE64): encoded, "someotherkey": "someothervalue"]
     def contextVisitor = new Base64MapContextVisitor()
-    def spanContext = new ExtractedContext(DDTraceId.ONE, 1, 0, null, 0, null, (TagMap)null, null, null, null, DATADOG)
+    def spanContext = new ExtractedContext(DDTraceId.ONE, 1, 0, null, 0,
+      null, (TagMap)null, null, null, globalTraceConfig, DATADOG)
     def baseContext = AgentSpan.fromSpanContext(spanContext).storeInto(root())
     def propagator = dataStreams.propagator()
 
-
     when:
     def extractedContext = propagator.extract(baseContext, carrier, contextVisitor)
     def extractedSpan = AgentSpan.fromContext(extractedContext)
@@ -614,13 +637,17 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
 
     then:
     extracted != null
+    encoded == "L+lDG/Pa9hRkZA=="
     if (globalDsmEnabled) {
       extracted.pathwayContext != null
       extracted.pathwayContext.isStarted()
     } else {
       extracted.pathwayContext == null
     }
 
+    cleanup:
+    AgentTracer.forceRegister(originalTracer)
+
     where:
     globalDsmEnabled << [true, false]
   }
@@ -661,4 +688,4 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
       }
     }
   }
-}
+}

dd-trace-core/src/traceAgentTest/groovy/DataStreamsIntegrationTest.groovy

@@ -3,6 +3,7 @@ import datadog.communication.ddagent.SharedCommunicationObjects
 import datadog.communication.http.OkHttpUtils
 import datadog.trace.api.Config
 import datadog.trace.api.TraceConfig
+import datadog.trace.api.datastreams.DataStreamsTags
 import datadog.trace.api.time.ControllableTimeSource
 import datadog.trace.api.datastreams.StatsPoint
 import datadog.trace.common.metrics.EventListener
@@ -46,12 +47,8 @@ class DataStreamsIntegrationTest extends AbstractTraceAgentTest {
     when:
     def dataStreams = new DefaultDataStreamsMonitoring(sink, sharedCommunicationObjects.featuresDiscovery(Config.get()), timeSource, { traceConfig }, Config.get())
     dataStreams.start()
-    def tags = DataStreamsTags
-      .withTopic("testTopic")
-      .withGroup("testGroup")
-      .withType("testType")
-      .build()
-    dataStreams.add(new StatsPoint(tags, 1, 2, 5, timeSource.currentTimeNanos, 0, 0, 0, null))
+    def tg = DataStreamsTags.create("testType", null, "testTopic", "testGroup", null)
+    dataStreams.add(new StatsPoint(tg, 1, 2, 5, timeSource.currentTimeNanos, 0, 0, 0, null))
     timeSource.advance(Config.get().getDataStreamsBucketDurationNanoseconds())
     dataStreams.report()