Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add new tag for enabled products / features to DSM checkpoints #8051

Merged
merged 10 commits into from
Dec 6, 2024
Merged

Add new tag for enabled products / features to DSM checkpoints #8051

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
6341a9b
Tag dsm checkpoints with product mask
kr-igor Dec 3, 2024
3fc206e
Merge branch 'master' into kr-igor/dsm-product-tags
kr-igor Dec 3, 2024
ea77f9c
Fixed failing tests
kr-igor Dec 3, 2024
b934995
Merge branch 'master' into kr-igor/dsm-product-tags
kr-igor Dec 4, 2024
6bb98df
Moved product mask logic from the config to DSM context
kr-igor Dec 4, 2024
408bde4
Moved product mask to DSM payload level
kr-igor Dec 4, 2024
0d0c395
Cleanup
kr-igor Dec 4, 2024
36168c3
Fixed serialization
kr-igor Dec 4, 2024
d0c8837
Fixed tests
kr-igor Dec 4, 2024
928343c
Merge branch 'master' into kr-igor/dsm-product-tags
kr-igor Dec 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultPathwayContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package datadog.trace.core.datastreams;

import static datadog.trace.core.datastreams.TagsProcessor.PRODUCTS_MASK;
import static java.nio.charset.StandardCharsets.ISO_8859_1;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
Expand Down Expand Up @@ -52,6 +53,7 @@ public class DefaultPathwayContext implements PathwayContext {
// direction != current direction
private long closestOppositeDirectionHash;
private String previousDirection;
private String productMaskTag;

private static final Set<String> hashableTagKeys =
new HashSet<String>(
Expand Down Expand Up @@ -117,6 +119,15 @@ public void setCheckpoint(
setCheckpoint(sortedTags, pointConsumer, defaultTimestamp, 0);
}

private String getProductMaskTag() {
// it's fine to cache the value per context
if (productMaskTag == null) {
productMaskTag = PRODUCTS_MASK + ":" + Config.get().enabledProductsMask();
}

return productMaskTag;
}

@Override
public void setCheckpoint(
LinkedHashMap<String, String> sortedTags,
Expand All @@ -129,7 +140,9 @@ public void setCheckpoint(
try {
// So far, each tag key has only one tag value, so we're initializing the capacity to match
// the number of tag keys for now. We should revisit this later if it's no longer the case.
List<String> allTags = new ArrayList<>(sortedTags.size());
List<String> allTags = new ArrayList<>(sortedTags.size() + 1);
allTags.add(getProductMaskTag());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you think of putting that in the "hash of known tags"?
That means we need to update the hashOfKnownsTags every time we enable / disable data streams via remote configuration, but it would give a boost in performance.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That will add one more field to hash, so I'm not sure about performance gains (the check for enabled products just checks a few variables). Also changing the configuration will change pathway hashes if we hash using this tag. Not critical, but it will create duplicating pathways during the overlapping time period.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. Let's just not use it for hashing.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you think about putting this mask as a new field on the message pack? It could be a top level field on the payload instead of being set on each checkpoint.
So we could call the config right before writing the message pack in the writer.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good idea! Moved the logic


PathwayHashBuilder pathwayHashBuilder =
new PathwayHashBuilder(hashOfKnownTags, serviceNameOverride);
DataSetHashBuilder aggregationHashBuilder = new DataSetHashBuilder();
Expand Down
1 change: 1 addition & 0 deletions dd-trace-core/src/main/java/datadog/trace/core/datastreams/TagsProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ public String apply(String key) {
}

public static final String MANUAL_TAG = "manual_checkpoint";
public static final String PRODUCTS_MASK = "products_mask";
public static final String TYPE_TAG = "type";
private static final DDCache<String, String> TYPE_TAG_CACHE = DDCaches.newFixedSizeCache(32);
private static final Function<String, String> TYPE_TAG_PREFIX = new StringPrefix("type:");
Expand Down
74 changes: 37 additions & 37 deletions ...race-core/src/test/groovy/datadog/trace/core/datastreams/DefaultPathwayContextTest.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,8 +74,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
pointConsumer.points.size() == 2
verifyFirstPoint(pointConsumer.points[0])
with(pointConsumer.points[1]) {
edgeTags == ["group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == 25
Expand All @@ -97,8 +97,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
context.isStarted()
pointConsumer.points.size() == 1
with(pointConsumer.points[0]) {
edgeTags == ["group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
hash != 0
payloadSizeBytes == 72
}
Expand All @@ -124,16 +124,16 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
pointConsumer.points.size() == 3
verifyFirstPoint(pointConsumer.points[0])
with(pointConsumer.points[1]) {
edgeTags == ["direction:in", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
edgeTags == ["products_mask:1", "direction:in", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 5
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == 25
edgeLatencyNano == 25
}
with(pointConsumer.points[2]) {
edgeTags == ["direction:in", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
edgeTags == ["products_mask:1", "direction:in", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 5
// this point should have the first point as parent,
// as the loop protection will reset the parent if two identical
// points (same hash for tag values) are about to form a hierarchy
Expand Down Expand Up @@ -201,8 +201,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
pointConsumer.points.size() == 2

with(pointConsumer.points[1]) {
edgeTags == ["group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(27)
Expand All @@ -221,8 +221,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
context.isStarted()
pointConsumer.points.size() == 1
with(pointConsumer.points[0]) {
edgeTags == ["type:internal"]
edgeTags.size() == 1
edgeTags == ["products_mask:1", "type:internal"]
edgeTags.size() == 2
parentHash == 0
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(200)
Expand Down Expand Up @@ -250,8 +250,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
decodedContext.isStarted()
pointConsumer.points.size() == 2
with(pointConsumer.points[1]) {
edgeTags == ["group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(26)
Expand All @@ -269,8 +269,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
secondDecode.isStarted()
pointConsumer.points.size() == 3
with(pointConsumer.points[2]) {
edgeTags == ["group:group", "topic:topicB", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topicB", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[1].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(58)
Expand Down Expand Up @@ -300,8 +300,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
decodedContext.isStarted()
pointConsumer.points.size() == 2
with(pointConsumer.points[1]) {
edgeTags == ["group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(26)
Expand All @@ -320,8 +320,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
secondDecode.isStarted()
pointConsumer.points.size() == 3
with(pointConsumer.points[2]) {
edgeTags == ["group:group", "topic:topicB", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topicB", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[1].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(58)
Expand Down Expand Up @@ -349,8 +349,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
pointConsumer.points.size() == 2

with(pointConsumer.points[1]) {
edgeTags == ["group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(27)
Expand Down Expand Up @@ -378,8 +378,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
decodedContext.isStarted()
pointConsumer.points.size() == 2
with(pointConsumer.points[1]) {
edgeTags == ["group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(26)
Expand All @@ -397,8 +397,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
secondDecode.isStarted()
pointConsumer.points.size() == 3
with(pointConsumer.points[2]) {
edgeTags == ["group:group", "topic:topicB", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topicB", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[1].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(58)
Expand Down Expand Up @@ -447,8 +447,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
decodedContext.isStarted()
pointConsumer.points.size() == 2
with(pointConsumer.points[1]) {
edgeTags == ["group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topic", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(26)
Expand All @@ -467,8 +467,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
secondDecode.isStarted()
pointConsumer.points.size() == 3
with(pointConsumer.points[2]) {
edgeTags == ["group:group", "topic:topicB", "type:kafka"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topicB", "type:kafka"]
edgeTags.size() == 4
parentHash == pointConsumer.points[1].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(58)
Expand Down Expand Up @@ -498,8 +498,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
decodedContext.isStarted()
pointConsumer.points.size() == 2
with(pointConsumer.points[1]) {
edgeTags == ["topic:topic", "type:sqs"]
edgeTags.size() == 2
edgeTags == ["products_mask:1", "topic:topic", "type:sqs"]
edgeTags.size() == 3
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(26)
Expand All @@ -518,8 +518,8 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
secondDecode.isStarted()
pointConsumer.points.size() == 3
with(pointConsumer.points[2]) {
edgeTags == ["topic:topicB", "type:sqs"]
edgeTags.size() == 2
edgeTags == ["products_mask:1", "topic:topicB", "type:sqs"]
edgeTags.size() == 3
parentHash == pointConsumer.points[1].hash
hash != 0
pathwayLatencyNano == MILLISECONDS.toNanos(58)
Expand All @@ -545,15 +545,15 @@ class DefaultPathwayContextTest extends DDCoreSpecification {
pointConsumer.points.size() == 3
verifyFirstPoint(pointConsumer.points[0])
with(pointConsumer.points[1]) {
edgeTags == ["group:group", "topic:topic", "type:type"]
edgeTags.size() == 3
edgeTags == ["products_mask:1", "group:group", "topic:topic", "type:type"]
edgeTags.size() == 4
parentHash == pointConsumer.points[0].hash
hash != 0
pathwayLatencyNano == 25
edgeLatencyNano == 25
}
with(pointConsumer.points[2]) {
edgeTags.size() == 0
edgeTags.size() == 1
parentHash == pointConsumer.points[1].hash
hash != 0
pathwayLatencyNano == 50
Expand Down
22 changes: 22 additions & 0 deletions internal-api/src/main/java/datadog/trace/api/Config.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3244,6 +3244,28 @@ public boolean isDataJobsEnabled() {
return dataJobsEnabled;
}

private static final int APM_PRODUCT = 1; // 00000001
private static final int DSM_PRODUCT = 2; // 00000010
private static final int DJM_PRODUCT = 4; // 00000100
private static final int PROFILING_PRODUCT = 8; // 00001000

// enabledProductsMask can be extended as needed
public long enabledProductsMask() {
long enabledProducts = APM_PRODUCT;

if (isDataStreamsEnabled()) {
enabledProducts |= DSM_PRODUCT;
}
if (isDataJobsEnabled()) {
enabledProducts |= DJM_PRODUCT;
}
if (isProfilingEnabled()) {
enabledProducts |= PROFILING_PRODUCT;
}

return enabledProducts;
}
PerfectSlayer marked this conversation as resolved.
Show resolved Hide resolved

public String getDataJobsCommandPattern() {
return dataJobsCommandPattern;
}
Expand Down
Loading