1use bytes::Bytes;
30use yaml_rust2::Yaml;
31
32use mira_proto::collector::logs::v1::ExportLogsServiceRequest;
33use mira_proto::collector::metrics::v1::ExportMetricsServiceRequest;
34use mira_proto::collector::trace::v1::ExportTraceServiceRequest;
35use mira_proto::common::v1::{AnyValue, ArrayValue, InstrumentationScope, KeyValue, KeyValueList};
36use mira_proto::logs::v1::{LogRecord, ResourceLogs, ScopeLogs};
37use mira_proto::metrics::v1::{
38 Exemplar, ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Histogram,
39 HistogramDataPoint, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, Sum, Summary,
40 SummaryDataPoint, exponential_histogram_data_point, metric, number_data_point,
41 summary_data_point,
42};
43use mira_proto::resource::v1::Resource;
44use mira_proto::trace::v1::{ResourceSpans, ScopeSpans, Span, Status, span};
45
46type R<T> = Result<T, String>;
47
48fn f<'a>(y: &'a Yaml, camel: &'static str, snake: &'static str) -> &'a Yaml {
57 match &y[camel] {
58 Yaml::BadValue => &y[snake],
59 v => v,
60 }
61}
62
63fn list(y: &Yaml) -> &[Yaml] {
66 y.as_vec().map_or(&[], Vec::as_slice)
67}
68
69fn top<'a>(doc: &'a Yaml, camel: &'static str, snake: &'static str) -> R<&'a [Yaml]> {
85 if doc.as_hash().is_none() {
86 return Err(format!(
87 "an OTLP JSON body must be a mapping carrying {camel:?}"
88 ));
89 }
90 match f(doc, camel, snake) {
91 Yaml::BadValue => Err(format!(
92 "missing {camel:?}: an OTLP JSON body carries exactly that member, \
93 and an export with no records in it is {{{camel:?}: []}}"
94 )),
95 Yaml::Null => Ok(&[]),
98 y => y
99 .as_vec()
100 .map(Vec::as_slice)
101 .ok_or_else(|| format!("{camel:?} must be a list")),
102 }
103}
104
105fn missing(y: &Yaml) -> bool {
106 matches!(y, Yaml::BadValue | Yaml::Null)
107}
108
109fn s(y: &Yaml) -> String {
110 y.as_str().unwrap_or_default().to_owned()
111}
112
113fn boolean(y: &Yaml) -> bool {
114 y.as_bool().unwrap_or_default()
115}
116
117fn int(y: &Yaml, what: &'static str) -> R<i64> {
120 match y {
121 Yaml::BadValue | Yaml::Null => Ok(0),
122 Yaml::Integer(n) => Ok(*n),
123 Yaml::String(t) if t.is_empty() => Ok(0),
124 Yaml::String(t) => t
125 .parse()
126 .map_err(|_| format!("{what}: {t:?} is not an integer")),
127 other => Err(format!("{what}: expected an integer, got {other:?}")),
128 }
129}
130
131fn uint(y: &Yaml, what: &'static str) -> R<u64> {
135 match y {
136 Yaml::Integer(n) if *n >= 0 => Ok(*n as u64),
137 Yaml::String(t) if !t.is_empty() => t
138 .parse()
139 .map_err(|_| format!("{what}: {t:?} is not an unsigned integer")),
140 _ => Ok(int(y, what)?.try_into().unwrap_or_default()),
141 }
142}
143
144fn u32f(y: &Yaml, what: &'static str) -> R<u32> {
145 let n = uint(y, what)?;
146 u32::try_from(n).map_err(|_| format!("{what}: {n} does not fit in 32 bits"))
147}
148
149fn i32f(y: &Yaml, what: &'static str) -> R<i32> {
150 let n = int(y, what)?;
151 i32::try_from(n).map_err(|_| format!("{what}: {n} does not fit in 32 bits"))
152}
153
154fn float(y: &Yaml, what: &'static str) -> R<f64> {
157 match y {
158 Yaml::BadValue | Yaml::Null => Ok(0.0),
159 Yaml::Real(_) | Yaml::Integer(_) => Ok(y.as_f64().unwrap_or_default()),
160 Yaml::String(t) => match t.as_str() {
161 "NaN" => Ok(f64::NAN),
162 "Infinity" => Ok(f64::INFINITY),
163 "-Infinity" => Ok(f64::NEG_INFINITY),
164 "" => Ok(0.0),
165 _ => t
166 .parse()
167 .map_err(|_| format!("{what}: {t:?} is not a number")),
168 },
169 other => Err(format!("{what}: expected a number, got {other:?}")),
170 }
171}
172
173fn hex(y: &Yaml, want: usize, what: &'static str) -> R<Bytes> {
180 let t = match y {
181 Yaml::BadValue | Yaml::Null => return Ok(Bytes::new()),
182 Yaml::String(t) if t.is_empty() => return Ok(Bytes::new()),
183 Yaml::String(t) => t,
184 other => return Err(format!("{what}: expected a hex string, got {other:?}")),
185 };
186 if t.len() != want * 2 {
187 return Err(format!(
188 "{what}: expected {} hex characters, got {}",
189 want * 2,
190 t.len()
191 ));
192 }
193 let mut out = Vec::with_capacity(want);
194 let b = t.as_bytes();
195 for pair in b.chunks_exact(2) {
196 let nib = |c: u8| match c {
197 b'0'..=b'9' => Ok(c - b'0'),
198 b'a'..=b'f' => Ok(c - b'a' + 10),
199 b'A'..=b'F' => Ok(c - b'A' + 10),
200 _ => Err(format!("{what}: {t:?} is not hex")),
201 };
202 out.push(nib(pair[0])? << 4 | nib(pair[1])?);
203 }
204 Ok(Bytes::from(out))
205}
206
207fn base64(y: &Yaml, what: &'static str) -> R<Bytes> {
211 let Some(t) = y.as_str() else {
212 return Ok(Bytes::new());
213 };
214 let mut out = Vec::with_capacity(t.len() / 4 * 3);
215 let (mut acc, mut bits) = (0u32, 0u32);
216 for c in t.bytes() {
217 let v = match c {
218 b'A'..=b'Z' => c - b'A',
219 b'a'..=b'z' => c - b'a' + 26,
220 b'0'..=b'9' => c - b'0' + 52,
221 b'+' | b'-' => 62,
222 b'/' | b'_' => 63,
223 b'=' | b'\r' | b'\n' => continue,
224 _ => return Err(format!("{what}: not base64")),
225 };
226 acc = acc << 6 | v as u32;
227 bits += 6;
228 if bits >= 8 {
229 bits -= 8;
230 out.push((acc >> bits) as u8);
231 }
232 }
233 Ok(Bytes::from(out))
234}
235
236fn enumerate(y: &Yaml, names: &[&str], what: &'static str) -> R<i32> {
239 match y {
240 Yaml::BadValue | Yaml::Null => Ok(0),
241 Yaml::String(t) => names
242 .iter()
243 .position(|n| *n == t)
244 .map(|i| i as i32)
245 .or_else(|| t.parse().ok())
247 .ok_or_else(|| format!("{what}: {t:?} is not a known value")),
248 other => i32f(other, what),
249 }
250}
251
252fn any_value(y: &Yaml) -> R<Option<AnyValue>> {
255 use mira_proto::common::v1::any_value::Value;
256 if missing(y) {
257 return Ok(None);
258 }
259 let v = if !missing(f(y, "stringValue", "string_value")) {
262 Value::StringValue(s(f(y, "stringValue", "string_value")))
263 } else if !missing(f(y, "boolValue", "bool_value")) {
264 Value::BoolValue(boolean(f(y, "boolValue", "bool_value")))
265 } else if !missing(f(y, "intValue", "int_value")) {
266 Value::IntValue(int(f(y, "intValue", "int_value"), "intValue")?)
267 } else if !missing(f(y, "doubleValue", "double_value")) {
268 Value::DoubleValue(float(f(y, "doubleValue", "double_value"), "doubleValue")?)
269 } else if !missing(f(y, "arrayValue", "array_value")) {
270 let vs = &f(y, "arrayValue", "array_value")["values"];
271 Value::ArrayValue(ArrayValue {
272 values: list(vs)
273 .iter()
274 .map(|v| Ok(any_value(v)?.unwrap_or_default()))
275 .collect::<R<_>>()?,
276 })
277 } else if !missing(f(y, "kvlistValue", "kvlist_value")) {
278 let vs = &f(y, "kvlistValue", "kvlist_value")["values"];
279 Value::KvlistValue(KeyValueList {
280 values: key_values(vs)?,
281 })
282 } else if !missing(f(y, "bytesValue", "bytes_value")) {
283 Value::BytesValue(base64(f(y, "bytesValue", "bytes_value"), "bytesValue")?)
284 } else {
285 return Ok(Some(AnyValue::default()));
288 };
289 Ok(Some(AnyValue { value: Some(v) }))
290}
291
292fn key_values(y: &Yaml) -> R<Vec<KeyValue>> {
293 list(y)
294 .iter()
295 .map(|kv| {
296 Ok(KeyValue {
297 key: s(&kv["key"]),
298 value: any_value(&kv["value"])?,
299 })
300 })
301 .collect()
302}
303
304fn resource(y: &Yaml) -> R<Option<Resource>> {
305 if missing(y) {
306 return Ok(None);
307 }
308 Ok(Some(Resource {
309 attributes: key_values(&y["attributes"])?,
310 dropped_attributes_count: u32f(
311 f(y, "droppedAttributesCount", "dropped_attributes_count"),
312 "droppedAttributesCount",
313 )?,
314 ..Default::default()
317 }))
318}
319
320fn scope(y: &Yaml) -> R<Option<InstrumentationScope>> {
321 if missing(y) {
322 return Ok(None);
323 }
324 Ok(Some(InstrumentationScope {
325 name: s(&y["name"]),
326 version: s(&y["version"]),
327 attributes: key_values(&y["attributes"])?,
328 dropped_attributes_count: u32f(
329 f(y, "droppedAttributesCount", "dropped_attributes_count"),
330 "droppedAttributesCount",
331 )?,
332 }))
333}
334
335fn dropped(y: &Yaml, camel: &'static str, snake: &'static str) -> R<u32> {
336 u32f(f(y, camel, snake), camel)
337}
338
339const SEVERITY: [&str; 25] = [
342 "SEVERITY_NUMBER_UNSPECIFIED",
343 "SEVERITY_NUMBER_TRACE",
344 "SEVERITY_NUMBER_TRACE2",
345 "SEVERITY_NUMBER_TRACE3",
346 "SEVERITY_NUMBER_TRACE4",
347 "SEVERITY_NUMBER_DEBUG",
348 "SEVERITY_NUMBER_DEBUG2",
349 "SEVERITY_NUMBER_DEBUG3",
350 "SEVERITY_NUMBER_DEBUG4",
351 "SEVERITY_NUMBER_INFO",
352 "SEVERITY_NUMBER_INFO2",
353 "SEVERITY_NUMBER_INFO3",
354 "SEVERITY_NUMBER_INFO4",
355 "SEVERITY_NUMBER_WARN",
356 "SEVERITY_NUMBER_WARN2",
357 "SEVERITY_NUMBER_WARN3",
358 "SEVERITY_NUMBER_WARN4",
359 "SEVERITY_NUMBER_ERROR",
360 "SEVERITY_NUMBER_ERROR2",
361 "SEVERITY_NUMBER_ERROR3",
362 "SEVERITY_NUMBER_ERROR4",
363 "SEVERITY_NUMBER_FATAL",
364 "SEVERITY_NUMBER_FATAL2",
365 "SEVERITY_NUMBER_FATAL3",
366 "SEVERITY_NUMBER_FATAL4",
367];
368
369pub fn logs(doc: &Yaml) -> R<ExportLogsServiceRequest> {
370 Ok(ExportLogsServiceRequest {
371 resource_logs: top(doc, "resourceLogs", "resource_logs")?
372 .iter()
373 .map(|rl| {
374 Ok(ResourceLogs {
375 resource: resource(&rl["resource"])?,
376 scope_logs: list(f(rl, "scopeLogs", "scope_logs"))
377 .iter()
378 .map(scope_logs)
379 .collect::<R<_>>()?,
380 schema_url: s(f(rl, "schemaUrl", "schema_url")),
381 })
382 })
383 .collect::<R<_>>()?,
384 })
385}
386
387fn scope_logs(sl: &Yaml) -> R<ScopeLogs> {
388 Ok(ScopeLogs {
389 scope: scope(&sl["scope"])?,
390 log_records: list(f(sl, "logRecords", "log_records"))
391 .iter()
392 .map(log_record)
393 .collect::<R<_>>()?,
394 schema_url: s(f(sl, "schemaUrl", "schema_url")),
395 })
396}
397
398fn log_record(r: &Yaml) -> R<LogRecord> {
399 Ok(LogRecord {
400 time_unix_nano: uint(f(r, "timeUnixNano", "time_unix_nano"), "timeUnixNano")?,
401 observed_time_unix_nano: uint(
402 f(r, "observedTimeUnixNano", "observed_time_unix_nano"),
403 "observedTimeUnixNano",
404 )?,
405 severity_number: enumerate(
406 f(r, "severityNumber", "severity_number"),
407 &SEVERITY,
408 "severityNumber",
409 )?,
410 severity_text: s(f(r, "severityText", "severity_text")),
411 body: any_value(&r["body"])?,
412 attributes: key_values(&r["attributes"])?,
413 dropped_attributes_count: dropped(r, "droppedAttributesCount", "dropped_attributes_count")?,
414 flags: u32f(&r["flags"], "flags")?,
415 trace_id: hex(f(r, "traceId", "trace_id"), 16, "traceId")?,
416 span_id: hex(f(r, "spanId", "span_id"), 8, "spanId")?,
417 event_name: s(f(r, "eventName", "event_name")),
418 })
419}
420
421const SPAN_KIND: [&str; 6] = [
424 "SPAN_KIND_UNSPECIFIED",
425 "SPAN_KIND_INTERNAL",
426 "SPAN_KIND_SERVER",
427 "SPAN_KIND_CLIENT",
428 "SPAN_KIND_PRODUCER",
429 "SPAN_KIND_CONSUMER",
430];
431const STATUS_CODE: [&str; 3] = ["STATUS_CODE_UNSET", "STATUS_CODE_OK", "STATUS_CODE_ERROR"];
432
433pub fn traces(doc: &Yaml) -> R<ExportTraceServiceRequest> {
434 Ok(ExportTraceServiceRequest {
435 resource_spans: top(doc, "resourceSpans", "resource_spans")?
436 .iter()
437 .map(|rs| {
438 Ok(ResourceSpans {
439 resource: resource(&rs["resource"])?,
440 scope_spans: list(f(rs, "scopeSpans", "scope_spans"))
441 .iter()
442 .map(scope_spans)
443 .collect::<R<_>>()?,
444 schema_url: s(f(rs, "schemaUrl", "schema_url")),
445 })
446 })
447 .collect::<R<_>>()?,
448 })
449}
450
451fn scope_spans(ss: &Yaml) -> R<ScopeSpans> {
452 Ok(ScopeSpans {
453 scope: scope(&ss["scope"])?,
454 spans: list(&ss["spans"]).iter().map(span).collect::<R<_>>()?,
455 schema_url: s(f(ss, "schemaUrl", "schema_url")),
456 })
457}
458
459fn span(sp: &Yaml) -> R<Span> {
460 Ok(Span {
461 trace_id: hex(f(sp, "traceId", "trace_id"), 16, "traceId")?,
462 span_id: hex(f(sp, "spanId", "span_id"), 8, "spanId")?,
463 trace_state: s(f(sp, "traceState", "trace_state")),
464 parent_span_id: hex(f(sp, "parentSpanId", "parent_span_id"), 8, "parentSpanId")?,
465 flags: u32f(&sp["flags"], "flags")?,
466 name: s(&sp["name"]),
467 kind: enumerate(&sp["kind"], &SPAN_KIND, "kind")?,
468 start_time_unix_nano: uint(
469 f(sp, "startTimeUnixNano", "start_time_unix_nano"),
470 "startTimeUnixNano",
471 )?,
472 end_time_unix_nano: uint(
473 f(sp, "endTimeUnixNano", "end_time_unix_nano"),
474 "endTimeUnixNano",
475 )?,
476 attributes: key_values(&sp["attributes"])?,
477 dropped_attributes_count: dropped(
478 sp,
479 "droppedAttributesCount",
480 "dropped_attributes_count",
481 )?,
482 events: list(&sp["events"])
483 .iter()
484 .map(|e| {
485 Ok(span::Event {
486 time_unix_nano: uint(f(e, "timeUnixNano", "time_unix_nano"), "timeUnixNano")?,
487 name: s(&e["name"]),
488 attributes: key_values(&e["attributes"])?,
489 dropped_attributes_count: dropped(
490 e,
491 "droppedAttributesCount",
492 "dropped_attributes_count",
493 )?,
494 })
495 })
496 .collect::<R<_>>()?,
497 dropped_events_count: dropped(sp, "droppedEventsCount", "dropped_events_count")?,
498 links: list(&sp["links"])
499 .iter()
500 .map(|l| {
501 Ok(span::Link {
502 trace_id: hex(f(l, "traceId", "trace_id"), 16, "traceId")?,
503 span_id: hex(f(l, "spanId", "span_id"), 8, "spanId")?,
504 trace_state: s(f(l, "traceState", "trace_state")),
505 attributes: key_values(&l["attributes"])?,
506 dropped_attributes_count: dropped(
507 l,
508 "droppedAttributesCount",
509 "dropped_attributes_count",
510 )?,
511 flags: u32f(&l["flags"], "flags")?,
512 })
513 })
514 .collect::<R<_>>()?,
515 dropped_links_count: dropped(sp, "droppedLinksCount", "dropped_links_count")?,
516 status: match &sp["status"] {
517 y if missing(y) => None,
518 y => Some(Status {
519 message: s(&y["message"]),
520 code: enumerate(&y["code"], &STATUS_CODE, "code")?,
521 }),
522 },
523 })
524}
525
526const TEMPORALITY: [&str; 3] = [
529 "AGGREGATION_TEMPORALITY_UNSPECIFIED",
530 "AGGREGATION_TEMPORALITY_DELTA",
531 "AGGREGATION_TEMPORALITY_CUMULATIVE",
532];
533
534pub fn metrics(doc: &Yaml) -> R<ExportMetricsServiceRequest> {
535 Ok(ExportMetricsServiceRequest {
536 resource_metrics: top(doc, "resourceMetrics", "resource_metrics")?
537 .iter()
538 .map(|rm| {
539 Ok(ResourceMetrics {
540 resource: resource(&rm["resource"])?,
541 scope_metrics: list(f(rm, "scopeMetrics", "scope_metrics"))
542 .iter()
543 .map(scope_metrics)
544 .collect::<R<_>>()?,
545 schema_url: s(f(rm, "schemaUrl", "schema_url")),
546 })
547 })
548 .collect::<R<_>>()?,
549 })
550}
551
552fn scope_metrics(sm: &Yaml) -> R<ScopeMetrics> {
553 Ok(ScopeMetrics {
554 scope: scope(&sm["scope"])?,
555 metrics: list(&sm["metrics"])
556 .iter()
557 .map(metric_of)
558 .collect::<R<_>>()?,
559 schema_url: s(f(sm, "schemaUrl", "schema_url")),
560 })
561}
562
563fn temporality(y: &Yaml) -> R<i32> {
564 enumerate(
565 f(y, "aggregationTemporality", "aggregation_temporality"),
566 &TEMPORALITY,
567 "aggregationTemporality",
568 )
569}
570
571fn metric_of(m: &Yaml) -> R<Metric> {
572 let data = if !missing(&m["gauge"]) {
573 Some(metric::Data::Gauge(Gauge {
574 data_points: number_points(&m["gauge"])?,
575 }))
576 } else if !missing(&m["sum"]) {
577 let g = &m["sum"];
578 Some(metric::Data::Sum(Sum {
579 data_points: number_points(g)?,
580 aggregation_temporality: temporality(g)?,
581 is_monotonic: boolean(f(g, "isMonotonic", "is_monotonic")),
582 }))
583 } else if !missing(&m["histogram"]) {
584 let g = &m["histogram"];
585 Some(metric::Data::Histogram(Histogram {
586 data_points: histogram_points(g)?,
587 aggregation_temporality: temporality(g)?,
588 }))
589 } else if !missing(f(m, "exponentialHistogram", "exponential_histogram")) {
590 let g = f(m, "exponentialHistogram", "exponential_histogram");
591 Some(metric::Data::ExponentialHistogram(ExponentialHistogram {
592 data_points: exp_histogram_points(g)?,
593 aggregation_temporality: temporality(g)?,
594 }))
595 } else if !missing(&m["summary"]) {
596 Some(metric::Data::Summary(Summary {
597 data_points: summary_points(&m["summary"])?,
598 }))
599 } else {
600 None
602 };
603 Ok(Metric {
604 name: s(&m["name"]),
605 description: s(&m["description"]),
606 unit: s(&m["unit"]),
607 metadata: key_values(&m["metadata"])?,
608 data,
609 })
610}
611
612fn data_points(g: &Yaml) -> &[Yaml] {
613 list(f(g, "dataPoints", "data_points"))
614}
615
616fn opt_float(y: &Yaml, what: &'static str) -> R<Option<f64>> {
619 if missing(y) {
620 Ok(None)
621 } else {
622 Ok(Some(float(y, what)?))
623 }
624}
625
626fn exemplars(p: &Yaml) -> R<Vec<Exemplar>> {
627 list(&p["exemplars"])
628 .iter()
629 .map(|e| {
630 Ok(Exemplar {
631 filtered_attributes: key_values(f(e, "filteredAttributes", "filtered_attributes"))?,
632 time_unix_nano: uint(f(e, "timeUnixNano", "time_unix_nano"), "timeUnixNano")?,
633 value: number_value(e, "exemplar")?.map(|v| match v {
634 number_data_point::Value::AsDouble(d) => {
635 mira_proto::metrics::v1::exemplar::Value::AsDouble(d)
636 }
637 number_data_point::Value::AsInt(i) => {
638 mira_proto::metrics::v1::exemplar::Value::AsInt(i)
639 }
640 }),
641 span_id: hex(f(e, "spanId", "span_id"), 8, "spanId")?,
642 trace_id: hex(f(e, "traceId", "trace_id"), 16, "traceId")?,
643 })
644 })
645 .collect()
646}
647
648fn number_value(p: &Yaml, what: &'static str) -> R<Option<number_data_point::Value>> {
650 if !missing(f(p, "asDouble", "as_double")) {
651 Ok(Some(number_data_point::Value::AsDouble(float(
652 f(p, "asDouble", "as_double"),
653 what,
654 )?)))
655 } else if !missing(f(p, "asInt", "as_int")) {
656 Ok(Some(number_data_point::Value::AsInt(int(
657 f(p, "asInt", "as_int"),
658 what,
659 )?)))
660 } else {
661 Ok(None)
662 }
663}
664
665fn number_points(g: &Yaml) -> R<Vec<NumberDataPoint>> {
666 data_points(g)
667 .iter()
668 .map(|p| {
669 Ok(NumberDataPoint {
670 attributes: key_values(&p["attributes"])?,
671 start_time_unix_nano: uint(
672 f(p, "startTimeUnixNano", "start_time_unix_nano"),
673 "startTimeUnixNano",
674 )?,
675 time_unix_nano: uint(f(p, "timeUnixNano", "time_unix_nano"), "timeUnixNano")?,
676 value: number_value(p, "dataPoint")?,
677 exemplars: exemplars(p)?,
678 flags: u32f(&p["flags"], "flags")?,
679 })
680 })
681 .collect()
682}
683
684fn floats(y: &Yaml, what: &'static str) -> R<Vec<f64>> {
685 list(y).iter().map(|v| float(v, what)).collect()
686}
687
688fn uints(y: &Yaml, what: &'static str) -> R<Vec<u64>> {
689 list(y).iter().map(|v| uint(v, what)).collect()
690}
691
692fn histogram_points(g: &Yaml) -> R<Vec<HistogramDataPoint>> {
693 data_points(g)
694 .iter()
695 .map(|p| {
696 Ok(HistogramDataPoint {
697 attributes: key_values(&p["attributes"])?,
698 start_time_unix_nano: uint(
699 f(p, "startTimeUnixNano", "start_time_unix_nano"),
700 "startTimeUnixNano",
701 )?,
702 time_unix_nano: uint(f(p, "timeUnixNano", "time_unix_nano"), "timeUnixNano")?,
703 count: uint(&p["count"], "count")?,
704 sum: opt_float(&p["sum"], "sum")?,
705 bucket_counts: uints(f(p, "bucketCounts", "bucket_counts"), "bucketCounts")?,
706 explicit_bounds: floats(
707 f(p, "explicitBounds", "explicit_bounds"),
708 "explicitBounds",
709 )?,
710 exemplars: exemplars(p)?,
711 flags: u32f(&p["flags"], "flags")?,
712 min: opt_float(&p["min"], "min")?,
713 max: opt_float(&p["max"], "max")?,
714 })
715 })
716 .collect()
717}
718
719fn buckets(y: &Yaml) -> R<Option<exponential_histogram_data_point::Buckets>> {
720 if missing(y) {
721 return Ok(None);
722 }
723 Ok(Some(exponential_histogram_data_point::Buckets {
724 offset: i32f(&y["offset"], "offset")?,
725 bucket_counts: uints(f(y, "bucketCounts", "bucket_counts"), "bucketCounts")?,
726 }))
727}
728
729fn exp_histogram_points(g: &Yaml) -> R<Vec<ExponentialHistogramDataPoint>> {
730 data_points(g)
731 .iter()
732 .map(|p| {
733 Ok(ExponentialHistogramDataPoint {
734 attributes: key_values(&p["attributes"])?,
735 start_time_unix_nano: uint(
736 f(p, "startTimeUnixNano", "start_time_unix_nano"),
737 "startTimeUnixNano",
738 )?,
739 time_unix_nano: uint(f(p, "timeUnixNano", "time_unix_nano"), "timeUnixNano")?,
740 count: uint(&p["count"], "count")?,
741 sum: opt_float(&p["sum"], "sum")?,
742 scale: i32f(&p["scale"], "scale")?,
743 zero_count: uint(f(p, "zeroCount", "zero_count"), "zeroCount")?,
744 positive: buckets(&p["positive"])?,
745 negative: buckets(&p["negative"])?,
746 flags: u32f(&p["flags"], "flags")?,
747 exemplars: exemplars(p)?,
748 min: opt_float(&p["min"], "min")?,
749 max: opt_float(&p["max"], "max")?,
750 zero_threshold: float(f(p, "zeroThreshold", "zero_threshold"), "zeroThreshold")?,
751 })
752 })
753 .collect()
754}
755
756fn summary_points(g: &Yaml) -> R<Vec<SummaryDataPoint>> {
757 data_points(g)
758 .iter()
759 .map(|p| {
760 Ok(SummaryDataPoint {
761 attributes: key_values(&p["attributes"])?,
762 start_time_unix_nano: uint(
763 f(p, "startTimeUnixNano", "start_time_unix_nano"),
764 "startTimeUnixNano",
765 )?,
766 time_unix_nano: uint(f(p, "timeUnixNano", "time_unix_nano"), "timeUnixNano")?,
767 count: uint(&p["count"], "count")?,
768 sum: float(&p["sum"], "sum")?,
769 quantile_values: list(f(p, "quantileValues", "quantile_values"))
770 .iter()
771 .map(|q| {
772 Ok(summary_data_point::ValueAtQuantile {
773 quantile: float(&q["quantile"], "quantile")?,
774 value: float(&q["value"], "value")?,
775 })
776 })
777 .collect::<R<_>>()?,
778 flags: u32f(&p["flags"], "flags")?,
779 })
780 })
781 .collect()
782}
783
784#[cfg(test)]
785mod tests {
786 use super::*;
787
788 fn doc(text: &str) -> Yaml {
789 crate::api::parse(text).expect("document")
790 }
791
792 #[test]
801 fn metrics_json_decodes_every_data_point_kind() {
802 use mira_proto::metrics::v1::exemplar;
803
804 let m = metrics(&doc(r#"{
805 "resourceMetrics": [{
806 "resource": {"attributes": [{"key":"service.name","value":{"stringValue":"m"}}],
807 "droppedAttributesCount": 2},
808 "scopeMetrics": [{
809 "scope": {"name":"s","version":"1",
810 "attributes":[{"key":"a","value":{"intValue":"1"}}]},
811 "schemaUrl": "https://schemas/1",
812 "metrics": [
813 {"name":"g","unit":"By","description":"a gauge",
814 "gauge":{"dataPoints":[{"timeUnixNano":"7","asDouble":1.5,
815 "exemplars":[{"timeUnixNano":"7","asInt":"3",
816 "traceId":"AABBCCDDEEFF00112233445566778899",
817 "spanId":"1122334455667788",
818 "filteredAttributes":[
819 {"key":"k","value":{"stringValue":"v"}}]}]},
820 {"timeUnixNano":"8",
821 "exemplars":[{"timeUnixNano":"8","asDouble":2.5}]}]}},
822 {"name":"c",
823 "sum":{"data_points":[{"time_unix_nano":8,"as_int":"9","flags":1}],
824 "aggregation_temporality":2,"is_monotonic":true}},
825 {"name":"h",
826 "histogram":{"dataPoints":[{"timeUnixNano":"9","count":"3","sum":"6.5",
827 "bucketCounts":["1","2"],"explicitBounds":[2.5],
828 "min":"NaN","max":"Infinity"}],
829 "aggregationTemporality":"AGGREGATION_TEMPORALITY_DELTA"}},
830 {"name":"e",
831 "exponentialHistogram":{"dataPoints":[{"timeUnixNano":"10","count":"4",
832 "scale":-1,"zeroCount":"1","zeroThreshold":"1e-9","min":"-Infinity",
833 "positive":{"offset":2,"bucketCounts":["1","3"]}}],
834 "aggregationTemporality":1}},
835 {"name":"q",
836 "summary":{"dataPoints":[{"timeUnixNano":"11","count":"5","sum":12.5,
837 "quantileValues":[{"quantile":0.99,"value":42.0}]}]}},
838 {"name":"nothing"}
839 ]
840 }]
841 }]
842 }"#))
843 .unwrap();
844
845 let rm = &m.resource_metrics[0];
846 assert_eq!(rm.resource.as_ref().unwrap().dropped_attributes_count, 2);
847 let sm = &rm.scope_metrics[0];
848 assert_eq!(sm.schema_url, "https://schemas/1");
849 assert_eq!(sm.scope.as_ref().unwrap().attributes.len(), 1);
850 let ms = &sm.metrics;
851 assert_eq!(ms.len(), 6);
852
853 let mut seen = Vec::new();
857 for m in ms {
858 seen.push(m.name.as_str());
859 match &m.data {
860 Some(metric::Data::Gauge(g)) => {
861 assert_eq!(m.name, "g");
862 assert_eq!(m.unit, "By");
863 let p = &g.data_points[0];
864 assert_eq!(p.value, Some(number_data_point::Value::AsDouble(1.5)));
865 let ex = &p.exemplars[0];
866 assert_eq!(ex.trace_id[..4], [0xaa, 0xbb, 0xcc, 0xdd]);
871 assert_eq!(ex.trace_id.len(), 16);
872 assert_eq!(ex.span_id.len(), 8);
873 assert_eq!(ex.value, Some(exemplar::Value::AsInt(3)));
874 assert_eq!(ex.filtered_attributes[0].key, "k");
875 let p = &g.data_points[1];
880 assert!(p.value.is_none());
881 assert_eq!(p.exemplars[0].value, Some(exemplar::Value::AsDouble(2.5)));
882 }
883 Some(metric::Data::Sum(sum)) => {
884 assert_eq!(m.name, "c");
885 assert!(sum.is_monotonic);
886 assert_eq!(sum.aggregation_temporality, 2);
887 let p = &sum.data_points[0];
888 assert_eq!(p.time_unix_nano, 8);
889 assert_eq!(p.flags, 1);
890 assert_eq!(p.value, Some(number_data_point::Value::AsInt(9)));
894 }
895 Some(metric::Data::Histogram(h)) => {
896 assert_eq!(m.name, "h");
897 assert_eq!(h.aggregation_temporality, 1, "DELTA, spelled by name");
898 let p = &h.data_points[0];
899 assert_eq!(p.count, 3);
900 assert_eq!(p.sum, Some(6.5));
901 assert_eq!(p.bucket_counts, [1, 2]);
902 assert_eq!(p.explicit_bounds, [2.5]);
903 assert!(p.min.unwrap().is_nan());
907 assert_eq!(p.max, Some(f64::INFINITY));
908 }
909 Some(metric::Data::ExponentialHistogram(e)) => {
910 assert_eq!(m.name, "e");
911 let p = &e.data_points[0];
912 assert_eq!(p.scale, -1);
913 assert_eq!(p.zero_count, 1);
914 assert_eq!(p.zero_threshold, 1e-9);
915 assert_eq!(p.min, Some(f64::NEG_INFINITY));
916 let pos = p.positive.as_ref().unwrap();
917 assert_eq!(pos.offset, 2);
918 assert_eq!(pos.bucket_counts, [1, 3]);
919 assert!(p.negative.is_none());
923 }
924 Some(metric::Data::Summary(q)) => {
925 assert_eq!(m.name, "q");
926 let p = &q.data_points[0];
927 assert_eq!(p.count, 5);
928 assert_eq!(p.sum, 12.5);
929 assert_eq!(p.quantile_values[0].quantile, 0.99);
930 assert_eq!(p.quantile_values[0].value, 42.0);
931 }
932 None => assert_eq!(m.name, "nothing"),
935 }
936 }
937 assert_eq!(seen, ["g", "c", "h", "e", "q", "nothing"]);
940 }
941
942 #[test]
949 fn an_unrecognisable_export_is_refused_by_name() {
950 for body in [
953 "[1,2,3]",
954 r#""hello""#,
955 "42",
956 "null",
957 "{}",
958 r#"{"resource_log":[{"scopeLogs":[]}]}"#,
959 r#"{"resourceLog":[]}"#,
960 ] {
961 let e = logs(&doc(body)).unwrap_err();
962 assert!(e.contains("resourceLogs"), "{body}: {e}");
965 }
966 assert!(traces(&doc("{}")).unwrap_err().contains("resourceSpans"));
968 assert!(metrics(&doc("[]")).unwrap_err().contains("resourceMetrics"));
969
970 assert!(
973 logs(&doc(r#"{"resourceLogs":[]}"#))
974 .unwrap()
975 .resource_logs
976 .is_empty()
977 );
978 assert!(
979 traces(&doc(r#"{"resource_spans":null}"#))
980 .unwrap()
981 .resource_spans
982 .is_empty()
983 );
984 assert!(
987 metrics(&doc(r#"{"resourceMetrics":{"resource":{}}}"#))
988 .unwrap_err()
989 .contains("must be a list")
990 );
991 }
992
993 #[test]
997 fn the_scalar_readers_are_strict_only_where_leniency_would_lose_data() {
998 let y = |t: &str| doc(&format!("{{\"v\":{t}}}"))["v"].clone();
999 let none = Yaml::BadValue;
1000
1001 assert_eq!(
1004 hex(&y(r#""AaBbCcDd00112233""#), 8, "id").unwrap()[..],
1005 [0xaa, 0xbb, 0xcc, 0xdd, 0x00, 0x11, 0x22, 0x33]
1006 );
1007 assert!(hex(&none, 8, "id").unwrap().is_empty());
1008 assert!(hex(&y(r#""""#), 8, "id").unwrap().is_empty());
1009 assert!(
1010 hex(&y(r#""abcd""#), 8, "id")
1011 .unwrap_err()
1012 .contains("expected 16 hex characters, got 4")
1013 );
1014 assert!(
1015 hex(&y(r#""zzzzzzzzzzzzzzzz""#), 8, "id")
1016 .unwrap_err()
1017 .contains("is not hex")
1018 );
1019 assert!(hex(&y("17"), 8, "id").unwrap_err().contains("hex string"));
1020
1021 assert_eq!(base64(&y(r#""aGVsbG8=""#), "b").unwrap()[..], b"hello"[..]);
1024 assert_eq!(
1025 base64(&y(r#""-_8=""#), "b").unwrap()[..],
1026 [0xfb, 0xff],
1027 "URL-safe"
1028 );
1029 assert!(base64(&y("3"), "b").unwrap().is_empty());
1030 assert!(base64(&y(r#""!!""#), "b").unwrap_err().contains("base64"));
1031
1032 assert_eq!(int(&y(r#""-5""#), "n").unwrap(), -5);
1035 assert_eq!(int(&none, "n").unwrap(), 0);
1036 assert_eq!(int(&y(r#""""#), "n").unwrap(), 0);
1037 assert_eq!(
1038 uint(&y(r#""18446744073709551615""#), "n").unwrap(),
1039 u64::MAX
1040 );
1041 assert_eq!(uint(&y("-1"), "n").unwrap(), 0, "clamped, not wrapped");
1042 assert!(int(&y("[1]"), "n").unwrap_err().contains("expected an int"));
1043 assert!(
1044 int(&y(r#""x""#), "n")
1045 .unwrap_err()
1046 .contains("not an integer")
1047 );
1048 assert!(u32f(&y(r#""4294967296""#), "n").unwrap_err().contains("32"));
1049 assert!(i32f(&y(r#""2147483648""#), "n").unwrap_err().contains("32"));
1050
1051 assert_eq!(float(&y("1.5"), "d").unwrap(), 1.5);
1053 assert_eq!(float(&y(r#""1e-9""#), "d").unwrap(), 1e-9);
1054 assert_eq!(float(&none, "d").unwrap(), 0.0);
1055 assert_eq!(float(&y(r#""""#), "d").unwrap(), 0.0);
1056 assert!(float(&y(r#""NaN""#), "d").unwrap().is_nan());
1057 assert!(
1058 float(&y(r#""x""#), "d")
1059 .unwrap_err()
1060 .contains("not a number")
1061 );
1062 assert!(
1063 float(&y("[1]"), "d")
1064 .unwrap_err()
1065 .contains("expected a num")
1066 );
1067
1068 assert_eq!(
1070 enumerate(&y(r#""AGGREGATION_TEMPORALITY_DELTA""#), &TEMPORALITY, "t").unwrap(),
1071 1
1072 );
1073 assert_eq!(enumerate(&y("2"), &TEMPORALITY, "t").unwrap(), 2);
1074 assert_eq!(enumerate(&y(r#""2""#), &TEMPORALITY, "t").unwrap(), 2);
1075 assert_eq!(enumerate(&none, &TEMPORALITY, "t").unwrap(), 0);
1076 assert!(
1077 enumerate(&y(r#""NOPE""#), &TEMPORALITY, "t")
1078 .unwrap_err()
1079 .contains("not a known")
1080 );
1081
1082 use mira_proto::common::v1::any_value::Value as V;
1088 for (text, want) in [
1089 (r#"{"stringValue":"a"}"#, "string"),
1090 (r#"{"bool_value":true}"#, "bool"),
1091 (r#"{"intValue":"2"}"#, "int"),
1092 (r#"{"doubleValue":1.5}"#, "double"),
1093 (
1094 r#"{"arrayValue":{"values":[{"stringValue":"a"},{"intValue":"2"},{}]}}"#,
1095 "array",
1096 ),
1097 (
1098 r#"{"kvlist_value":{"values":[{"key":"k","value":{"bool_value":true}}]}}"#,
1099 "kvlist",
1100 ),
1101 (r#"{"bytesValue":"aGVsbG8="}"#, "bytes"),
1102 (r#"{}"#, "empty"),
1103 ] {
1104 let got = match any_value(&y(text)).unwrap().unwrap().value {
1105 Some(V::StringValue(s)) => {
1106 assert_eq!(s, "a");
1107 "string"
1108 }
1109 Some(V::BoolValue(b)) => {
1110 assert!(b);
1111 "bool"
1112 }
1113 Some(V::IntValue(i)) => {
1114 assert_eq!(i, 2);
1115 "int"
1116 }
1117 Some(V::DoubleValue(d)) => {
1118 assert_eq!(d, 1.5);
1119 "double"
1120 }
1121 Some(V::ArrayValue(a)) => {
1122 assert_eq!(a.values.len(), 3);
1123 assert!(a.values[2].value.is_none(), "`{{}}` is a valid AnyValue");
1124 "array"
1125 }
1126 Some(V::KvlistValue(l)) => {
1127 assert_eq!(l.values[0].key, "k");
1128 "kvlist"
1129 }
1130 Some(V::BytesValue(b)) => {
1131 assert_eq!(&b[..], b"hello");
1132 "bytes"
1133 }
1134 None => "empty",
1135 };
1136 assert_eq!(got, want, "{text}");
1137 }
1138 assert!(any_value(&none).unwrap().is_none());
1139 }
1140
1141 #[test]
1151 fn a_bad_scalar_is_refused_by_the_name_of_the_field_that_carried_it() {
1152 let log =
1155 |r: &str| format!(r#"{{"resourceLogs":[{{"scopeLogs":[{{"logRecords":[{r}]}}]}}]}}"#);
1156 let sp = |s: &str| format!(r#"{{"resourceSpans":[{{"scopeSpans":[{{"spans":[{s}]}}]}}]}}"#);
1157 let me = |m: &str| {
1158 format!(r#"{{"resourceMetrics":[{{"scopeMetrics":[{{"metrics":[{m}]}}]}}]}}"#)
1159 };
1160 let point = |kind: &str, p: &str| {
1161 me(&format!(
1162 r#"{{"name":"m","{kind}":{{"dataPoints":[{p}]}}}}"#
1163 ))
1164 };
1165
1166 for (body, want) in [
1167 (
1170 r#"{"resourceLogs":[{"resource":{"droppedAttributesCount":4294967296}}]}"#.to_owned(),
1171 "droppedAttributesCount: 4294967296 does not fit",
1172 ),
1173 (
1174 r#"{"resourceLogs":[{"scopeLogs":[{"scope":{"droppedAttributesCount":4294967297}}]}]}"#
1175 .to_owned(),
1176 "droppedAttributesCount: 4294967297 does not fit",
1177 ),
1178 (
1179 log(r#"{"observedTimeUnixNano":"whenever"}"#),
1180 r#"observedTimeUnixNano: "whenever" is not an unsigned integer"#,
1181 ),
1182 (
1183 log(r#"{"severityNumber":"SEVERITY_NUMBER_LOUD"}"#),
1184 r#"severityNumber: "SEVERITY_NUMBER_LOUD" is not a known value"#,
1185 ),
1186 ] {
1187 let e = logs(&doc(&body)).unwrap_err();
1188 assert!(e.contains(want), "{body}\n got: {e}");
1189 }
1190
1191 for (body, want) in [
1192 (
1193 sp(r#"{"startTimeUnixNano":"dawn"}"#),
1194 r#"startTimeUnixNano: "dawn" is not an unsigned integer"#,
1195 ),
1196 (
1197 sp(r#"{"endTimeUnixNano":"dusk"}"#),
1198 r#"endTimeUnixNano: "dusk" is not an unsigned integer"#,
1199 ),
1200 (
1201 sp(r#"{"droppedAttributesCount":4294967298}"#),
1202 "droppedAttributesCount: 4294967298 does not fit",
1203 ),
1204 (
1205 sp(r#"{"events":[{"droppedAttributesCount":4294967299}]}"#),
1206 "droppedAttributesCount: 4294967299 does not fit",
1207 ),
1208 (
1209 sp(r#"{"links":[{"droppedAttributesCount":4294967300}]}"#),
1210 "droppedAttributesCount: 4294967300 does not fit",
1211 ),
1212 ] {
1213 let e = traces(&doc(&body)).unwrap_err();
1214 assert!(e.contains(want), "{body}\n got: {e}");
1215 }
1216
1217 for (body, want) in [
1221 (
1222 point("gauge", r#"{"asDouble":"about 1.5"}"#),
1223 r#"dataPoint: "about 1.5" is not a number"#,
1224 ),
1225 (
1226 point("gauge", r#"{"asInt":"about 9"}"#),
1227 r#"dataPoint: "about 9" is not an integer"#,
1228 ),
1229 (
1230 point("gauge", r#"{"startTimeUnixNano":"gauge-dawn"}"#),
1231 r#"startTimeUnixNano: "gauge-dawn" is not an unsigned integer"#,
1232 ),
1233 (
1234 point("histogram", r#"{"startTimeUnixNano":"hist-dawn"}"#),
1235 r#"startTimeUnixNano: "hist-dawn" is not an unsigned integer"#,
1236 ),
1237 (
1238 point("histogram", r#"{"explicitBounds":["about 2.5"]}"#),
1239 r#"explicitBounds: "about 2.5" is not a number"#,
1240 ),
1241 (
1242 point(
1243 "exponentialHistogram",
1244 r#"{"startTimeUnixNano":"exp-dawn"}"#,
1245 ),
1246 r#"startTimeUnixNano: "exp-dawn" is not an unsigned integer"#,
1247 ),
1248 (
1249 point("summary", r#"{"startTimeUnixNano":"sum-dawn"}"#),
1250 r#"startTimeUnixNano: "sum-dawn" is not an unsigned integer"#,
1251 ),
1252 ] {
1253 let e = metrics(&doc(&body)).unwrap_err();
1254 assert!(e.contains(want), "{body}\n got: {e}");
1255 }
1256 }
1257}