fix(trace sampler): correctness test for probabilistic sampler + fixes

Makefile

@@ -534,7 +534,7 @@ test-all: test test-property test-docs test-miri test-loom
 
 .PHONY: test-correctness
 test-correctness: ## Runs the complete correctness suite
-test-correctness: test-correctness-dsd-plain test-correctness-dsd-origin-detection test-correctness-otlp-metrics test-correctness-otlp-traces test-correctness-otlp-traces-ets test-correctness-otlp-traces-ottl-filtering test-correctness-otlp-traces-ottl-transform
+test-correctness: test-correctness-dsd-plain test-correctness-dsd-origin-detection test-correctness-otlp-metrics test-correctness-otlp-traces test-correctness-otlp-traces-ets test-correctness-otlp-traces-ottl-filtering test-correctness-otlp-traces-ottl-transform test-correctness-otlp-traces-probabilistic
 
 .PHONY: test-correctness-dsd-plain
 test-correctness-dsd-plain: build-ground-truth
@@ -578,6 +578,12 @@ test-correctness-otlp-traces-ottl-transform: ## Runs the 'otlp-traces-ottl-trans
 	@echo "[*] Running 'otlp-traces-ottl-transform' correctness test case..."
 	@target/release/ground-truth $(shell pwd)/test/correctness/otlp-traces-ottl-transform/config.yaml
 
+.PHONY: test-correctness-otlp-traces-probabilistic
+test-correctness-otlp-traces-probabilistic: build-ground-truth
+test-correctness-otlp-traces-probabilistic: ## Runs the 'otlp-traces-probabilistic' correctness test (probabilistic sampler at 50%)
+	@echo "[*] Running 'otlp-traces-probabilistic' correctness test case..."
+	@target/release/ground-truth $(shell pwd)/test/correctness/otlp-traces-probabilistic/config.yaml
+
 .PHONY: build-panoramic
 build-panoramic: check-rust-build-tools
 build-panoramic: ## Builds the panoramic binary (ADP integration test runner)

bin/correctness/ground-truth/src/analysis/traces/mod.rs

@@ -300,13 +300,15 @@ impl TracesAnalyzer {
 
         // Ensure that we observe at least one span on each side where Single Step Instrumentation-related metadata is
         // present (when not in OTLP-direct mode).
-        if !self.options.otlp_direct_analysis_mode && !self.baseline_ssi_metadata_present {
-            error!("No Single Step Instrumentation metadata found in baseline spans.");
-            error_count += 1;
-        }
-        if !self.comparison_ssi_metadata_present {
-            error!("No Single Step Instrumentation metadata found in comparison spans.");
-            error_count += 1;
+        if !self.options.otlp_direct_analysis_mode {
+            if !self.baseline_ssi_metadata_present {
+                error!("No Single Step Instrumentation metadata found in baseline spans.");
+                error_count += 1;
+            }
+            if !self.comparison_ssi_metadata_present {
+                error!("No Single Step Instrumentation metadata found in comparison spans.");
+                error_count += 1;
+            }
         }
 
         if error_count > 0 {

lib/saluki-components/src/transforms/trace_sampler/mod.rs

@@ -364,7 +364,7 @@ impl TraceSampler {
                 // Run probabilistic sampler - use root span's trace ID
                 let root_trace_id = trace.spans()[root_span_idx].trace_id();
                 if self.sample_probabilistic(root_trace_id) {
-                    decision_maker = DECISION_MAKER_PROBABILISTIC; // probabilistic sampling
+                    decision_maker = DECISION_MAKER_PROBABILISTIC;
                     prob_keep = true;
 
                     if let Some(root_span) = trace.spans_mut().get_mut(root_span_idx) {
@@ -464,7 +464,11 @@ impl TraceSampler {
         };
 
         let meta = root_span_value.meta_mut();
-        if priority > 0 {
+        // When the APM-level probabilistic sampler is used with OTLP traces, the DD Agent writes
+        // _dd.p.dm to trace chunk tags only (not span meta). For the legacy OTLP sampling path,
+        // it is written to both. We match that behavior by skipping the span meta write only when
+        // both conditions hold; the DM value still flows through TraceSampling to the encoder.
+        if priority > 0 && !(is_otlp && self.probabilistic_sampler_enabled) {
             if let Some(dm) = decision_maker_meta.as_ref() {
                 meta.insert(MetaString::from(TAG_DECISION_MAKER), dm.clone());
             }

test/correctness/otlp-traces-probabilistic/config.yaml

@@ -0,0 +1,24 @@
+analysis_mode: traces
+otlp_direct_analysis_mode: true
+millstone:
+  image: saluki-images/millstone:latest
+  config_path: millstone.yaml
+datadog_intake:
+  image: saluki-images/datadog-intake:latest
+  config_path: ../datadog-intake.yaml
+baseline:
+  image: saluki-images/datadog-agent:testing-release
+  files:
+    - datadog.yaml:/etc/datadog-agent/datadog.yaml
+  additional_env_vars:
+    - DD_API_KEY=correctness-test
+    - DD_APM_PROBABILISTIC_SAMPLER_ENABLED=true
+comparison:
+  image: saluki-images/datadog-agent:testing-release
+  files:
+    - datadog.yaml:/etc/datadog-agent/datadog.yaml
+  additional_env_vars:
+    - DD_API_KEY=correctness-test
+    - DD_DATA_PLANE_ENABLED=true
+    - DD_DATA_PLANE_OTLP_ENABLED=true
+    - DD_APM_PROBABILISTIC_SAMPLER_ENABLED=true

test/correctness/otlp-traces-probabilistic/datadog.yaml

@@ -0,0 +1,28 @@
+hostname: "correctness-testing"
+api_key: dummy-api-key-correctness-testing
+health_port: 5555
+log_level: debug
+
+process_config:
+  process_collection:
+    enabled: false
+  container_collection:
+    enabled: false
+
+dd_url: "http://datadog-intake:2049"
+
+apm_config:
+  enabled: true
+  apm_dd_url: "http://datadog-intake:2049"
+  features: ["enable_otlp_compute_top_level_by_span_kind"]
+  probabilistic_sampler:
+    enabled: true
+    sampling_percentage: 50
+
+otlp_config:
+  receiver:
+    protocols:
+      grpc:
+        endpoint: "0.0.0.0:4317"
+  traces:
+    enable_otlp_compute_top_level_by_span_kind: true

test/correctness/otlp-traces-probabilistic/millstone.yaml

@@ -0,0 +1,69 @@
+seed: [2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71, 73, 79, 83, 89, 97, 101, 103, 107, 109, 113, 127, 131]
+target: "grpc://target:4317/opentelemetry.proto.collector.trace.v1.TraceService/Export"
+aggregation_bucket_width_secs: 10
+volume: 1000
+corpus:
+  size: 1000
+  payload:
+    opentelemetry_traces:
+      error_rate: 0.01
+      services:
+      - name: api-gateway
+        service_type: http
+        scope_name: com.example.gateway
+        resource_attributes:
+        - key: deployment.environment
+          value: production
+        - key: cloud.region
+          value:
+            dictionary: cloud_regions
+        operations:
+        - id: get-product
+          method: GET
+          route: /api/v1/products/{id}
+          suboperations:
+          - to: product-service/get-product
+        - id: list-products
+          method: GET
+          route: /api/v1/products
+          suboperations:
+          - to: product-service/list-products
+      - name: product-service
+        service_type: grpc
+        grpc:
+          service: ProductService
+        scope_name: com.example.products
+        operations:
+        - id: get-product
+          method: GetProduct
+          suboperations:
+          - to: product-cache/get-product-by-id
+          - to: product-db/select-product-by-id
+            rate: 0.1
+        - id: list-products
+          method: ListProducts
+          suboperations:
+          - to: product-cache/get-products
+          - to: product-db/select-products
+            rate: 0.1
+      - name: product-cache
+        service_type: database
+        database:
+          system: redis
+        operations:
+        - id: get-product-by-id
+          query: GET products:by_id:$1
+        - id: get-products
+          query: GET products:full
+      - name: product-db
+        service_type: database
+        database:
+          system: postgresql
+          name: products
+        operations:
+        - id: select-product-by-id
+          table: products
+          query: SELECT * FROM products WHERE id = $1
+        - id: select-products
+          table: products
+          query: SELECT * FROM products LIMIT 50