Tag dsm checkpoints with product mask

Author: kr-igor

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

@@ -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;
@@ -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>(
@@ -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,
@@ -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());
+
       PathwayHashBuilder pathwayHashBuilder =
           new PathwayHashBuilder(hashOfKnownTags, serviceNameOverride);
       DataSetHashBuilder aggregationHashBuilder = new DataSetHashBuilder();

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

@@ -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:");

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

@@ -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;
+  }
+
   public String getDataJobsCommandPattern() {
     return dataJobsCommandPattern;
   }