fix(correctness): decode V3 metric resources

Author: rayz

bin/correctness/stele/src/metrics.rs

@@ -411,13 +411,21 @@ impl Metric {
         let names_dict = parse_dict_strings(&data.dictNameStr)?;
         let tags_dict = parse_dict_strings(&data.dictTagStr)?;
         let tagsets_dict = parse_tagsets(&data.dictTagsets, &tags_dict)?;
+        let resources_dict = parse_resources(
+            &data.dictResourceLen,
+            &data.dictResourceType,
+            &data.dictResourceName,
+            &parse_dict_strings(&data.dictResourceStr)?,
+        )?;
 
         // Delta-decode index arrays.
         let mut name_refs = data.nameRefs;
         let mut tagset_refs = data.tagsetRefs;
+        let mut resources_refs = data.resourcesRefs;
         let mut timestamps = data.timestamps;
         delta_decode(&mut name_refs);
         delta_decode(&mut tagset_refs);
+        delta_decode(&mut resources_refs);
         delta_decode(&mut timestamps);
 
         // Delta-decode sketch bin keys (per-sketch sequences are individually delta-encoded,
@@ -446,7 +454,7 @@ impl Metric {
 
             // Resolve tags (1-based index).
             let tagset_ref = tagset_refs[i] as usize;
-            let tags = if tagset_ref == 0 {
+            let mut tags = if tagset_ref == 0 {
                 Vec::new()
             } else {
                 tagsets_dict
@@ -457,6 +465,23 @@ impl Metric {
                     .clone()
             };
 
+            let resource_ref = resources_refs.get(i).copied().unwrap_or(0) as usize;
+            if resource_ref != 0 {
+                let resources = resources_dict.get(resource_ref - 1).ok_or_else(|| {
+                    generic_error!(
+                        "Invalid resource ref {} (dict size {})",
+                        resource_ref,
+                        resources_dict.len()
+                    )
+                })?;
+                if let Some((_, host_name)) = resources
+                    .iter()
+                    .find(|(resource_type, resource_name)| resource_type == "host" && !resource_name.is_empty())
+                {
+                    tags.push(format!("host:{}", host_name));
+                }
+            }
+
             let mut values = Vec::with_capacity(num_points);
 
             if metric_type == V3_METRIC_TYPE_SKETCH {
@@ -468,14 +493,8 @@ impl Metric {
                     let timestamp = u64::try_from(ts).map_err(|_| generic_error!("Invalid timestamp: {}", ts))?;
                     cursors.timestamp += 1;
 
-                    // Sketch count is always in valsSint64.
-                    let cnt = *data
-                        .valsSint64
-                        .get(cursors.sint64)
-                        .ok_or_else(|| generic_error!("Ran out of sint64 values for sketch count"))?;
-                    cursors.sint64 += 1;
-
-                    // Sum, min, max are stored as 3 consecutive values based on value_type.
+                    // The Agent writes sketch summaries as sum, min, max, then count. Count is always in valsSint64,
+                    // but integer summaries can share that column, so count must be read after the summary values.
                     let sum = read_value(
                         value_type,
                         &mut cursors,
@@ -497,6 +516,11 @@ impl Metric {
                         &data.valsFloat32,
                         &data.valsFloat64,
                     )?;
+                    let cnt = *data
+                        .valsSint64
+                        .get(cursors.sint64)
+                        .ok_or_else(|| generic_error!("Ran out of sint64 values for sketch count"))?;
+                    cursors.sint64 += 1;
                     let avg = if cnt != 0 { sum / cnt as f64 } else { 0.0 };
 
                     // Read bin data.
@@ -669,6 +693,52 @@ fn parse_tagsets(dict_tagsets: &[i64], tags_dict: &[String]) -> Result<Vec<Vec<S
     Ok(tagsets)
 }
 
+/// Parse resource sets from V3 resource dictionaries.
+///
+/// Each resource set is encoded as one length entry plus that many locally delta-encoded type/name dictionary indexes.
+fn parse_resources(
+    dict_resource_len: &[i64], dict_resource_type: &[i64], dict_resource_name: &[i64], resource_strings: &[String],
+) -> Result<Vec<Vec<(String, String)>>, GenericError> {
+    let mut resources = Vec::with_capacity(dict_resource_len.len());
+    let mut offset = 0;
+
+    for &count in dict_resource_len {
+        let count = usize::try_from(count).map_err(|_| generic_error!("Invalid negative resource count: {}", count))?;
+        if offset + count > dict_resource_type.len() || offset + count > dict_resource_name.len() {
+            return Err(generic_error!("Resource set extends past resource dictionary arrays"));
+        }
+
+        let mut type_indices = dict_resource_type[offset..offset + count].to_vec();
+        let mut name_indices = dict_resource_name[offset..offset + count].to_vec();
+        delta_decode(&mut type_indices);
+        delta_decode(&mut name_indices);
+
+        let mut resource_set = Vec::with_capacity(count);
+        for (&type_idx, &name_idx) in type_indices.iter().zip(name_indices.iter()) {
+            let resource_type = resource_strings
+                .get(resource_index(type_idx)?)
+                .ok_or_else(|| generic_error!("Invalid resource type index {}", type_idx))?
+                .clone();
+            let resource_name = resource_strings
+                .get(resource_index(name_idx)?)
+                .ok_or_else(|| generic_error!("Invalid resource name index {}", name_idx))?
+                .clone();
+            resource_set.push((resource_type, resource_name));
+        }
+
+        resources.push(resource_set);
+        offset += count;
+    }
+
+    Ok(resources)
+}
+
+fn resource_index(idx: i64) -> Result<usize, GenericError> {
+    let idx = usize::try_from(idx).map_err(|_| generic_error!("Invalid negative resource index: {}", idx))?;
+    idx.checked_sub(1)
+        .ok_or_else(|| generic_error!("Invalid zero resource index"))
+}
+
 /// Read the next f64 value from the appropriate value array based on `value_type`.
 fn read_value(
     value_type: u64, cursors: &mut V3ValueCursors, vals_sint64: &[i64], vals_float32: &[f32], vals_float64: &[f64],
@@ -834,4 +904,78 @@ mod tests {
         assert!(metrics[0].context.tags.contains(&"host:server-1".to_string()));
         assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
     }
+
+    #[test]
+    fn try_from_v3_folds_host_resource_into_tags() {
+        use datadog_protos::metrics::v3::{MetricData, Payload};
+
+        let mut data = MetricData::new();
+        data.dictNameStr = length_prefixed_strings(["my.metric"]);
+        data.dictTagStr = length_prefixed_strings(["env:prod"]);
+        data.dictTagsets = vec![1, 1];
+        data.dictResourceStr = length_prefixed_strings(["host", "server-1", "device", "eth0"]);
+        data.dictResourceLen = vec![2];
+        data.dictResourceType = vec![1, 2];
+        data.dictResourceName = vec![2, 2];
+        data.types = vec![V3_METRIC_TYPE_COUNT | V3_VALUE_TYPE_ZERO];
+        data.nameRefs = vec![1];
+        data.tagsetRefs = vec![1];
+        data.resourcesRefs = vec![1];
+        data.intervals = vec![0];
+        data.numPoints = vec![1];
+        data.timestamps = vec![1];
+
+        let mut payload = Payload::new();
+        payload.metricData = Some(data).into();
+
+        let metrics = Metric::try_from_v3(payload).expect("parse should succeed");
+        assert_eq!(metrics.len(), 1);
+        assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
+        assert!(metrics[0].context.tags.contains(&"host:server-1".to_string()));
+        assert!(!metrics[0].context.tags.iter().any(|tag| tag.starts_with("device:")));
+    }
+
+    #[test]
+    fn try_from_v3_decodes_integer_sketch_summary_order() {
+        use datadog_protos::metrics::v3::{MetricData, Payload};
+
+        let mut data = MetricData::new();
+        data.dictNameStr = length_prefixed_strings(["my.sketch"]);
+        data.types = vec![V3_METRIC_TYPE_SKETCH | V3_VALUE_TYPE_SINT64];
+        data.nameRefs = vec![1];
+        data.tagsetRefs = vec![0];
+        data.resourcesRefs = vec![0];
+        data.intervals = vec![0];
+        data.numPoints = vec![1];
+        data.timestamps = vec![123];
+        // Agent V3 sketch ordering is sum, min, max, count when integer summaries share valsSint64.
+        data.valsSint64 = vec![10, 1, 4, 4];
+        data.sketchNumBins = vec![1];
+        data.sketchBinKeys = vec![0];
+        data.sketchBinCnts = vec![4];
+
+        let mut payload = Payload::new();
+        payload.metricData = Some(data).into();
+
+        let metrics = Metric::try_from_v3(payload).expect("parse should succeed");
+        assert_eq!(metrics.len(), 1);
+
+        let MetricValue::Sketch { sketch } = &metrics[0].values[0].1 else {
+            panic!("expected sketch value");
+        };
+        assert_eq!(sketch.count(), 4);
+        assert_eq!(sketch.sum(), Some(10.0));
+        assert_eq!(sketch.min(), Some(1.0));
+        assert_eq!(sketch.max(), Some(4.0));
+        assert_eq!(sketch.avg(), Some(2.5));
+    }
+
+    fn length_prefixed_strings(strings: impl IntoIterator<Item = &'static str>) -> Vec<u8> {
+        let mut bytes = Vec::new();
+        for s in strings {
+            bytes.push(s.len() as u8);
+            bytes.extend_from_slice(s.as_bytes());
+        }
+        bytes
+    }
 }

test/correctness/cases/dsd-plain-v3-validation/config.yaml

@@ -0,0 +1,19 @@
+type: correctness
+runtime: docker
+analysis_mode: metrics
+baseline:
+  image: saluki-images/datadog-agent:testing-release
+  files:
+    - datadog.yaml:/etc/datadog-agent/datadog.yaml
+  additional_env_vars:
+    - DD_API_KEY=correctness-test
+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_STANDALONE_MODE=true
+    - DD_DATA_PLANE_DOGSTATSD_ENABLED=true
+    - DD_AGGREGATE_CONTEXT_LIMIT=500000

test/correctness/cases/dsd-plain-v3-validation/datadog.yaml

@@ -0,0 +1,36 @@
+# Using a fixed hostname is both required to avoid errors, and also will ensure consistent tags between DSD/ADP.
+hostname: "correctness-testing"
+
+# Dummy API key.
+api_key: dummy-api-key-correctness-testing
+
+# We have to specifically configure the health port to use.
+health_port: 5555
+
+# Point ourselves at the datadog-intake service.
+dd_url: "http://datadog-intake:2049"
+
+# Turn off UDP and listen on a UDS socket instead.
+dogstatsd_port: 0
+dogstatsd_socket: /airlock/metrics.sock
+
+# Ensure origin detection is disabled since we can't support it with ADP in standalone mode.
+dogstatsd_origin_detection: false
+
+# Gauges can be processed out-of-order when multiple workers are used, while ADP does not use multiple workers, so ADP
+# always ends up with the correct (last seen) value, while DSD might return the last seen value... or the value seen
+# four updates ago, etc etc.
+dogstatsd_workers_count: 1
+
+# Enable V3 metrics encoding in validation mode: both V2 and V3 payloads are sent simultaneously,
+# paired by X-Metrics-Request-ID. V3 payloads are counted in the metrics dump; V2 payloads are
+# used only for comparison against V3 to validate encoding correctness.
+serializer_experimental_use_v3_api:
+  series:
+    endpoints:
+      - "http://datadog-intake:2049"
+    validate: true
+  sketches:
+    endpoints:
+      - "http://datadog-intake:2049"
+    validate: true

test/correctness/cases/dsd-plain-v3-validation/millstone.yaml

@@ -0,0 +1,57 @@
+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: "unixgram:///$GROUP-airlock/metrics.sock"
+aggregation_bucket_width_secs: 10
+volume: 10000
+corpus:
+  # TODO: This is a little confusing, because we're specifying the number of metrics to generate (which we _will_
+  # honor faithfully) but since we're specifying the contexts count in the payload definition, we might not
+  # actually generate 10,000 unique contexts, but instead somewhere below 3,000, where each of them is repeated a
+  # few times to reach the total count.
+  #
+  # We need to figure that out, since the intent is that specifying a fixed count should lead to that many metrics
+  # (and no more) being generated, such that you could depend on that for testing purposes.
+  size: 10000
+  payload:
+    dogstatsd:
+      contexts:
+        constant: 3000
+      name_length:
+        inclusive:
+          min: 1
+          max: 32
+      tag_length:
+        inclusive:
+          min: 3
+          max: 16
+      tags_per_msg:
+        inclusive:
+          min: 2
+          max: 8
+      value:
+        float_probability: 0.5
+        range:
+          inclusive:
+            min: -9999999
+            max: 9999999
+      multivalue_count:
+        inclusive:
+          min: 2
+          max: 32
+      multivalue_pack_probability: 0.08
+      kind_weights:
+        metric: 100
+        event: 0
+        service_check: 0
+      # Weights based on analyzing internal Datadog usage data of metric type for metrics sent to the Agent over DogStatsD.
+      metric_weights:
+        count: 208
+        gauge: 66
+        timer: 0
+        distribution: 72
+        # We specifically _don't_ want to generate sets, because we can't assert their correctness once they've been
+        # aggregated: a gauge is generated for each aggregator flush that represents the unique number of values in a
+        # given set, but in general, gauges are meant to be last-write-wins, so unless the metric names/tags can
+        # indicate that they're for a set, we can't know that it's safe for us to _aggregate_ the gauge values, and with
+        # our default behavior of taking the latest gauge value... we end up with non-deterministic results.
+        set: 0
+        histogram: 1

test/correctness/cases/dsd-plain-v3/config.yaml

@@ -0,0 +1,19 @@
+type: correctness
+runtime: docker
+analysis_mode: metrics
+baseline:
+  image: saluki-images/datadog-agent:testing-release
+  files:
+    - datadog.yaml:/etc/datadog-agent/datadog.yaml
+  additional_env_vars:
+    - DD_API_KEY=correctness-test
+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_STANDALONE_MODE=true
+    - DD_DATA_PLANE_DOGSTATSD_ENABLED=true
+    - DD_AGGREGATE_CONTEXT_LIMIT=500000

test/correctness/cases/dsd-plain-v3/datadog.yaml

@@ -0,0 +1,34 @@
+# Using a fixed hostname is both required to avoid errors, and also will ensure consistent tags between DSD/ADP.
+hostname: "correctness-testing"
+
+# Dummy API key.
+api_key: dummy-api-key-correctness-testing
+
+# We have to specifically configure the health port to use.
+health_port: 5555
+
+# Point ourselves at the datadog-intake service.
+dd_url: "http://datadog-intake:2049"
+
+# Turn off UDP and listen on a UDS socket instead.
+dogstatsd_port: 0
+dogstatsd_socket: /airlock/metrics.sock
+
+# Ensure origin detection is disabled since we can't support it with ADP in standalone mode.
+dogstatsd_origin_detection: false
+
+# Gauges can be processed out-of-order when multiple workers are used, while ADP does not use multiple workers, so ADP
+# always ends up with the correct (last seen) value, while DSD might return the last seen value... or the value seen
+# four updates ago, etc etc.
+dogstatsd_workers_count: 1
+
+# Enable V3 metrics encoding for all endpoints.
+#
+# Validation mode is off here so this case compares Agent V3 output against ADP V3 output directly.
+serializer_experimental_use_v3_api:
+  series:
+    endpoints:
+      - "http://datadog-intake:2049"
+  sketches:
+    endpoints:
+      - "http://datadog-intake:2049"