Remove containerTagsHash/processTags from DSM primary pathway hash

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

@@ -515,7 +515,7 @@ void checkContextExtractorDecoratorBehavior(boolean dynamicConfigEnabled) throws
               payloadWriter,
               DEFAULT_BUCKET_DURATION_NANOS);
 
-      BaseHash.updateBaseHash(BASE_HASH);
+      BaseHash.updateIdentityHash(BASE_HASH);
       DefaultPathwayContext context = new DefaultPathwayContext(timeSource, null);
       timeSource.advance(MILLISECONDS.toNanos(50));
       context.setCheckpoint(
@@ -575,7 +575,7 @@ void checkContextExtractorDecoratorBehaviorWhenTraceDataIsNull(boolean globalDsm
               payloadWriter,
               DEFAULT_BUCKET_DURATION_NANOS);
 
-      BaseHash.updateBaseHash(BASE_HASH);
+      BaseHash.updateIdentityHash(BASE_HASH);
       DefaultPathwayContext context = new DefaultPathwayContext(timeSource, null);
       timeSource.advance(MILLISECONDS.toNanos(50));
       context.setCheckpoint(
@@ -635,7 +635,7 @@ void checkContextExtractorDecoratorBehaviorWhenLocalTraceConfigIsNull(boolean gl
               payloadWriter,
               DEFAULT_BUCKET_DURATION_NANOS);
 
-      BaseHash.updateBaseHash(BASE_HASH);
+      BaseHash.updateIdentityHash(BASE_HASH);
       DefaultPathwayContext context = new DefaultPathwayContext(timeSource, null);
       timeSource.advance(MILLISECONDS.toNanos(50));
       context.setCheckpoint(

internal-api/src/main/java/datadog/trace/api/BaseHash.java

@@ -7,14 +7,24 @@ public final class BaseHash {
   private static volatile String baseHashStr;
   private static volatile String lastContainerTagsHash;
 
+  // 0 means "not yet computed". service/env/primaryTag are fixed for the JVM's lifetime once
+  // Config is built, so this only needs to be calculated once, lazily, on first read - computing
+  // it eagerly in a field initializer would risk capturing a premature Config snapshot if this
+  // class gets loaded (e.g. via DataStreamsTags.EMPTY) before Config settles.
+  private static volatile long identityHash;
+
   private BaseHash() {}
 
   public static void recalcBaseHash(String containerTagsHash) {
     lastContainerTagsHash = containerTagsHash;
-    updateBaseHash(calc(containerTagsHash));
+    recalc();
   }
 
   static void recalcBaseHash() {
+    recalc();
+  }
+
+  private static void recalc() {
     updateBaseHash(calc(lastContainerTagsHash));
   }
 
@@ -31,6 +41,28 @@ public static String getBaseHashStr() {
     return baseHashStr;
   }
 
+  /**
+   * DSM topology-identity hash: service + env + primary tag only. Unlike {@link #getBaseHash()}
+   * (which also folds in process tags and the agent-reported container-tags hash for DBM's
+   * per-container SQL attribution use case), this is stable across pod restarts / rolling deploys,
+   * so it's safe to use as the seed for DSM's cardinality-sensitive pathway hash.
+   */
+  public static long getIdentityHash() {
+    if (identityHash == 0) {
+      // Deliberately unsynchronized: concurrent callers all recompute from the same Config, so a
+      // race just means a few redundant, identical writes rather than a correctness issue.
+      identityHash =
+          calcIdentity(
+              Config.get().getServiceName(), Config.get().getEnv(), Config.get().getPrimaryTag());
+    }
+    return identityHash;
+  }
+
+  /** Test-only: lets tests set the identity hash without going through {@link Config}. */
+  public static void updateIdentityHash(long hash) {
+    identityHash = hash;
+  }
+
   public static long calc(String containerTagsHash) {
     return calc(
         Config.get().getServiceName(),
@@ -40,15 +72,22 @@ public static long calc(String containerTagsHash) {
         containerTagsHash);
   }
 
+  static long calcIdentity(CharSequence serviceName, CharSequence env, String primaryTag) {
+    long hash = FNV64Hash.generateHash(serviceName.toString(), FNV64Hash.Version.v1);
+    hash = FNV64Hash.continueHash(hash, env.toString(), FNV64Hash.Version.v1);
+    if (primaryTag != null) {
+      hash = FNV64Hash.continueHash(hash, primaryTag, FNV64Hash.Version.v1);
+    }
+    return hash;
+  }
+
   private static long calc(
       CharSequence serviceName,
       CharSequence env,
       String primaryTag,
       CharSequence processTags,
       String containerTagsHash) {
-    long hash = FNV64Hash.generateHash(serviceName.toString(), FNV64Hash.Version.v1);
-    hash = FNV64Hash.continueHash(hash, env.toString(), FNV64Hash.Version.v1);
-    if (primaryTag != null) hash = FNV64Hash.continueHash(hash, primaryTag, FNV64Hash.Version.v1);
+    long hash = calcIdentity(serviceName, env, primaryTag);
     if (processTags != null) {
       hash = FNV64Hash.continueHash(hash, processTags.toString(), FNV64Hash.Version.v1);
       if (containerTagsHash != null && !containerTagsHash.isEmpty()) {

internal-api/src/main/java/datadog/trace/api/datastreams/DataStreamsTags.java

@@ -325,7 +325,10 @@ public DataStreamsTags(
         kafkaClusterId != null ? KAFKA_CLUSTER_ID_TAG + ":" + kafkaClusterId : null;
     this.partition = partition != null ? PARTITION_TAG + ":" + partition : null;
 
-    this.hash = BaseHash.getBaseHash();
+    // seeded from service+env+primaryTag only: process tags and the agent-reported
+    // container-tags hash vary per-pod/per-rollout and must not fragment pathway identity
+    // (see DSM2-335) — they're folded into aggregationHash below instead.
+    this.hash = BaseHash.getIdentityHash();
 
     if (DataStreamsTags.serviceNameOverride != null) {
       String val = DataStreamsTags.serviceNameOverride.get();
@@ -343,8 +346,15 @@ public DataStreamsTags(
       }
     }
 
-    // aggregation tags are 7-11: datasetName, datasetNamespace, isManual, group, consumerGroup
     this.aggregationHash = this.hash;
+
+    // DSM2-335: process tags and the agent-reported container-tags hash used to be folded in
+    // here (per-pod/per-process metadata inherited from DBM's BaseHash, see BaseHash.getBaseHash()
+    // and PR #9282). They're dropped entirely rather than moved to the aggregation tier: the
+    // backend never decodes them into discrete tags today, so they only added opaque cardinality
+    // with no user-visible benefit.
+
+    // aggregation tags are 7-11: datasetName, datasetNamespace, isManual, group, consumerGroup
     for (int i = 7; i < 12; i++) {
       String tag = this.tagByIndex(i);
       if (tag != null) {

internal-api/src/test/groovy/datadog/trace/api/datastreams/DataStreamsTagsTest.groovy

@@ -1,11 +1,18 @@
 package datadog.trace.api.datastreams
 
 import datadog.trace.api.BaseHash
+import datadog.trace.api.Config
+import datadog.trace.api.ProcessTags
 import spock.lang.Specification
 import java.nio.ByteBuffer
 
 
 class DataStreamsTagsTest extends Specification {
+  def cleanup() {
+    BaseHash.recalcBaseHash(null)
+    ProcessTags.reset(Config.get())
+  }
+
   def getTags(int idx) {
     return new DataStreamsTags("bus" + idx, DataStreamsTags.Direction.OUTBOUND, "exchange" + idx, "topic" + idx, "type" + idx, "subscription" + idx,
       "dataset_name" + idx, "dataset_namespace" + idx, true, "group" + idx, "consumer_group" + idx, true,
@@ -80,7 +87,7 @@ class DataStreamsTagsTest extends Specification {
     DataStreamsTags.setServiceNameOverride(serviceName)
     def two = getTags(0)
 
-    BaseHash.updateBaseHash(12)
+    BaseHash.updateIdentityHash(12)
     def three = getTags(0)
 
     expect:

internal-api/src/test/java/datadog/trace/api/BaseHashIdentityTest.java

@@ -0,0 +1,67 @@
+package datadog.trace.api;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+import datadog.trace.test.junit.utils.config.WithConfigExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(WithConfigExtension.class)
+class BaseHashIdentityTest {
+
+  @BeforeEach
+  void setup() {
+    ProcessTags.reset();
+  }
+
+  @AfterEach
+  void cleanup() {
+    ProcessTags.reset();
+  }
+
+  @Test
+  void identityHashDependsOnServiceEnvAndPrimaryTag() {
+    // identityHash is calculated once (service/env/primaryTag are fixed for the JVM's
+    // lifetime), so this exercises the underlying hashing function directly rather than
+    // via Config + recalcBaseHash.
+    long base = BaseHash.calcIdentity("service", "env", "region-1");
+
+    assertNotEquals(base, BaseHash.calcIdentity("service-2", "env", "region-1"));
+    assertNotEquals(base, BaseHash.calcIdentity("service", "env-2", "region-1"));
+    assertNotEquals(base, BaseHash.calcIdentity("service", "env", "region-2"));
+    assertEquals(base, BaseHash.calcIdentity("service", "env", "region-1"));
+  }
+
+  @Test
+  void identityHashIsUnaffectedByContainerTagsHashOrProcessTags() {
+    BaseHash.recalcBaseHash(null);
+    long baseIdentityHash = BaseHash.getIdentityHash();
+
+    BaseHash.recalcBaseHash("some-container-tags-hash");
+    long withContainerTagsHash = BaseHash.getIdentityHash();
+
+    ProcessTags.addTag("foo", "bar");
+    long withProcessTags = BaseHash.getIdentityHash();
+
+    // DSM2-335: identity hash must not be perturbed by per-pod/per-rollout inputs
+    assertEquals(baseIdentityHash, withContainerTagsHash);
+    assertEquals(baseIdentityHash, withProcessTags);
+  }
+
+  @Test
+  void getIdentityHashRecalculatesFromConfigWhenUnset() {
+    // 0 means "not yet computed" - simulate that state, e.g. before this class is ever touched
+    BaseHash.updateIdentityHash(0L);
+
+    long recalculated = BaseHash.getIdentityHash();
+    assertNotEquals(0L, recalculated);
+
+    // once non-zero, later reads don't recalculate - a caller that needs a fresh value can
+    // still force one directly (e.g. tests via updateIdentityHash)
+    BaseHash.updateIdentityHash(42L);
+    assertEquals(42L, BaseHash.getIdentityHash());
+  }
+}

internal-api/src/test/java/datadog/trace/api/datastreams/DataStreamsTagsContainerProcessTagsTest.java

@@ -0,0 +1,67 @@
+package datadog.trace.api.datastreams;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import datadog.trace.api.BaseHash;
+import datadog.trace.api.Config;
+import datadog.trace.api.ProcessTags;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+class DataStreamsTagsContainerProcessTagsTest {
+
+  @AfterEach
+  void cleanup() {
+    BaseHash.recalcBaseHash(null);
+    ProcessTags.reset(Config.get());
+  }
+
+  private static DataStreamsTags getTags(int idx) {
+    return new DataStreamsTags(
+        "bus" + idx,
+        DataStreamsTags.Direction.OUTBOUND,
+        "exchange" + idx,
+        "topic" + idx,
+        "type" + idx,
+        "subscription" + idx,
+        "dataset_name" + idx,
+        "dataset_namespace" + idx,
+        true,
+        "group" + idx,
+        "consumer_group" + idx,
+        true,
+        "kafka_cluster_id" + idx,
+        "partition" + idx);
+  }
+
+  @Test
+  void containerTagsHashDoesNotAffectAnyHashTier() {
+    // simulate the Agent reporting the pod/container's tags hash at startup
+    BaseHash.recalcBaseHash("container-tags-hash-1");
+    DataStreamsTags base = getTags(0);
+
+    // a rolling deploy changes the container-tags hash the Agent reports
+    BaseHash.recalcBaseHash("container-tags-hash-2");
+    DataStreamsTags afterRollingDeploy = getTags(0);
+
+    // DSM2-335: no hash tier is affected - container-tags hash is dropped entirely from DSM
+    assertEquals(base.getHash(), afterRollingDeploy.getHash());
+    assertEquals(base.getAggregationHash(), afterRollingDeploy.getAggregationHash());
+    assertEquals(base, afterRollingDeploy);
+  }
+
+  @Test
+  void processTagsDoNotAffectAnyHashTier() {
+    BaseHash.recalcBaseHash(null);
+    DataStreamsTags base = getTags(0);
+
+    // a process tag is added (e.g. cluster.name discovered after startup)
+    ProcessTags.addTag("cluster.name", "new-cluster");
+    DataStreamsTags withProcessTag = getTags(0);
+
+    // DSM2-335: no hash tier is affected - process tags are dropped entirely from DSM
+    assertEquals(base.getHash(), withProcessTag.getHash());
+    assertEquals(base.getAggregationHash(), withProcessTag.getAggregationHash());
+    assertEquals(base, withProcessTag);
+  }
+}