diff --git a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultPathwayContextTest.java b/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultPathwayContextTest.java index 8860cf1571c..576ae28165c 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultPathwayContextTest.java +++ b/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( diff --git a/internal-api/src/main/java/datadog/trace/api/BaseHash.java b/internal-api/src/main/java/datadog/trace/api/BaseHash.java index 8aa17c4a0ff..73910b8867f 100644 --- a/internal-api/src/main/java/datadog/trace/api/BaseHash.java +++ b/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()) { diff --git a/internal-api/src/main/java/datadog/trace/api/datastreams/DataStreamsTags.java b/internal-api/src/main/java/datadog/trace/api/datastreams/DataStreamsTags.java index df3eb214813..fccc9a407e1 100644 --- a/internal-api/src/main/java/datadog/trace/api/datastreams/DataStreamsTags.java +++ b/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) { diff --git a/internal-api/src/test/groovy/datadog/trace/api/datastreams/DataStreamsTagsTest.groovy b/internal-api/src/test/groovy/datadog/trace/api/datastreams/DataStreamsTagsTest.groovy index aa26cb9ebbb..16dc93ad321 100644 --- a/internal-api/src/test/groovy/datadog/trace/api/datastreams/DataStreamsTagsTest.groovy +++ b/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: diff --git a/internal-api/src/test/java/datadog/trace/api/BaseHashIdentityTest.java b/internal-api/src/test/java/datadog/trace/api/BaseHashIdentityTest.java new file mode 100644 index 00000000000..4ed33972f9a --- /dev/null +++ b/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()); + } +} diff --git a/internal-api/src/test/java/datadog/trace/api/datastreams/DataStreamsTagsContainerProcessTagsTest.java b/internal-api/src/test/java/datadog/trace/api/datastreams/DataStreamsTagsContainerProcessTagsTest.java new file mode 100644 index 00000000000..cc01a741ce1 --- /dev/null +++ b/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); + } +}