1use std::{collections::BTreeMap, fmt};
2
3use datadog_protos::metrics::{v3::Payload as V3Payload, Dogsketch, MetricPayload, MetricType, SketchPayload};
4use ddsketch::DDSketch;
5use float_cmp::ApproxEqRatio as _;
6use saluki_error::{generic_error, GenericError};
7use serde::{Deserialize, Serialize};
8
9#[derive(Deserialize)]
11struct V1SeriesEnvelope {
12 series: Vec<V1Serie>,
13}
14
15#[derive(Deserialize)]
21struct V1Serie {
22 metric: String,
23 points: Vec<(i64, f64)>,
24 #[serde(default)]
25 tags: Vec<String>,
26 #[serde(default)]
27 host: String,
28 #[serde(default)]
29 device: String,
30 #[serde(rename = "type", default)]
31 mtype: String,
32 #[serde(default)]
33 interval: i64,
34}
35
36#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
42pub struct MetricContext {
43 name: String,
44 tags: Vec<String>,
45}
46
47impl MetricContext {
48 pub fn name(&self) -> &str {
50 &self.name
51 }
52
53 pub fn tags(&self) -> &[String] {
57 &self.tags
58 }
59
60 pub fn into_parts(self) -> (String, Vec<String>) {
62 (self.name, self.tags)
63 }
64}
65
66impl fmt::Display for MetricContext {
67 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
68 write!(f, "{}", self.name)?;
69
70 if !self.tags.is_empty() {
71 write!(f, " {{{}}}", self.tags.join(", "))?;
72 }
73
74 Ok(())
75 }
76}
77
78#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
80pub struct Metric {
81 context: MetricContext,
82 values: Vec<(u64, MetricValue)>,
83}
84
85impl Metric {
86 pub fn context(&self) -> &MetricContext {
88 &self.context
89 }
90
91 pub fn values(&self) -> &[(u64, MetricValue)] {
93 &self.values
94 }
95}
96
97#[derive(Clone, Debug, Deserialize, Serialize)]
120#[serde(tag = "mtype")]
121pub enum MetricValue {
122 Count {
124 value: f64,
126 },
127
128 Rate {
133 interval: u64,
135
136 value: f64,
138 },
139
140 Gauge {
142 value: f64,
144 },
145
146 Sketch {
148 sketch: DDSketch,
150 },
151}
152
153impl PartialEq for MetricValue {
154 fn eq(&self, other: &Self) -> bool {
155 const RATIO_ERROR: f64 = 0.00000001;
157
158 match (self, other) {
159 (MetricValue::Count { value: value_a }, MetricValue::Count { value: value_b }) => {
160 value_a.approx_eq_ratio(value_b, RATIO_ERROR)
161 }
162 (
163 MetricValue::Rate {
164 interval: interval_a,
165 value: value_a,
166 },
167 MetricValue::Rate {
168 interval: interval_b,
169 value: value_b,
170 },
171 ) => interval_a == interval_b && value_a.approx_eq_ratio(value_b, RATIO_ERROR),
172 (MetricValue::Gauge { value: value_a }, MetricValue::Gauge { value: value_b }) => {
173 value_a.approx_eq_ratio(value_b, RATIO_ERROR)
174 }
175 (MetricValue::Sketch { sketch: sketch_a }, MetricValue::Sketch { sketch: sketch_b }) => {
176 approx_eq_ratio_optional(sketch_a.min(), sketch_b.min(), RATIO_ERROR)
177 && approx_eq_ratio_optional(sketch_a.max(), sketch_b.max(), RATIO_ERROR)
178 && approx_eq_ratio_optional(sketch_a.avg(), sketch_b.avg(), RATIO_ERROR)
179 && approx_eq_ratio_optional(sketch_a.sum(), sketch_b.sum(), RATIO_ERROR)
180 && sketch_a.count() == sketch_b.count()
181 && sketch_bins_equal(sketch_a, sketch_b)
182 }
183 _ => false,
184 }
185 }
186}
187
188impl Eq for MetricValue {}
189
190impl Metric {
191 pub fn try_from_series_v1(payload: &[u8]) -> Result<Vec<Self>, GenericError> {
202 let envelope: V1SeriesEnvelope = serde_json::from_slice(payload)
203 .map_err(|e| generic_error!("Failed to parse V1 series JSON payload: {}", e))?;
204
205 let mut metrics = Vec::with_capacity(envelope.series.len());
206
207 for serie in envelope.series {
208 let mut tags = serie.tags;
209 if !serie.host.is_empty() {
210 tags.push(format!("host:{}", serie.host));
211 }
212 if !serie.device.is_empty() {
213 tags.push(format!("device:{}", serie.device));
214 }
215
216 let mut values = Vec::with_capacity(serie.points.len());
217 for (ts, value) in serie.points {
218 let timestamp =
219 u64::try_from(ts).map_err(|_| generic_error!("Invalid timestamp in V1 series payload: {}", ts))?;
220
221 let metric_value = match serie.mtype.as_str() {
222 "count" => MetricValue::Count { value },
223 "rate" => MetricValue::Rate {
224 interval: serie.interval as u64,
225 value,
226 },
227 "gauge" => MetricValue::Gauge { value },
228 other => {
229 return Err(generic_error!(
230 "Unknown metric type '{}' in V1 series payload (metric '{}')",
231 other,
232 serie.metric
233 ));
234 }
235 };
236 values.push((timestamp, metric_value));
237 }
238
239 metrics.push(Metric {
240 context: MetricContext {
241 name: serie.metric,
242 tags,
243 },
244 values,
245 });
246 }
247
248 Ok(metrics)
249 }
250
251 pub fn try_from_series_v2(payload: MetricPayload) -> Result<Vec<Self>, GenericError> {
257 let mut metrics = Vec::new();
258
259 for series in payload.series {
260 let name = series.metric().to_string();
261 let mut tags: Vec<String> = series.tags().iter().map(|tag| tag.to_string()).collect();
262 if let Some(host) = series.resources.iter().find(|r| r.type_() == "host") {
267 let host_name = host.name();
268 if !host_name.is_empty() {
269 tags.push(format!("host:{}", host_name));
270 }
271 }
272 let mut values = Vec::new();
273
274 match series.type_() {
275 MetricType::UNSPECIFIED => {
276 return Err(generic_error!("Received metric series with UNSPECIFIED type."));
277 }
278 MetricType::COUNT => {
279 for point in series.points {
280 let timestamp = u64::try_from(point.timestamp)
281 .map_err(|_| generic_error!("Invalid timestamp for point: {}", point.timestamp))?;
282 values.push((timestamp, MetricValue::Count { value: point.value }));
283 }
284 }
285 MetricType::RATE => {
286 for point in series.points {
287 let timestamp = u64::try_from(point.timestamp)
288 .map_err(|_| generic_error!("Invalid timestamp for point: {}", point.timestamp))?;
289 values.push((
290 timestamp,
291 MetricValue::Rate {
292 interval: series.interval as u64,
293 value: point.value,
294 },
295 ));
296 }
297 }
298 MetricType::GAUGE => {
299 for point in series.points {
300 let timestamp = u64::try_from(point.timestamp)
301 .map_err(|_| generic_error!("Invalid timestamp for point: {}", point.timestamp))?;
302 values.push((timestamp, MetricValue::Gauge { value: point.value }));
303 }
304 }
305 }
306
307 metrics.push(Metric {
308 context: MetricContext { name, tags },
309 values,
310 })
311 }
312
313 Ok(metrics)
314 }
315
316 pub fn try_from_sketch(payload: SketchPayload) -> Result<Vec<Self>, GenericError> {
322 let mut metrics = Vec::new();
323
324 for sketch in payload.sketches {
325 let name = sketch.metric().to_string();
326 let mut tags: Vec<String> = sketch.tags().iter().map(|tag| tag.to_string()).collect();
327 let host = sketch.host();
330 if !host.is_empty() {
331 tags.push(format!("host:{}", host));
332 }
333 let mut values = Vec::new();
334
335 for dogsketch in sketch.dogsketches {
336 let timestamp = u64::try_from(dogsketch.ts)
337 .map_err(|_| generic_error!("Invalid timestamp for sketch: {}", dogsketch.ts))?;
338 let sketch = DDSketch::try_from(dogsketch)
339 .map_err(|e| generic_error!("Failed to convert DogSketch to DDSketch: {}", e))?;
340 values.push((timestamp, MetricValue::Sketch { sketch }));
341 }
342
343 metrics.push(Metric {
344 context: MetricContext { name, tags },
345 values,
346 })
347 }
348
349 Ok(metrics)
350 }
351}
352
353const V3_METRIC_TYPE_COUNT: u64 = 1;
355const V3_METRIC_TYPE_RATE: u64 = 2;
356const V3_METRIC_TYPE_GAUGE: u64 = 3;
357const V3_METRIC_TYPE_SKETCH: u64 = 4;
358
359const V3_VALUE_TYPE_ZERO: u64 = 0x00;
361const V3_VALUE_TYPE_SINT64: u64 = 0x10;
362const V3_VALUE_TYPE_FLOAT32: u64 = 0x20;
363const V3_VALUE_TYPE_FLOAT64: u64 = 0x30;
364
365struct V3ValueCursors {
367 timestamp: usize,
368 sint64: usize,
369 float32: usize,
370 float64: usize,
371 sketch_point: usize,
372 sketch_bin_key: usize,
373 sketch_bin_cnt: usize,
374}
375
376impl V3ValueCursors {
377 fn new() -> Self {
378 Self {
379 timestamp: 0,
380 sint64: 0,
381 float32: 0,
382 float64: 0,
383 sketch_point: 0,
384 sketch_bin_key: 0,
385 sketch_bin_cnt: 0,
386 }
387 }
388}
389
390impl Metric {
391 pub fn try_from_v3(mut payload: V3Payload) -> Result<Vec<Self>, GenericError> {
399 let data = payload
400 .metricData
401 .take()
402 .ok_or_else(|| generic_error!("V3 payload missing metricData"))?;
403
404 let num_metrics = data.types.len();
405 if num_metrics == 0 {
406 return Ok(Vec::new());
407 }
408
409 let names_dict = parse_dict_strings(&data.dictNameStr)?;
411 let tags_dict = parse_dict_strings(&data.dictTagStr)?;
412 let tagsets_dict = parse_tagsets(&data.dictTagsets, &tags_dict)?;
413 let resources_dict = parse_resources(
414 &data.dictResourceLen,
415 &data.dictResourceType,
416 &data.dictResourceName,
417 &parse_dict_strings(&data.dictResourceStr)?,
418 )?;
419
420 let mut name_refs = data.nameRefs;
422 let mut tagset_refs = data.tagsetRefs;
423 let mut resources_refs = data.resourcesRefs;
424 let mut timestamps = data.timestamps;
425 delta_decode(&mut name_refs);
426 delta_decode(&mut tagset_refs);
427 delta_decode(&mut resources_refs);
428 delta_decode(&mut timestamps);
429
430 let mut sketch_bin_keys = data.sketchBinKeys;
433
434 let mut cursors = V3ValueCursors::new();
435 let mut metrics = Vec::with_capacity(num_metrics);
436
437 for i in 0..num_metrics {
438 let type_field = data
439 .types
440 .get(i)
441 .copied()
442 .ok_or_else(|| generic_error!("Ran out of metric types"))?;
443 let metric_type = type_field & 0x0F;
444 let value_type = type_field & 0xF0;
445 let num_points = data
446 .numPoints
447 .get(i)
448 .copied()
449 .ok_or_else(|| generic_error!("Ran out of numPoints"))
450 .and_then(|num_points| u64_to_usize(num_points, "numPoints"))?;
451
452 let name_ref = name_refs
454 .get(i)
455 .copied()
456 .ok_or_else(|| generic_error!("Ran out of nameRefs"))
457 .and_then(|name_ref| i64_to_usize(name_ref, "name ref"))?;
458 let name = if name_ref == 0 {
459 String::new()
460 } else {
461 names_dict
462 .get(name_ref - 1)
463 .ok_or_else(|| generic_error!("Invalid name ref {} (dict size {})", name_ref, names_dict.len()))?
464 .clone()
465 };
466
467 let tagset_ref = tagset_refs
469 .get(i)
470 .copied()
471 .ok_or_else(|| generic_error!("Ran out of tagsetRefs"))
472 .and_then(|tagset_ref| i64_to_usize(tagset_ref, "tagset ref"))?;
473 let mut tags = if tagset_ref == 0 {
474 Vec::new()
475 } else {
476 tagsets_dict
477 .get(tagset_ref - 1)
478 .ok_or_else(|| {
479 generic_error!("Invalid tagset ref {} (dict size {})", tagset_ref, tagsets_dict.len())
480 })?
481 .clone()
482 };
483
484 let resource_ref = resources_refs
485 .get(i)
486 .copied()
487 .map(|resource_ref| i64_to_usize(resource_ref, "resource ref"))
488 .transpose()?
489 .unwrap_or(0);
490 if resource_ref != 0 {
491 let resources = resources_dict.get(resource_ref - 1).ok_or_else(|| {
492 generic_error!(
493 "Invalid resource ref {} (dict size {})",
494 resource_ref,
495 resources_dict.len()
496 )
497 })?;
498 if let Some((_, host_name)) = resources
499 .iter()
500 .find(|(resource_type, resource_name)| resource_type == "host" && !resource_name.is_empty())
501 {
502 tags.push(format!("host:{}", host_name));
503 }
504 }
505
506 let mut values = Vec::with_capacity(num_points);
507
508 if metric_type == V3_METRIC_TYPE_SKETCH {
509 for _ in 0..num_points {
510 let ts = *timestamps
512 .get(cursors.timestamp)
513 .ok_or_else(|| generic_error!("Ran out of timestamps"))?;
514 let timestamp = u64::try_from(ts).map_err(|_| generic_error!("Invalid timestamp: {}", ts))?;
515 cursors.timestamp += 1;
516
517 let sum = read_value(
520 value_type,
521 &mut cursors,
522 &data.valsSint64,
523 &data.valsFloat32,
524 &data.valsFloat64,
525 )?;
526 let min = read_value(
527 value_type,
528 &mut cursors,
529 &data.valsSint64,
530 &data.valsFloat32,
531 &data.valsFloat64,
532 )?;
533 let max = read_value(
534 value_type,
535 &mut cursors,
536 &data.valsSint64,
537 &data.valsFloat32,
538 &data.valsFloat64,
539 )?;
540 let cnt = *data
541 .valsSint64
542 .get(cursors.sint64)
543 .ok_or_else(|| generic_error!("Ran out of sint64 values for sketch count"))?;
544 cursors.sint64 += 1;
545 let avg = if cnt != 0 { sum / cnt as f64 } else { 0.0 };
546
547 let num_bins = *data
549 .sketchNumBins
550 .get(cursors.sketch_point)
551 .ok_or_else(|| generic_error!("Ran out of sketchNumBins"))?
552 as usize;
553 cursors.sketch_point += 1;
554
555 let bin_key_start = cursors.sketch_bin_key;
556 let bin_key_end = bin_key_start + num_bins;
557 if bin_key_end > sketch_bin_keys.len() {
558 return Err(generic_error!("Ran out of sketch bin keys"));
559 }
560
561 delta_decode_i32(&mut sketch_bin_keys[bin_key_start..bin_key_end]);
563
564 let k: Vec<i32> = sketch_bin_keys[bin_key_start..bin_key_end].to_vec();
565 cursors.sketch_bin_key = bin_key_end;
566
567 let bin_cnt_start = cursors.sketch_bin_cnt;
568 let bin_cnt_end = bin_cnt_start + num_bins;
569 if bin_cnt_end > data.sketchBinCnts.len() {
570 return Err(generic_error!("Ran out of sketch bin counts"));
571 }
572 let n: Vec<u32> = data.sketchBinCnts[bin_cnt_start..bin_cnt_end].to_vec();
573 cursors.sketch_bin_cnt = bin_cnt_end;
574
575 let mut dogsketch = Dogsketch::new();
577 dogsketch.ts = ts;
578 dogsketch.cnt = cnt;
579 dogsketch.min = min;
580 dogsketch.max = max;
581 dogsketch.avg = avg;
582 dogsketch.sum = sum;
583 dogsketch.set_k(k);
584 dogsketch.set_n(n);
585
586 let sketch = DDSketch::try_from(dogsketch)
587 .map_err(|e| generic_error!("Failed to convert v3 sketch to DDSketch: {}", e))?;
588 values.push((timestamp, MetricValue::Sketch { sketch }));
589 }
590 } else {
591 for _ in 0..num_points {
592 let ts = *timestamps
594 .get(cursors.timestamp)
595 .ok_or_else(|| generic_error!("Ran out of timestamps"))?;
596 let timestamp = u64::try_from(ts).map_err(|_| generic_error!("Invalid timestamp: {}", ts))?;
597 cursors.timestamp += 1;
598
599 let value = read_value(
601 value_type,
602 &mut cursors,
603 &data.valsSint64,
604 &data.valsFloat32,
605 &data.valsFloat64,
606 )?;
607
608 let metric_value = match metric_type {
609 V3_METRIC_TYPE_COUNT => MetricValue::Count { value },
610 V3_METRIC_TYPE_RATE => MetricValue::Rate {
611 interval: data
612 .intervals
613 .get(i)
614 .copied()
615 .ok_or_else(|| generic_error!("Ran out of intervals"))?,
616 value,
617 },
618 V3_METRIC_TYPE_GAUGE => MetricValue::Gauge { value },
619 other => return Err(generic_error!("Unknown v3 metric type: {}", other)),
620 };
621
622 values.push((timestamp, metric_value));
623 }
624 }
625
626 metrics.push(Metric {
627 context: MetricContext { name, tags },
628 values,
629 });
630 }
631
632 Ok(metrics)
633 }
634}
635
636fn delta_decode(s: &mut [i64]) {
638 for i in 1..s.len() {
639 s[i] += s[i - 1];
640 }
641}
642
643fn delta_decode_i32(s: &mut [i32]) {
645 for i in 1..s.len() {
646 s[i] += s[i - 1];
647 }
648}
649
650fn read_varint(data: &[u8]) -> Result<(u64, usize), GenericError> {
652 let mut value: u64 = 0;
653 let mut shift = 0;
654 for (i, &byte) in data.iter().enumerate() {
655 value |= ((byte & 0x7F) as u64) << shift;
656 if byte & 0x80 == 0 {
657 return Ok((value, i + 1));
658 }
659 shift += 7;
660 if shift >= 64 {
661 return Err(generic_error!("Varint too large"));
662 }
663 }
664 Err(generic_error!("Unexpected end of data reading varint"))
665}
666
667fn parse_dict_strings(data: &[u8]) -> Result<Vec<String>, GenericError> {
669 let mut strings = Vec::new();
670 let mut offset = 0;
671 while offset < data.len() {
672 let (len, varint_size) = read_varint(&data[offset..])?;
673 offset += varint_size;
674 let len = len as usize;
675 if offset + len > data.len() {
676 return Err(generic_error!("Dictionary string extends past end of buffer"));
677 }
678 let s = simdutf8::basic::from_utf8(&data[offset..offset + len])
679 .map_err(|e| generic_error!("Invalid UTF-8 in dictionary string: {}", e))?;
680 strings.push(s.to_string());
681 offset += len;
682 }
683 Ok(strings)
684}
685
686fn parse_tagsets(dict_tagsets: &[i64], tags_dict: &[String]) -> Result<Vec<Vec<String>>, GenericError> {
699 let mut tagsets: Vec<Vec<String>> = Vec::new();
700 let mut offset = 0;
701 while offset < dict_tagsets.len() {
702 let count = i64_to_usize(dict_tagsets[offset], "tagset length")?;
703 offset += 1;
704 if offset + count > dict_tagsets.len() {
705 return Err(generic_error!("Tagset extends past end of dictTagsets array"));
706 }
707
708 let mut entries: Vec<i64> = dict_tagsets[offset..offset + count].to_vec();
710 delta_decode(&mut entries);
711
712 let mut tags = Vec::with_capacity(count);
713 for &entry in &entries {
714 match entry.cmp(&0) {
715 std::cmp::Ordering::Less => {
716 let prefix_id = i64_to_usize(-entry, "tagset prefix reference")?;
718 let prefix_idx = prefix_id
719 .checked_sub(1)
720 .ok_or_else(|| generic_error!("Invalid zero tagset prefix reference"))?;
721 let prefix_tags = tagsets.get(prefix_idx).ok_or_else(|| {
722 generic_error!(
723 "Invalid tagset prefix reference {} (only {} tagsets parsed so far)",
724 prefix_id,
725 tagsets.len()
726 )
727 })?;
728 tags.extend(prefix_tags.iter().cloned());
729 }
730 std::cmp::Ordering::Equal => continue,
731 std::cmp::Ordering::Greater => {
732 let idx = i64_to_usize(entry, "tag index")?;
733 let tag = tags_dict
734 .get(idx - 1)
735 .ok_or_else(|| generic_error!("Invalid tag index {} (dict size {})", idx, tags_dict.len()))?;
736 tags.push(tag.clone());
737 }
738 }
739 }
740 tagsets.push(tags);
741 offset += count;
742 }
743 Ok(tagsets)
744}
745
746fn u64_to_usize(value: u64, field: &str) -> Result<usize, GenericError> {
747 usize::try_from(value).map_err(|_| generic_error!("Invalid {}: {}", field, value))
748}
749
750fn i64_to_usize(value: i64, field: &str) -> Result<usize, GenericError> {
751 usize::try_from(value).map_err(|_| generic_error!("Invalid negative {}: {}", field, value))
752}
753
754fn parse_resources(
758 dict_resource_len: &[i64], dict_resource_type: &[i64], dict_resource_name: &[i64], resource_strings: &[String],
759) -> Result<Vec<Vec<(String, String)>>, GenericError> {
760 let mut resources = Vec::with_capacity(dict_resource_len.len());
761 let mut offset = 0;
762
763 for &count in dict_resource_len {
764 let count = usize::try_from(count).map_err(|_| generic_error!("Invalid negative resource count: {}", count))?;
765 if offset + count > dict_resource_type.len() || offset + count > dict_resource_name.len() {
766 return Err(generic_error!("Resource set extends past resource dictionary arrays"));
767 }
768
769 let mut type_indices = dict_resource_type[offset..offset + count].to_vec();
770 let mut name_indices = dict_resource_name[offset..offset + count].to_vec();
771 delta_decode(&mut type_indices);
772 delta_decode(&mut name_indices);
773
774 let mut resource_set = Vec::with_capacity(count);
775 for (&type_idx, &name_idx) in type_indices.iter().zip(name_indices.iter()) {
776 let resource_type = resource_strings
777 .get(resource_index(type_idx)?)
778 .ok_or_else(|| generic_error!("Invalid resource type index {}", type_idx))?
779 .clone();
780 let resource_name = resource_strings
781 .get(resource_index(name_idx)?)
782 .ok_or_else(|| generic_error!("Invalid resource name index {}", name_idx))?
783 .clone();
784 resource_set.push((resource_type, resource_name));
785 }
786
787 resources.push(resource_set);
788 offset += count;
789 }
790
791 Ok(resources)
792}
793
794fn resource_index(idx: i64) -> Result<usize, GenericError> {
795 let idx = usize::try_from(idx).map_err(|_| generic_error!("Invalid negative resource index: {}", idx))?;
796 idx.checked_sub(1)
797 .ok_or_else(|| generic_error!("Invalid zero resource index"))
798}
799
800fn read_value(
802 value_type: u64, cursors: &mut V3ValueCursors, vals_sint64: &[i64], vals_float32: &[f32], vals_float64: &[f64],
803) -> Result<f64, GenericError> {
804 match value_type {
805 V3_VALUE_TYPE_ZERO => Ok(0.0),
806 V3_VALUE_TYPE_SINT64 => {
807 let v = *vals_sint64
808 .get(cursors.sint64)
809 .ok_or_else(|| generic_error!("Ran out of sint64 values"))?;
810 cursors.sint64 += 1;
811 Ok(v as f64)
812 }
813 V3_VALUE_TYPE_FLOAT32 => {
814 let v = *vals_float32
815 .get(cursors.float32)
816 .ok_or_else(|| generic_error!("Ran out of float32 values"))?;
817 cursors.float32 += 1;
818 Ok(v as f64)
819 }
820 V3_VALUE_TYPE_FLOAT64 => {
821 let v = *vals_float64
822 .get(cursors.float64)
823 .ok_or_else(|| generic_error!("Ran out of float64 values"))?;
824 cursors.float64 += 1;
825 Ok(v)
826 }
827 _ => Err(generic_error!("Unknown v3 value type: {:#x}", value_type)),
828 }
829}
830
831fn approx_eq_ratio_optional(a: Option<f64>, b: Option<f64>, ratio: f64) -> bool {
832 match (a, b) {
833 (Some(a), Some(b)) => a.approx_eq_ratio(&b, ratio),
834 (None, None) => true,
835 _ => false,
836 }
837}
838
839fn sketch_bins_equal(a: &DDSketch, b: &DDSketch) -> bool {
840 let Some(a) = canonical_bin_counts(a) else {
841 return false;
842 };
843 let Some(b) = canonical_bin_counts(b) else {
844 return false;
845 };
846
847 a == b
848}
849
850fn canonical_bin_counts(sketch: &DDSketch) -> Option<BTreeMap<i32, u64>> {
851 let mut counts = BTreeMap::new();
852
853 for bin in sketch.bins() {
854 if bin.count() == 0 {
855 continue;
856 }
857
858 let count = counts.entry(bin.key()).or_insert(0_u64);
859 *count = count.checked_add(u64::from(bin.count()))?;
860 }
861
862 Some(counts)
863}
864
865#[cfg(test)]
866mod tests {
867 use super::*;
868
869 fn sketch_value(keys: Vec<i32>, counts: Vec<u32>) -> MetricValue {
870 let mut dogsketch = Dogsketch::new();
871 dogsketch.cnt = 70_000;
872 dogsketch.min = 1.0;
873 dogsketch.max = 1.0;
874 dogsketch.avg = 1.0;
875 dogsketch.sum = 70_000.0;
876 dogsketch.set_k(keys);
877 dogsketch.set_n(counts);
878
879 MetricValue::Sketch {
880 sketch: DDSketch::try_from(dogsketch).expect("test sketch should be valid"),
881 }
882 }
883
884 #[test]
885 fn sketch_equality_coalesces_duplicate_bin_keys() {
886 let split = sketch_value(vec![42, 42], vec![u16::MAX.into(), 4_465]);
887 let coalesced = sketch_value(vec![42], vec![70_000]);
888
889 assert_eq!(split, coalesced);
890 }
891
892 #[test]
893 fn sketch_equality_compares_bin_keys_and_counts() {
894 let original = sketch_value(vec![42], vec![70_000]);
895 let different_key = sketch_value(vec![43], vec![70_000]);
896 let different_count = sketch_value(vec![42], vec![69_999]);
897
898 assert_ne!(original, different_key);
899 assert_ne!(original, different_count);
900 }
901
902 #[test]
903 fn try_from_series_v1_parses_count_gauge_rate() {
904 let body = br#"{"series":[
905 {"metric":"a.count","points":[[100,5.0]],"tags":["env:prod"],"host":"h","type":"count","interval":0},
906 {"metric":"a.gauge","points":[[101,12.0]],"tags":[],"host":"h","type":"gauge","interval":0},
907 {"metric":"a.rate","points":[[102,3.0]],"tags":[],"host":"h","type":"rate","interval":10}
908 ]}"#;
909
910 let metrics = Metric::try_from_series_v1(body).expect("parse should succeed");
911 assert_eq!(metrics.len(), 3);
912
913 assert_eq!(metrics[0].context.name, "a.count");
914 assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
915 assert!(metrics[0].context.tags.contains(&"host:h".to_string()));
916 assert_eq!(metrics[0].values, vec![(100, MetricValue::Count { value: 5.0 })]);
917
918 assert_eq!(metrics[1].context.name, "a.gauge");
919 assert_eq!(metrics[1].context.tags, vec!["host:h".to_string()]);
920 assert_eq!(metrics[1].values, vec![(101, MetricValue::Gauge { value: 12.0 })]);
921
922 assert_eq!(metrics[2].context.name, "a.rate");
923 assert_eq!(metrics[2].context.tags, vec!["host:h".to_string()]);
924 assert_eq!(
925 metrics[2].values,
926 vec![(
927 102,
928 MetricValue::Rate {
929 interval: 10,
930 value: 3.0
931 }
932 )]
933 );
934 }
935
936 #[test]
937 fn try_from_series_v1_omits_host_tag_when_empty() {
938 let body = br#"{"series":[
941 {"metric":"m","points":[[1,1.0]],"tags":[],"type":"count","interval":0}
942 ]}"#;
943
944 let metrics = Metric::try_from_series_v1(body).expect("parse should succeed");
945 assert!(!metrics[0].context.tags.iter().any(|t| t.starts_with("host:")));
946 }
947
948 #[test]
949 fn try_from_series_v1_reinjects_device_tag() {
950 let body = br#"{"series":[
951 {"metric":"m","points":[[1,1.0]],"tags":["env:prod"],"host":"h","device":"eth0","type":"count","interval":0}
952 ]}"#;
953
954 let metrics = Metric::try_from_series_v1(body).expect("parse should succeed");
955 assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
956 assert!(metrics[0].context.tags.contains(&"device:eth0".to_string()));
957 }
958
959 #[test]
960 fn try_from_series_v1_rejects_unknown_type() {
961 let body = br#"{"series":[
962 {"metric":"m","points":[[1,1.0]],"tags":[],"host":"h","type":"weird","interval":0}
963 ]}"#;
964
965 assert!(Metric::try_from_series_v1(body).is_err());
966 }
967
968 #[test]
969 fn try_from_series_v2_folds_host_into_tags() {
970 use datadog_protos::metrics::metric_payload::{
971 MetricPoint, MetricSeries, MetricType as ProtoMetricType, Resource,
972 };
973 use datadog_protos::metrics::MetricPayload;
974
975 let mut payload = MetricPayload::new();
976
977 let mut series = MetricSeries::new();
978 series.set_metric("my.metric".into());
979 series.set_type(ProtoMetricType::COUNT);
980 series.tags.push("env:prod".into());
981
982 let mut host_res = Resource::new();
983 host_res.set_type("host".into());
984 host_res.set_name("server-1".into());
985 series.resources.push(host_res);
986
987 let mut device_res = Resource::new();
989 device_res.set_type("device".into());
990 device_res.set_name("eth0".into());
991 series.resources.push(device_res);
992
993 let mut point = MetricPoint::new();
994 point.value = 1.0;
995 point.timestamp = 1;
996 series.points.push(point);
997
998 payload.series.push(series);
999
1000 let metrics = Metric::try_from_series_v2(payload).expect("parse should succeed");
1001 assert_eq!(metrics.len(), 1);
1002 assert!(metrics[0].context.tags.contains(&"host:server-1".to_string()));
1003 assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
1004 assert!(!metrics[0].context.tags.iter().any(|t| t.starts_with("device:")));
1005 }
1006
1007 #[test]
1008 fn try_from_sketch_folds_host_into_tags() {
1009 use datadog_protos::metrics::sketch_payload::Sketch;
1010 use datadog_protos::metrics::SketchPayload;
1011
1012 let mut payload = SketchPayload::new();
1013 let mut sketch = Sketch::new();
1014 sketch.set_metric("my.metric".into());
1015 sketch.set_host("server-1".into());
1016 sketch.tags.push("env:prod".into());
1017 payload.sketches.push(sketch);
1018
1019 let metrics = Metric::try_from_sketch(payload).expect("parse should succeed");
1020 assert_eq!(metrics.len(), 1);
1021 assert!(metrics[0].context.tags.contains(&"host:server-1".to_string()));
1022 assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
1023 }
1024
1025 #[test]
1026 fn try_from_v3_folds_host_resource_into_tags() {
1027 use datadog_protos::metrics::v3::{MetricData, Payload};
1028
1029 let mut data = MetricData::new();
1030 data.dictNameStr = length_prefixed_strings(["my.metric"]);
1031 data.dictTagStr = length_prefixed_strings(["env:prod"]);
1032 data.dictTagsets = vec![1, 1];
1033 data.dictResourceStr = length_prefixed_strings(["host", "server-1", "device", "eth0"]);
1034 data.dictResourceLen = vec![2];
1035 data.dictResourceType = vec![1, 2];
1036 data.dictResourceName = vec![2, 2];
1037 data.types = vec![V3_METRIC_TYPE_COUNT | V3_VALUE_TYPE_ZERO];
1038 data.nameRefs = vec![1];
1039 data.tagsetRefs = vec![1];
1040 data.resourcesRefs = vec![1];
1041 data.intervals = vec![0];
1042 data.numPoints = vec![1];
1043 data.timestamps = vec![1];
1044
1045 let mut payload = Payload::new();
1046 payload.metricData = Some(data).into();
1047
1048 let metrics = Metric::try_from_v3(payload).expect("parse should succeed");
1049 assert_eq!(metrics.len(), 1);
1050 assert!(metrics[0].context.tags.contains(&"env:prod".to_string()));
1051 assert!(metrics[0].context.tags.contains(&"host:server-1".to_string()));
1052 assert!(!metrics[0].context.tags.iter().any(|tag| tag.starts_with("device:")));
1053 }
1054
1055 #[test]
1056 fn try_from_v3_decodes_integer_sketch_summary_order() {
1057 use datadog_protos::metrics::v3::{MetricData, Payload};
1058
1059 let mut data = MetricData::new();
1060 data.dictNameStr = length_prefixed_strings(["my.sketch"]);
1061 data.types = vec![V3_METRIC_TYPE_SKETCH | V3_VALUE_TYPE_SINT64];
1062 data.nameRefs = vec![1];
1063 data.tagsetRefs = vec![0];
1064 data.resourcesRefs = vec![0];
1065 data.intervals = vec![0];
1066 data.numPoints = vec![1];
1067 data.timestamps = vec![123];
1068 data.valsSint64 = vec![10, 1, 4, 4];
1070 data.sketchNumBins = vec![1];
1071 data.sketchBinKeys = vec![0];
1072 data.sketchBinCnts = vec![4];
1073
1074 let mut payload = Payload::new();
1075 payload.metricData = Some(data).into();
1076
1077 let metrics = Metric::try_from_v3(payload).expect("parse should succeed");
1078 assert_eq!(metrics.len(), 1);
1079
1080 let MetricValue::Sketch { sketch } = &metrics[0].values[0].1 else {
1081 panic!("expected sketch value");
1082 };
1083 assert_eq!(sketch.count(), 4);
1084 assert_eq!(sketch.sum(), Some(10.0));
1085 assert_eq!(sketch.min(), Some(1.0));
1086 assert_eq!(sketch.max(), Some(4.0));
1087 assert_eq!(sketch.avg(), Some(2.5));
1088 }
1089
1090 fn length_prefixed_strings(strings: impl IntoIterator<Item = &'static str>) -> Vec<u8> {
1091 let mut bytes = Vec::new();
1092 for s in strings {
1093 bytes.push(s.len() as u8);
1094 bytes.extend_from_slice(s.as_bytes());
1095 }
1096 bytes
1097 }
1098
1099 #[test]
1100 fn parse_tagsets_resolves_positive_indices() {
1101 let tags_dict = vec!["a:1".to_string(), "b:2".to_string(), "c:3".to_string()];
1104 let dict_tagsets = vec![1, 1, 2, 1, 1 ];
1105 let tagsets = parse_tagsets(&dict_tagsets, &tags_dict).expect("parse should succeed");
1106 assert_eq!(
1107 tagsets,
1108 vec![vec!["a:1".to_string()], vec!["a:1".to_string(), "b:2".to_string()]]
1109 );
1110 }
1111
1112 #[test]
1113 fn parse_tagsets_resolves_negative_prefix_reference() {
1114 let tags_dict = vec!["a:1".to_string(), "b:2".to_string()];
1118 let dict_tagsets = vec![
1119 1, 1, 2, -1, 3, ];
1122 let tagsets = parse_tagsets(&dict_tagsets, &tags_dict).expect("parse should succeed");
1123 assert_eq!(tagsets.len(), 2);
1124 assert_eq!(tagsets[0], vec!["a:1".to_string()]);
1125 assert_eq!(tagsets[1], vec!["a:1".to_string(), "b:2".to_string()]);
1127 }
1128
1129 #[test]
1130 fn parse_tagsets_rejects_forward_prefix_reference() {
1131 let tags_dict = vec!["a:1".to_string()];
1133 let dict_tagsets = vec![1, -5];
1134 assert!(parse_tagsets(&dict_tagsets, &tags_dict).is_err());
1135 }
1136
1137 fn single_scalar_payload(type_field: u64, sint64: Vec<i64>, float32: Vec<f32>, float64: Vec<f64>) -> V3Payload {
1140 use datadog_protos::metrics::v3::{MetricData, Payload};
1141
1142 let mut data = MetricData::new();
1143 data.dictNameStr = length_prefixed_strings(["m"]);
1144 data.types = vec![type_field];
1145 data.nameRefs = vec![1];
1146 data.tagsetRefs = vec![0];
1147 data.resourcesRefs = vec![0];
1148 data.intervals = vec![10];
1149 data.numPoints = vec![1];
1150 data.timestamps = vec![1000];
1151 data.valsSint64 = sint64;
1152 data.valsFloat32 = float32;
1153 data.valsFloat64 = float64;
1154
1155 let mut payload = Payload::new();
1156 payload.metricData = Some(data).into();
1157 payload
1158 }
1159
1160 #[test]
1161 fn try_from_v3_decodes_all_scalar_type_and_value_combinations() {
1162 struct Case {
1167 name: &'static str,
1168 value_type: u64,
1169 sint64: Vec<i64>,
1170 float32: Vec<f32>,
1171 float64: Vec<f64>,
1172 value: f64,
1173 }
1174
1175 let value_cases = [
1176 Case {
1177 name: "zero",
1178 value_type: V3_VALUE_TYPE_ZERO,
1179 sint64: vec![],
1180 float32: vec![],
1181 float64: vec![],
1182 value: 0.0,
1183 },
1184 Case {
1185 name: "sint64",
1186 value_type: V3_VALUE_TYPE_SINT64,
1187 sint64: vec![7],
1188 float32: vec![],
1189 float64: vec![],
1190 value: 7.0,
1191 },
1192 Case {
1193 name: "float32",
1194 value_type: V3_VALUE_TYPE_FLOAT32,
1195 sint64: vec![],
1196 float32: vec![1.5],
1197 float64: vec![],
1198 value: 1.5,
1199 },
1200 Case {
1201 name: "float64",
1202 value_type: V3_VALUE_TYPE_FLOAT64,
1203 sint64: vec![],
1204 float32: vec![],
1205 float64: vec![2.25],
1206 value: 2.25,
1207 },
1208 ];
1209
1210 for case in &value_cases {
1211 let expectations: [(&str, u64, MetricValue); 3] = [
1212 ("count", V3_METRIC_TYPE_COUNT, MetricValue::Count { value: case.value }),
1213 (
1214 "rate",
1215 V3_METRIC_TYPE_RATE,
1216 MetricValue::Rate {
1217 interval: 10,
1218 value: case.value,
1219 },
1220 ),
1221 ("gauge", V3_METRIC_TYPE_GAUGE, MetricValue::Gauge { value: case.value }),
1222 ];
1223
1224 for (metric_name, metric_type, expected) in expectations {
1225 let label = format!("{metric_name}/{}", case.name);
1226 let payload = single_scalar_payload(
1227 metric_type | case.value_type,
1228 case.sint64.clone(),
1229 case.float32.clone(),
1230 case.float64.clone(),
1231 );
1232 let metrics =
1233 Metric::try_from_v3(payload).unwrap_or_else(|e| panic!("{label}: decode should succeed: {e}"));
1234 assert_eq!(metrics.len(), 1, "{label}: metric count");
1235 assert_eq!(metrics[0].values.len(), 1, "{label}: point count");
1236 assert_eq!(metrics[0].values[0].0, 1000, "{label}: timestamp");
1237 assert_eq!(metrics[0].values[0].1, expected, "{label}: value");
1238 }
1239 }
1240 }
1241
1242 #[test]
1243 fn try_from_v3_decodes_sketch_summaries_across_value_encodings() {
1244 use datadog_protos::metrics::v3::{MetricData, Payload};
1248
1249 struct SketchCase {
1250 name: &'static str,
1251 value_type: u64,
1252 sint64: Vec<i64>,
1253 float32: Vec<f32>,
1254 float64: Vec<f64>,
1255 expected_sum: f64,
1256 expected_min: f64,
1257 expected_max: f64,
1258 }
1259
1260 let cases = [
1261 SketchCase {
1262 name: "zero",
1263 value_type: V3_VALUE_TYPE_ZERO,
1264 sint64: vec![4], float32: vec![],
1266 float64: vec![],
1267 expected_sum: 0.0,
1268 expected_min: 0.0,
1269 expected_max: 0.0,
1270 },
1271 SketchCase {
1272 name: "float32",
1273 value_type: V3_VALUE_TYPE_FLOAT32,
1274 sint64: vec![4], float32: vec![10.0, 1.0, 4.0], float64: vec![],
1277 expected_sum: 10.0,
1278 expected_min: 1.0,
1279 expected_max: 4.0,
1280 },
1281 SketchCase {
1282 name: "float64",
1283 value_type: V3_VALUE_TYPE_FLOAT64,
1284 sint64: vec![4], float32: vec![],
1286 float64: vec![10.0, 1.0, 4.0], expected_sum: 10.0,
1288 expected_min: 1.0,
1289 expected_max: 4.0,
1290 },
1291 ];
1292
1293 for case in &cases {
1294 let mut data = MetricData::new();
1295 data.dictNameStr = length_prefixed_strings(["s"]);
1296 data.types = vec![V3_METRIC_TYPE_SKETCH | case.value_type];
1297 data.nameRefs = vec![1];
1298 data.tagsetRefs = vec![0];
1299 data.resourcesRefs = vec![0];
1300 data.intervals = vec![0];
1301 data.numPoints = vec![1];
1302 data.timestamps = vec![123];
1303 data.valsSint64 = case.sint64.clone();
1304 data.valsFloat32 = case.float32.clone();
1305 data.valsFloat64 = case.float64.clone();
1306 data.sketchNumBins = vec![1];
1307 data.sketchBinKeys = vec![0];
1308 data.sketchBinCnts = vec![4];
1309
1310 let mut payload = Payload::new();
1311 payload.metricData = Some(data).into();
1312
1313 let metrics =
1314 Metric::try_from_v3(payload).unwrap_or_else(|e| panic!("{}: decode should succeed: {e}", case.name));
1315 assert_eq!(metrics.len(), 1, "{}: metric count", case.name);
1316
1317 let MetricValue::Sketch { sketch } = &metrics[0].values[0].1 else {
1318 panic!("{}: expected a sketch value", case.name);
1319 };
1320 assert_eq!(sketch.count(), 4, "{}: count", case.name);
1321 assert_eq!(sketch.sum(), Some(case.expected_sum), "{}: sum", case.name);
1322 assert_eq!(sketch.min(), Some(case.expected_min), "{}: min", case.name);
1323 assert_eq!(sketch.max(), Some(case.expected_max), "{}: max", case.name);
1324 }
1325 }
1326
1327 #[test]
1328 fn metric_value_equality_honors_the_documented_ratio_tolerance() {
1329 const RATIO: f64 = 0.00000001;
1332 let base = 1_000.0;
1333 let within = base * (1.0 - RATIO / 2.0); let outside = base * (1.0 - RATIO * 2.0); assert_eq!(
1337 MetricValue::Count { value: base },
1338 MetricValue::Count { value: within },
1339 "difference just inside the ratio tolerance must compare equal"
1340 );
1341 assert_ne!(
1342 MetricValue::Count { value: base },
1343 MetricValue::Count { value: outside },
1344 "difference just outside the ratio tolerance must compare not-equal"
1345 );
1346
1347 assert_eq!(MetricValue::Gauge { value: base }, MetricValue::Gauge { value: within });
1349 assert_ne!(
1350 MetricValue::Gauge { value: base },
1351 MetricValue::Gauge { value: outside }
1352 );
1353 }
1354}