mira_core/series.rs
1//! The metrics read path: match metrics, gather points, group into series.
2//!
3//! A different shape from [`crate::query::search`], and deliberately a separate
4//! function rather than a mode of it. A log search filters one root table and
5//! returns rows newest-first; a metrics query filters a *descriptor* table,
6//! gathers points from four point tables that share one id space, and then
7//! regroups them by a key that has to be stable across blocks. Folding both into
8//! one code path would mean inventing the abstraction that both are special
9//! cases of, which is a planner.
10//!
11//! The grouping key is the hard part and the reason this is not trivial. Ids are
12//! rebased per block — that is what makes the attribute joins array stores — so
13//! nothing block-local can identify a series across two blocks. The key is built
14//! from values instead: the metric's name, unit and kind, plus the merged
15//! attribute map from all four levels. That is one string built per point, which
16//! is the honest cost of a layout optimized for writing and pruning rather than
17//! for grouping.
18//!
19//! ## What V0 returns
20//!
21//! Gauges and sums come back as their own value. Histograms, exponential
22//! histograms and summaries come back as two derived series each, `<name>.count`
23//! and `<name>.sum` — the same convention Prometheus uses, and the same two
24//! numbers that answer "how often" and "how much". Bucket and quantile maths is
25//! not here: it is a heatmap feature, it needs the UI to exist first, and
26//! shipping it wrong would be worse than shipping it later. Everything needed
27//! for it is on disk already — `bucket_counts`, `bounds_id`, `scale`, `quantile`
28//! — so this is a read-path gap, not a storage one.
29//!
30//! Blocking: mmaps and page-faults, same as `search`. Callers on an async
31//! runtime must use `spawn_blocking`.
32
33use std::collections::{BTreeMap, BTreeSet, HashMap};
34use std::path::Path;
35use std::sync::Arc;
36
37use arrow_array::cast::AsArray;
38use arrow_array::types::{
39 Float64Type, Int64Type, TimestampNanosecondType, UInt8Type, UInt16Type, UInt32Type, UInt64Type,
40};
41use arrow_array::{Array, RecordBatch};
42
43use crate::block::{self, Src};
44use crate::error::Result;
45use crate::json::Json;
46use crate::query::{Attrs, Results, Stats, Term, attr_key, attr_parents, dict_index, emit_attr};
47use crate::schema::MetricKind;
48use crate::signal::Open;
49
50/// A metrics query.
51///
52/// Note what is missing: no `step`, no aggregation, no rate. Downsampling in the
53/// engine would need a fill policy, an alignment rule and a choice of aggregator
54/// per metric kind, all of which are presentation decisions the caller is better
55/// placed to make and all of which are wrong for somebody. Raw points, bounded
56/// by `max_points`, and the chart decides.
57#[derive(Debug, Clone)]
58pub struct SeriesQuery {
59 /// Exact metric name. `None` matches every metric, which is what the name
60 /// listing uses and what an exploratory query does before it knows better.
61 pub name: Option<String>,
62 /// Inclusive nanosecond bounds.
63 pub from: i64,
64 pub to: i64,
65 /// Attribute filters. A term is satisfied if it matches at *any* level —
66 /// resource, scope, metric or data point — because a caller filtering on
67 /// `service.name` should not have to know which of the four carries it.
68 pub terms: Vec<Term>,
69 pub max_series: usize,
70 /// Per series, not in total: truncating one busy series must not silently
71 /// empty the others charted beside it.
72 pub max_points: usize,
73}
74
75/// One point. Integers stay integers: an OTLP `as_int` is `sfixed64`, and a
76/// counter past 2^53 pushed through an f64 loses its low bits — which is exactly
77/// the moment a counter is interesting.
78#[derive(Debug, Clone, Copy)]
79enum Pt {
80 Int(i64),
81 Double(f64),
82}
83
84struct Series {
85 /// Rendered `"name":...,"unit":...,` etc. Built once when the series is
86 /// first seen; identical for every point in it by construction.
87 desc: String,
88 /// Rendered `{...}` attribute object.
89 attrs: String,
90 points: Vec<(i64, Pt)>,
91 /// Points dropped by `max_points`, reported rather than hidden. A chart
92 /// missing its spike because the engine quietly truncated is the failure
93 /// mode this exists to prevent.
94 dropped: usize,
95 /// The trace ids behind the points: `(time, rendered object)`, rendered
96 /// here because the block they were read from is unmapped before the
97 /// response is written.
98 exemplars: Vec<(i64, String)>,
99}
100
101impl Series {
102 /// Discard all but the newest `max` points.
103 ///
104 /// Which points a cap keeps is a correctness question, not a tuning one.
105 /// Points arrive in block-scan order, so refusing them once the buffer is
106 /// full kept whichever ones the directory listing happened to reach first
107 /// — an arbitrary subset that the sort at render time then dressed up as a
108 /// contiguous series. Newest-wins is the same rule `limit` already uses on
109 /// the record side, and it is a window a reader can reason about.
110 ///
111 /// Called at twice `max`, so every call throws away half of what it sorts
112 /// and the amortised cost is constant per point. A bounded heap would hold
113 /// exactly `max` at all times and pay `log max` on every point instead;
114 /// at these sizes the occasional sort is cheaper and much less code.
115 ///
116 /// ponytail: truncation, not downsampling. A chart that needs the whole
117 /// range at a lower resolution needs an aggregation window — until that
118 /// exists the response says `dropped_points` rather than pretending.
119 fn compact(&mut self, max: usize) {
120 // Descending, so `truncate` keeps the newest.
121 self.points.sort_unstable_by_key(|p| std::cmp::Reverse(p.0));
122 self.dropped += self.points.len() - max;
123 self.points.truncate(max);
124 }
125}
126
127/// Exemplars kept per series.
128///
129/// An exemplar is one sampled measurement per collection interval per bucket,
130/// so a well-behaved exporter sends few — but nothing in OTLP bounds it, and an
131/// unbounded array here would make one misconfigured SDK able to inflate every
132/// chart response. ponytail: a flat cap, not a reservoir sample; the first ones
133/// in a window are as good a sample as any until someone proves otherwise.
134const MAX_EXEMPLARS: usize = 64;
135
136/// One attribute as `(key, rendered JSON value)`. Rendered once and carried as
137/// a string because the same value is written into both the grouping key and
138/// the response.
139type Attr = (String, String);
140
141/// A descriptor row's cached contribution: its rendered JSON members, and the
142/// attributes every point beneath it inherits.
143type Prefix = (String, Vec<Attr>);
144
145/// The four point tables, and how many name suffixes each contributes.
146///
147/// One table per OTLP point type rather than one wide table: measured, the split
148/// layout is 2.34x smaller (73.0 vs 170.6 bytes/point), because a wide table
149/// pays for every column of every type on every row.
150const DP_TABLES: [&str; 4] = ["number_dp", "hist_dp", "exp_hist_dp", "summary_dp"];
151
152/// The suffixes a histogram's or summary's name is derived with. Named rather
153/// than written twice, because the name filter has to accept back exactly the
154/// names the renderer hands out and two literal lists would drift apart.
155const COUNT: &str = ".count";
156const SUM: &str = ".sum";
157const DERIVED: [&str; 2] = [COUNT, SUM];
158
159pub fn series(root: &Path, q: &SeriesQuery) -> Result<Results> {
160 series_open(root, q, &[])
161}
162
163/// As [`series`], but also reads the open block's snapshot — see
164/// [`crate::query::search_open`], which this mirrors exactly.
165pub fn series_open(root: &Path, q: &SeriesQuery, open: &[Arc<Open>]) -> Result<Results> {
166 let disk = block::scan(root, "metrics")?;
167 let refs = block::sources(&disk, open);
168 let mut stats = Stats {
169 blocks_total: refs.len(),
170 ..Default::default()
171 };
172 let mut out: BTreeMap<String, Series> = BTreeMap::new();
173 let mut dropped: BTreeSet<String> = BTreeSet::new();
174
175 for bref in &refs {
176 // The directory name is the whole index; a block outside the window is
177 // never opened.
178 if bref.max_ts < q.from || bref.min_ts > q.to {
179 continue;
180 }
181 stats.blocks_scanned += 1;
182 collect_block(bref, q, &mut out, &mut dropped, &mut stats)?;
183 }
184 stats.dropped_series = dropped.len();
185
186 // The last compaction of each series, and the only one for a series that
187 // never reached the trigger. Sorting ascending here as well means the
188 // render below reads `points` directly instead of cloning it.
189 for s in out.values_mut() {
190 if s.points.len() > q.max_points {
191 s.compact(q.max_points);
192 }
193 s.points.sort_unstable_by_key(|p| p.0);
194 }
195
196 // A `BTreeMap`, so this is already in key order: two identical queries
197 // produce byte-identical responses, which an ETag, a diff and a cache all
198 // depend on and which a `HashMap` iteration order does not provide. It is
199 // also already at most `max_series` long — see [`bound`].
200 let mut j = Json::new();
201 j.arr(|j| {
202 for s in out.values() {
203 let pts = &s.points;
204 j.obj(|j| {
205 j.raw(&s.desc);
206 j.key("attributes");
207 j.raw(&s.attrs);
208 if s.dropped > 0 {
209 j.key("dropped_points");
210 j.u64(s.dropped as u64);
211 }
212 j.key("points");
213 j.arr(|j| {
214 for &(ts, v) in pts {
215 j.arr(|j| {
216 // Both 64-bit halves go out as strings, for the
217 // reason [`Json::i64_str`] gives: a nanosecond
218 // timestamp is ~1.7e18 and an OTLP `as_int` is an
219 // `sfixed64`, and a counter past 2^53 read back
220 // through a double loses exactly the low bits that
221 // made it worth charting. A double stays a number
222 // — it was never anything else.
223 j.i64_str(ts);
224 match v {
225 Pt::Int(i) => j.i64_str(i),
226 Pt::Double(d) => j.f64(d),
227 }
228 });
229 }
230 });
231 // The answer to "which trace made this spike". Omitted rather
232 // than emitted empty, because most series have none and this is
233 // the response a chart polls.
234 if !s.exemplars.is_empty() {
235 let mut ex = s.exemplars.clone();
236 ex.sort_unstable_by_key(|e| e.0);
237 j.key("exemplars");
238 j.arr(|j| {
239 for (_, rendered) in &ex {
240 j.raw(rendered);
241 }
242 });
243 }
244 });
245 }
246 });
247 Ok(Results {
248 json: j.into_string(),
249 stats,
250 // Metrics are bounded by `max_series` and `max_points`, which cap what
251 // a chart can render rather than cut a list short. Nothing to page.
252 next: None,
253 })
254}
255
256/// Every metric name present in the window, with its unit and kind.
257///
258/// This is what a UI puts in a dropdown and what an agent reads before writing
259/// its first query, so it is a first-class endpoint rather than something to
260/// derive from an unfiltered `series` call — the descriptor table is tens of
261/// rows per block, while the points it describes are hundreds of thousands.
262pub fn names(root: &Path, from: i64, to: i64) -> Result<Results> {
263 names_open(root, from, to, &[])
264}
265
266/// As [`names`], but also reads the open block's snapshot. A metric name that
267/// has only ever been written to the open block is exactly the one a dropdown
268/// must not omit — it is the new one.
269pub fn names_open(root: &Path, from: i64, to: i64, open: &[Arc<Open>]) -> Result<Results> {
270 let disk = block::scan(root, "metrics")?;
271 let refs = block::sources(&disk, open);
272 let mut stats = Stats {
273 blocks_total: refs.len(),
274 ..Default::default()
275 };
276 let mut seen: HashMap<String, (String, u8)> = HashMap::new();
277
278 for bref in &refs {
279 if bref.max_ts < from || bref.min_ts > to {
280 continue;
281 }
282 stats.blocks_scanned += 1;
283 let Some(m) = load(bref, "metrics")? else {
284 continue;
285 };
286 stats.rows_scanned += m.num_rows();
287 for r in 0..m.num_rows() {
288 let name = dict_str(&m, "name", r).unwrap_or("").to_owned();
289 let unit = dict_str(&m, "unit", r).unwrap_or("").to_owned();
290 let kind = u8_col(&m, "kind", r);
291 seen.entry(name).or_insert((unit, kind));
292 }
293 }
294 stats.rows_matched = seen.len();
295
296 let mut names: Vec<&String> = seen.keys().collect();
297 names.sort_unstable();
298 let mut j = Json::new();
299 j.arr(|j| {
300 for n in names {
301 let (unit, kind) = &seen[n];
302 j.obj(|j| {
303 j.key("name");
304 j.str(n);
305 j.key("unit");
306 j.str(unit);
307 j.key("kind");
308 j.str(kind_name(*kind));
309 });
310 }
311 });
312 Ok(Results {
313 json: j.into_string(),
314 stats,
315 // Metrics are bounded by `max_series` and `max_points`, which cap what
316 // a chart can render rather than cut a list short. Nothing to page.
317 next: None,
318 })
319}
320
321fn kind_name(k: u8) -> &'static str {
322 match k {
323 x if x == MetricKind::Gauge as u8 => "gauge",
324 x if x == MetricKind::Sum as u8 => "sum",
325 x if x == MetricKind::Histogram as u8 => "histogram",
326 x if x == MetricKind::ExponentialHistogram as u8 => "exponential_histogram",
327 x if x == MetricKind::Summary as u8 => "summary",
328 _ => "unset",
329 }
330}
331
332fn load(bref: &Src, name: &str) -> Result<Option<RecordBatch>> {
333 bref.load(name)
334}
335
336/// Child rows grouped by `parent_id`, indexed by it.
337///
338/// Ids are rebased dense from zero per block, so the parent id *is* the slot —
339/// no hash map, and the whole thing is one pass over a `u32` buffer.
340pub(crate) fn index_by_parent(b: &RecordBatch) -> Vec<Vec<u32>> {
341 let Some(col) = b.column_by_name("parent_id") else {
342 return Vec::new();
343 };
344 let parents = col.as_primitive::<UInt32Type>().values();
345 let mut out: Vec<Vec<u32>> =
346 vec![Vec::new(); parents.iter().copied().max().unwrap_or(0) as usize + 1];
347 for (r, &p) in parents.iter().enumerate() {
348 out[p as usize].push(r as u32);
349 }
350 out
351}
352
353fn exemplar_time(b: &RecordBatch, row: u32) -> i64 {
354 b.column_by_name("time_unix_nano").map_or(0, |c| {
355 c.as_primitive::<TimestampNanosecondType>()
356 .value(row as usize)
357 })
358}
359
360fn dict_str<'a>(b: &'a RecordBatch, col: &str, row: usize) -> Option<&'a str> {
361 let c = b.column_by_name(col)?;
362 let d = c.as_dictionary::<UInt16Type>();
363 if d.is_null(row) {
364 return None;
365 }
366 Some(
367 d.values()
368 .as_string::<i32>()
369 .value(d.keys().value(row) as usize),
370 )
371}
372
373fn u8_col(b: &RecordBatch, col: &str, row: usize) -> u8 {
374 b.column_by_name(col)
375 .map_or(0, |c| c.as_primitive::<UInt8Type>().value(row))
376}
377
378/// Hold `out` to `max_series` entries by evicting its largest key.
379///
380/// The result is the same set `sort(keys).truncate(max_series)` produced, and
381/// the argument is short enough to check: the answer is the `max_series`
382/// smallest keys, and a key that belongs in it can never be the largest key of
383/// a map that is already one over — everything else in that map is smaller, so
384/// there are already `max_series` keys ahead of it. Evicting the maximum
385/// therefore only ever discards a key the sort would have truncated. It also
386/// cannot come back: the evicted key is greater than every key in the map, and
387/// the map's maximum only falls, so the same key arriving from a later block is
388/// evicted again on sight.
389///
390/// The difference is *when*, and that is the whole point. The old map grew one
391/// entry per distinct (name, unit, kind, temporality, monotonic, attributes)
392/// tuple until the query finished, each carrying a rendered descriptor, a
393/// rendered attribute object and a key concatenating both — about a kilobyte.
394/// `max_points` bounded the points inside a series and nothing bounded the
395/// series, so a `query_metric` with no `name` over a store with a request id in
396/// a data-point attribute allocated until the process died.
397///
398/// The evicted key is kept, without its payload, so the response can say how
399/// many series it is not showing. ponytail: that record is itself capped at
400/// `max_series` keys, so `dropped_series` saturates rather than counting an
401/// unbounded number of them exactly — remembering every key to count it is the
402/// allocation this function exists to prevent. `dropped_series == max_series`
403/// reads as "at least". Counting evictions instead of distinct keys was the
404/// other option and it is worse: a series above the cap is re-inserted and
405/// re-evicted once per point, so a store with 100 series and a cap of 64 would
406/// report tens of thousands dropped.
407fn bound(out: &mut BTreeMap<String, Series>, dropped: &mut BTreeSet<String>, max: usize) {
408 while out.len() > max {
409 let Some((k, _)) = out.pop_last() else { break };
410 if dropped.len() < max {
411 dropped.insert(k);
412 }
413 }
414}
415
416fn collect_block(
417 bref: &Src,
418 q: &SeriesQuery,
419 out: &mut BTreeMap<String, Series>,
420 dropped: &mut BTreeSet<String>,
421 stats: &mut Stats,
422) -> Result<()> {
423 // A missing descriptor table means retention is unlinking this block under
424 // us. Normal, not an error.
425 let Some(metrics) = load(bref, "metrics")? else {
426 return Ok(());
427 };
428 let n_metrics = metrics.num_rows();
429
430 // Name filter: resolve the string against the dictionary once, then compare
431 // u16 codes. A block whose dictionary lacks the name has no rows to check.
432 //
433 // Above the six loads below, not below them: each is an `mmap` and a CRC of
434 // the whole table body, so a block that does not hold the requested metric
435 // should cost one of them and not six.
436 let mut wanted = vec![true; n_metrics];
437 // Set when the requested name was a derived one, so only that half of the
438 // histogram is emitted rather than both.
439 let mut only_suffix = "";
440 if let Some(want) = &q.name {
441 let d = metrics
442 .column_by_name("name")
443 .map(|c| c.as_dictionary::<UInt16Type>());
444 let Some(d) = d else { return Ok(()) };
445 let names = d.values().as_string::<i32>();
446 // The descriptor dictionary holds `http.server.duration`; this function
447 // hands back series called `http.server.duration.count`. A caller
448 // pasting a name off its own previous answer — a chart legend, an agent
449 // reading the result it just got — must not get a silent empty series
450 // list, so a miss retries against the base name. A metric genuinely
451 // called `foo.count` matches on the first try and keeps winning.
452 let mut code = dict_index(names, want);
453 if code.is_none() {
454 for s in DERIVED {
455 if let Some(base) = want.strip_suffix(s) {
456 code = dict_index(names, base);
457 only_suffix = s;
458 break;
459 }
460 }
461 }
462 let Some(code) = code else { return Ok(()) };
463 let codes = d.keys().values();
464 for (r, w) in wanted.iter_mut().enumerate() {
465 *w = codes[r] == code;
466 }
467 }
468
469 // Wrapped rather than raw, because `own_attrs` below runs once per matched
470 // point and `Attrs` turns each of those from a scan of the whole table into
471 // two binary searches.
472 let metric_attrs = load(bref, "metric_attrs")?.map(Attrs::new);
473 let dp_attrs = load(bref, "dp_attrs")?.map(Attrs::new);
474 // Data point ids are one dense space across all four point tables, so one
475 // list indexed by id serves every table below — the same property that lets
476 // `dp_attrs` carry no discriminant.
477 let exemplars = load(bref, "exemplars")?;
478 let by_point = exemplars.as_ref().map(index_by_parent).unwrap_or_default();
479 let resource_attrs = load(bref, "resource_attrs")?.map(Attrs::new);
480 let scope_attrs = load(bref, "scope_attrs")?.map(Attrs::new);
481
482 // For each term, which metric rows satisfy it *above* the point level. A
483 // point still qualifies if its own attributes satisfy the term, so this is
484 // one half of an OR evaluated per point below.
485 let above: Vec<Vec<bool>> = q
486 .terms
487 .iter()
488 .map(|t| {
489 let key = match &t.target {
490 crate::query::Target::Attr(k) | crate::query::Target::Field(k) => k,
491 };
492 let mut hit = vec![false; n_metrics];
493 // Metric level: parent_id is the descriptor row number.
494 if let Some(a) = &metric_attrs {
495 for pid in attr_parents(&a.rows, key, t.op, &t.value) {
496 if let Some(s) = hit.get_mut(pid as usize) {
497 *s = true;
498 }
499 }
500 }
501 // Resource and scope level: parent_id is an entity id, and the
502 // descriptor row carries the foreign key.
503 for (table, fk) in [(&resource_attrs, "resource_id"), (&scope_attrs, "scope_id")] {
504 let (Some(a), Some(col)) = (table, metrics.column_by_name(fk)) else {
505 continue;
506 };
507 let ids = attr_parents(&a.rows, key, t.op, &t.value);
508 let Some(&top) = ids.iter().max() else {
509 continue;
510 };
511 let mut want = vec![false; top as usize + 1];
512 for id in ids {
513 want[id as usize] = true;
514 }
515 for (r, &id) in col.as_primitive::<UInt16Type>().values().iter().enumerate() {
516 if want.get(id as usize).copied().unwrap_or(false) {
517 hit[r] = true;
518 }
519 }
520 }
521 hit
522 })
523 .collect();
524
525 // Same, at the point level. Data point ids are one shared space across all
526 // four point tables — that is why `dp_attrs` needs no table discriminant —
527 // so one bitmap per term serves every table below.
528 let dp_hit: Vec<Vec<bool>> = q
529 .terms
530 .iter()
531 .map(|t| {
532 let key = match &t.target {
533 crate::query::Target::Attr(k) | crate::query::Target::Field(k) => k,
534 };
535 let mut hit = Vec::new();
536 if let Some(a) = &dp_attrs {
537 for pid in attr_parents(&a.rows, key, t.op, &t.value) {
538 if hit.len() <= pid as usize {
539 hit.resize(pid as usize + 1, false);
540 }
541 hit[pid as usize] = true;
542 }
543 }
544 hit
545 })
546 .collect();
547
548 // Attributes above the point level, rendered once per descriptor row rather
549 // than once per point. A metric with 10k points would otherwise re-render
550 // its resource attributes 10k times.
551 let mut prefix: Vec<Option<Prefix>> = vec![None; n_metrics];
552
553 for table in DP_TABLES {
554 let Some(dp) = load(bref, table)? else {
555 continue;
556 };
557 let n = dp.num_rows();
558 stats.rows_scanned += n;
559 let (Some(time), Some(mid), Some(did)) = (
560 dp.column_by_name("time_unix_nano")
561 .map(|c| &**c.as_primitive::<TimestampNanosecondType>().values()),
562 dp.column_by_name("metric_id")
563 .map(|c| &**c.as_primitive::<UInt32Type>().values()),
564 dp.column_by_name("id")
565 .map(|c| &**c.as_primitive::<UInt32Type>().values()),
566 ) else {
567 continue;
568 };
569 let vals = Values::for_table(table);
570
571 for r in 0..n {
572 let m = mid[r] as usize;
573 if !(q.from..=q.to).contains(&time[r]) || !wanted.get(m).copied().unwrap_or(false) {
574 continue;
575 }
576 let d = did[r] as usize;
577 if !(0..q.terms.len()).all(|t| {
578 above[t].get(m).copied().unwrap_or(false)
579 || dp_hit[t].get(d).copied().unwrap_or(false)
580 }) {
581 continue;
582 }
583 stats.rows_matched += 1;
584
585 let (desc_prefix, upper) = prefix[m].get_or_insert_with(|| {
586 (
587 describe(&metrics, m),
588 upper_attrs(&metrics, m, &metric_attrs, &resource_attrs, &scope_attrs),
589 )
590 });
591 // The point's own attributes, merged over the inherited ones.
592 let own = own_attrs(&dp_attrs, d as u32);
593 let attrs = merge(upper, &own);
594
595 for (suffix, v) in vals.at(&dp, r) {
596 if !only_suffix.is_empty() && suffix != only_suffix {
597 continue;
598 }
599 let key = format!("{desc_prefix}\u{1}{suffix}\u{1}{attrs}");
600 let s = out.entry(key).or_insert_with(|| Series {
601 desc: with_suffix(desc_prefix, suffix),
602 attrs: attrs.clone(),
603 points: Vec::new(),
604 dropped: 0,
605 exemplars: Vec::new(),
606 });
607 s.points.push((time[r], v));
608 if s.points.len() >= 2 * q.max_points {
609 s.compact(q.max_points);
610 }
611 // A histogram yields two derived series from one point, and its
612 // exemplars belong to both: whichever of `.count` and `.sum` is
613 // charted, the spike in it points at the same traces.
614 if let (Some(ex), Some(rows)) = (&exemplars, by_point.get(d)) {
615 for &er in rows {
616 if s.exemplars.len() >= MAX_EXEMPLARS {
617 break;
618 }
619 let mut j = Json::new();
620 j.obj(|j| crate::query::emit_fields(j, ex, er));
621 s.exemplars.push((exemplar_time(ex, er), j.into_string()));
622 }
623 }
624 // Applied per point, not once at the end: the whole reason
625 // `max_series` exists is that the map between here and the end
626 // is what runs the process out of memory.
627 bound(out, dropped, q.max_series);
628 }
629 }
630 }
631 Ok(())
632}
633
634/// The descriptor's identity as rendered JSON members, without the trailing
635/// comma. Doubles as the stable half of the series key.
636fn describe(metrics: &RecordBatch, row: usize) -> String {
637 let mut j = Json::new();
638 j.key("name");
639 j.str(dict_str(metrics, "name", row).unwrap_or(""));
640 j.key("unit");
641 j.str(dict_str(metrics, "unit", row).unwrap_or(""));
642 j.key("kind");
643 j.str(kind_name(u8_col(metrics, "kind", row)));
644 j.key("temporality");
645 j.u64(u8_col(metrics, "temporality", row) as u64);
646 j.key("monotonic");
647 j.bool(
648 metrics
649 .column_by_name("is_monotonic")
650 .is_some_and(|c| c.as_boolean().value(row)),
651 );
652 j.into_string()
653}
654
655/// `describe` with the derived-series suffix folded into the name, so a
656/// `.count` series reports the name a caller can query it back by.
657///
658/// That is only true because `collect_block` strips a [`DERIVED`] suffix when
659/// the descriptor dictionary does not hold the requested name. Emitting a name
660/// here that the filter there does not accept is the same bug as returning a
661/// cursor nobody can page with.
662fn with_suffix(desc: &str, suffix: &str) -> String {
663 if suffix.is_empty() {
664 return desc.to_owned();
665 }
666 // `describe` always emits `"name":"..."` first, so the closing quote of the
667 // name is the second unescaped quote after the colon. Splitting on the known
668 // prefix is cheaper and less fragile than re-rendering.
669 match desc.find("\",\"unit\"") {
670 Some(i) => format!("{}{suffix}{}", &desc[..i], &desc[i..]),
671 None => desc.to_owned(),
672 }
673}
674
675/// Attributes from the metric, resource and scope levels, sorted and deduped
676/// with the most specific level winning.
677fn upper_attrs(
678 metrics: &RecordBatch,
679 row: usize,
680 metric_attrs: &Option<Attrs>,
681 resource_attrs: &Option<Attrs>,
682 scope_attrs: &Option<Attrs>,
683) -> Vec<Attr> {
684 let fk = |name: &str| {
685 metrics
686 .column_by_name(name)
687 .map(|c| c.as_primitive::<UInt16Type>().value(row) as u32)
688 };
689 let mut v = Vec::new();
690 // Least specific first: `merge_into` keeps the last write per key.
691 for (table, parent) in [
692 (resource_attrs, fk("resource_id")),
693 (scope_attrs, fk("scope_id")),
694 (metric_attrs, Some(row as u32)),
695 ] {
696 let (Some(a), Some(p)) = (table, parent) else {
697 continue;
698 };
699 collect_attrs(a, p, &mut v);
700 }
701 dedup_last(&mut v);
702 v
703}
704
705fn own_attrs(dp_attrs: &Option<Attrs>, dp_id: u32) -> Vec<Attr> {
706 let mut v = Vec::new();
707 if let Some(a) = dp_attrs {
708 collect_attrs(a, dp_id, &mut v);
709 }
710 dedup_last(&mut v);
711 v
712}
713
714/// The rows one parent owns, rendered.
715///
716/// This runs once per matched point, so the search matters: `dp_attrs` on a
717/// block holding 7,752 points is thousands of rows, and scanning all of them
718/// per point is the quadratic shape [`Attrs`] exists to remove. Empty for a
719/// table whose parent column is not a `u32`, matching the read path — no schema
720/// in this tree writes one, and a block someone else wrote should not panic.
721fn collect_attrs(a: &Attrs, parent: u32, out: &mut Vec<Attr>) {
722 let parents = Attrs::parents(&a.rows).unwrap_or_default();
723 for r in a.run(parents, parent).filter(|&r| parents[r] == parent) {
724 let mut j = Json::new();
725 emit_attr(&mut j, &a.rows, r);
726 out.push((attr_key(&a.rows, r).to_owned(), j.into_string()));
727 }
728}
729
730fn dedup_last(v: &mut Vec<Attr>) {
731 // Stable, so entries pushed later — from the more specific level — sort
732 // after their earlier namesakes and win the dedup.
733 v.sort_by(|a, b| a.0.cmp(&b.0));
734 v.dedup_by(|a, b| {
735 if a.0 == b.0 {
736 std::mem::swap(a, b);
737 true
738 } else {
739 false
740 }
741 });
742}
743
744/// Render the union of two sorted attribute lists as a JSON object, `own`
745/// winning on collision. A merge, not a concatenation and a re-sort: `upper` is
746/// already sorted and shared by every point of the metric.
747fn merge(upper: &[Attr], own: &[Attr]) -> String {
748 let mut j = Json::new();
749 j.obj(|j| {
750 let (mut i, mut k) = (0, 0);
751 while i < upper.len() || k < own.len() {
752 let take_own = match (upper.get(i), own.get(k)) {
753 (Some(u), Some(o)) => {
754 if u.0 == o.0 {
755 i += 1;
756 }
757 o.0 <= u.0
758 }
759 (None, Some(_)) => true,
760 _ => false,
761 };
762 let (key, val) = if take_own {
763 k += 1;
764 &own[k - 1]
765 } else {
766 i += 1;
767 &upper[i - 1]
768 };
769 j.key(key);
770 j.raw(val);
771 }
772 });
773 j.into_string()
774}
775
776/// Which columns of a point table carry chartable values, resolved once per
777/// table instead of per row.
778enum Values {
779 /// `int` and `double`, exactly one set per row.
780 Number,
781 /// `count` and `sum`, emitted as `<name>.count` and `<name>.sum`.
782 CountSum,
783 None,
784}
785
786impl Values {
787 fn for_table(table: &str) -> Values {
788 match table {
789 "number_dp" => Values::Number,
790 "hist_dp" | "exp_hist_dp" | "summary_dp" => Values::CountSum,
791 _ => Values::None,
792 }
793 }
794
795 fn at(&self, dp: &RecordBatch, r: usize) -> Vec<(&'static str, Pt)> {
796 match self {
797 Values::Number => {
798 if let Some(c) = dp.column_by_name("int") {
799 let a = c.as_primitive::<Int64Type>();
800 if !a.is_null(r) {
801 return vec![("", Pt::Int(a.value(r)))];
802 }
803 }
804 if let Some(c) = dp.column_by_name("double") {
805 let a = c.as_primitive::<Float64Type>();
806 if !a.is_null(r) {
807 return vec![("", Pt::Double(a.value(r)))];
808 }
809 }
810 Vec::new()
811 }
812 Values::CountSum => {
813 let mut v = Vec::with_capacity(2);
814 if let Some(c) = dp.column_by_name("count") {
815 let a = c.as_primitive::<UInt64Type>();
816 if !a.is_null(r) {
817 v.push((COUNT, Pt::Int(a.value(r) as i64)));
818 }
819 }
820 if let Some(c) = dp.column_by_name("sum") {
821 let a = c.as_primitive::<Float64Type>();
822 if !a.is_null(r) {
823 v.push((SUM, Pt::Double(a.value(r))));
824 }
825 }
826 v
827 }
828 Values::None => Vec::new(),
829 }
830 }
831}
832
833#[cfg(test)]
834mod tests {
835 use super::*;
836 use arrow_array::{ArrayRef, Float64Array, Int64Array, UInt32Array, UInt64Array};
837 use mira_proto::collector::metrics::v1::ExportMetricsServiceRequest;
838 use mira_proto::common::v1::any_value::Value as AnyVal;
839 use mira_proto::common::v1::{AnyValue, InstrumentationScope, KeyValue};
840 use mira_proto::metrics::v1::metric::Data;
841 use mira_proto::metrics::v1::number_data_point::Value as NumValue;
842 use mira_proto::metrics::v1::{
843 Exemplar, ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Histogram,
844 HistogramDataPoint, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, Sum, Summary,
845 SummaryDataPoint, exemplar,
846 };
847 use mira_proto::resource::v1::Resource;
848
849 /// A named temporary directory, emptied first so a crashed run does not
850 /// hand the next one a half-written store.
851 fn tmp(name: &str) -> std::path::PathBuf {
852 let d = std::env::temp_dir().join(format!("mira-{name}-{}", std::process::id()));
853 let _ = std::fs::remove_dir_all(&d);
854 d
855 }
856
857 fn kv(k: &str, v: &str) -> KeyValue {
858 KeyValue {
859 key: k.into(),
860 value: Some(AnyValue {
861 value: Some(AnyVal::StringValue(v.into())),
862 }),
863 }
864 }
865
866 /// Publish `metrics` as one block, and hand back its directory so a test
867 /// can damage it the way retention or an older writer would.
868 fn publish(
869 dir: &std::path::Path,
870 seq: u64,
871 resource: Option<Resource>,
872 scope: Option<InstrumentationScope>,
873 metrics: Vec<Metric>,
874 ) -> std::path::PathBuf {
875 let req = ExportMetricsServiceRequest {
876 resource_metrics: vec![ResourceMetrics {
877 resource,
878 scope_metrics: vec![ScopeMetrics {
879 scope,
880 metrics,
881 ..Default::default()
882 }],
883 ..Default::default()
884 }],
885 };
886 let mut b = crate::metrics::MetricsBuilder::new();
887 b.append_request(&req).unwrap();
888 let sealed = b.finish().unwrap();
889 crate::block::publish(dir, "metrics", crate::block::node_id("a"), seq, 0, &sealed)
890 .unwrap()
891 .dir
892 }
893
894 /// A gauge of one integer point per timestamp, the simplest thing that
895 /// produces a series.
896 fn gauge(name: &str, points: &[(u64, i64)]) -> Metric {
897 Metric {
898 name: name.into(),
899 data: Some(Data::Gauge(Gauge {
900 data_points: points
901 .iter()
902 .map(|&(t, v)| NumberDataPoint {
903 time_unix_nano: t,
904 value: Some(NumValue::AsInt(v)),
905 ..Default::default()
906 })
907 .collect(),
908 })),
909 ..Default::default()
910 }
911 }
912
913 fn query(dir: &std::path::Path, q: &SeriesQuery) -> Results {
914 series(dir, q).unwrap()
915 }
916
917 /// The query every test below varies one field of.
918 fn wide() -> SeriesQuery {
919 SeriesQuery {
920 name: None,
921 from: 0,
922 to: i64::MAX,
923 terms: Vec::new(),
924 max_series: 100,
925 max_points: 100,
926 }
927 }
928
929 /// A histogram comes back as two series called `<name>.count` and
930 /// `<name>.sum`, and those are the names a chart legend shows and an agent
931 /// reads off its own previous answer. Matching only the base name against
932 /// the descriptor dictionary made that round trip return an empty list with
933 /// a 200, which reads as "the metric stopped reporting".
934 #[test]
935 fn a_derived_series_name_queries_back_to_the_series_it_names() {
936 let dir = std::env::temp_dir().join(format!("mira-derived-{}", std::process::id()));
937 let _ = std::fs::remove_dir_all(&dir);
938
939 let req = ExportMetricsServiceRequest {
940 resource_metrics: vec![ResourceMetrics {
941 scope_metrics: vec![ScopeMetrics {
942 metrics: vec![
943 Metric {
944 name: "http.server.duration".into(),
945 data: Some(Data::Histogram(Histogram {
946 data_points: vec![HistogramDataPoint {
947 time_unix_nano: 1_000,
948 count: 3,
949 sum: Some(1.5),
950 ..Default::default()
951 }],
952 ..Default::default()
953 })),
954 ..Default::default()
955 },
956 // A gauge whose name already ends in `.count`. The base
957 // name has to win the lookup, or this metric becomes
958 // unreachable the moment the stripping is added.
959 Metric {
960 name: "queue.depth.count".into(),
961 data: Some(Data::Gauge(Gauge {
962 data_points: vec![NumberDataPoint {
963 time_unix_nano: 1_000,
964 value: Some(NumValue::AsInt(42)),
965 ..Default::default()
966 }],
967 })),
968 ..Default::default()
969 },
970 ],
971 ..Default::default()
972 }],
973 ..Default::default()
974 }],
975 };
976
977 let mut b = crate::metrics::MetricsBuilder::new();
978 b.append_request(&req).unwrap();
979 let sealed = b.finish().unwrap();
980 crate::block::publish(&dir, "metrics", crate::block::node_id("a"), 0, 0, &sealed).unwrap();
981
982 let ask = |name: &str| {
983 series(
984 &dir,
985 &SeriesQuery {
986 name: Some(name.into()),
987 from: 0,
988 to: 10_000,
989 terms: Vec::new(),
990 max_series: 100,
991 max_points: 100,
992 },
993 )
994 .unwrap()
995 .json
996 };
997
998 // The base name still returns both halves.
999 let both = ask("http.server.duration");
1000 for half in [COUNT, SUM] {
1001 assert!(
1002 both.contains(&format!(r#""name":"http.server.duration{half}""#)),
1003 "{both}"
1004 );
1005 }
1006
1007 // Each derived name returns exactly the series it names, and only it.
1008 let c = ask("http.server.duration.count");
1009 assert!(c.contains(r#""name":"http.server.duration.count""#), "{c}");
1010 assert!(!c.contains(SUM), "{c}");
1011 // Both halves of a point are strings when the value is an integer: the
1012 // timestamp is nanoseconds and a `.count` is a `uint64`, so both are
1013 // 64-bit and both go out the OTLP/JSON way. `.sum` below is a double
1014 // and stays a bare number, which is what makes this pair worth having.
1015 assert!(c.contains(r#"["1000","3"]"#), "{c}");
1016 let s = ask("http.server.duration.sum");
1017 assert!(s.contains(r#""name":"http.server.duration.sum""#), "{s}");
1018 assert!(!s.contains(COUNT), "{s}");
1019 assert!(s.contains(r#"["1000",1.5]"#), "{s}");
1020
1021 // A metric that really is called `x.count` matches before the suffix is
1022 // stripped.
1023 let q = ask("queue.depth.count");
1024 assert!(q.contains(r#""name":"queue.depth.count""#), "{q}");
1025 assert!(q.contains(r#"["1000","42"]"#), "{q}");
1026
1027 // The retry widens the lookup; it does not make it match anything. A
1028 // derived name on a metric that has no derived series is still empty,
1029 // and so is a base name nothing carries.
1030 assert_eq!(ask("queue.depth.count.sum"), "[]");
1031 assert_eq!(ask("nope.count"), "[]");
1032
1033 let _ = std::fs::remove_dir_all(&dir);
1034 }
1035
1036 /// `max_series` bounds the map, not just the render.
1037 ///
1038 /// A high-cardinality attribute — a request id, a customer id, a pod name in
1039 /// a cluster that recycles them — makes one metric name expand into a series
1040 /// per distinct value. Accumulating them all and truncating at the end meant
1041 /// the cap the caller set had no effect on the memory the query took, and
1042 /// the store is the one thing here with no bound on how many values it
1043 /// holds.
1044 #[test]
1045 fn the_series_cap_bounds_the_map_and_says_what_it_refused() {
1046 let dir = std::env::temp_dir().join(format!("mira-cap-{}", std::process::id()));
1047 let _ = std::fs::remove_dir_all(&dir);
1048
1049 // Ten series of one metric, distinguished only by an attribute, and
1050 // named so that key order is the order a reader would guess.
1051 let req = ExportMetricsServiceRequest {
1052 resource_metrics: vec![ResourceMetrics {
1053 scope_metrics: vec![ScopeMetrics {
1054 metrics: vec![Metric {
1055 name: "rpc.duration".into(),
1056 data: Some(Data::Gauge(Gauge {
1057 data_points: (0..10)
1058 .map(|i| NumberDataPoint {
1059 time_unix_nano: 1_000,
1060 value: Some(NumValue::AsInt(i)),
1061 attributes: vec![mira_proto::common::v1::KeyValue {
1062 key: "peer".into(),
1063 value: Some(mira_proto::common::v1::AnyValue {
1064 value: Some(
1065 mira_proto::common::v1::any_value::Value::StringValue(
1066 format!("p{i}"),
1067 ),
1068 ),
1069 }),
1070 }],
1071 ..Default::default()
1072 })
1073 .collect(),
1074 })),
1075 ..Default::default()
1076 }],
1077 ..Default::default()
1078 }],
1079 ..Default::default()
1080 }],
1081 };
1082
1083 let mut b = crate::metrics::MetricsBuilder::new();
1084 b.append_request(&req).unwrap();
1085 let sealed = b.finish().unwrap();
1086 crate::block::publish(&dir, "metrics", crate::block::node_id("a"), 0, 0, &sealed).unwrap();
1087
1088 let ask = |max_series: usize| {
1089 series(
1090 &dir,
1091 &SeriesQuery {
1092 name: None,
1093 from: 0,
1094 to: 10_000,
1095 terms: Vec::new(),
1096 max_series,
1097 max_points: 100,
1098 },
1099 )
1100 .unwrap()
1101 };
1102
1103 // Under the cap: everything, and nothing refused.
1104 let r = ask(100);
1105 assert_eq!(r.json.matches(r#""peer""#).count(), 10, "{}", r.json);
1106 assert_eq!(r.stats.dropped_series, 0);
1107
1108 // Over it: the smallest keys survive, which is the same set the old
1109 // sort-then-truncate produced — that equivalence is the whole safety
1110 // argument for evicting the maximum as we go, so it is what this checks
1111 // rather than the count alone.
1112 let r = ask(8);
1113 for i in 0..10 {
1114 assert_eq!(
1115 r.json.contains(&format!(r#""peer":"p{i}""#)),
1116 i < 8,
1117 "p{i}: {}",
1118 r.json
1119 );
1120 }
1121 assert_eq!(r.stats.dropped_series, 2);
1122
1123 // The ponytail ceiling, asserted rather than left to be discovered: the
1124 // record of what was refused is itself capped at `max_series`, so with
1125 // six dropped and a cap of four the answer is four and reads as "at
1126 // least". Remembering every key exactly is the allocation the cap
1127 // exists to prevent.
1128 let r = ask(4);
1129 assert_eq!(r.json.matches(r#""peer""#).count(), 4, "{}", r.json);
1130 assert_eq!(r.stats.dropped_series, 4);
1131
1132 let _ = std::fs::remove_dir_all(&dir);
1133 }
1134
1135 /// A point's value lives in whichever column its table set, and a row that
1136 /// set none must yield no point rather than a zero. A gauge with no `value`
1137 /// on the wire means "not reported"; charting it as 0 invents a reading
1138 /// nobody took, and on a `.sum` it would drag an average down.
1139 ///
1140 /// The null cases are built by hand because today's writer never emits
1141 /// them — `count` is non-nullable and every gauge point has one of `int`
1142 /// and `double`. The reader is the half that outlives the writer that wrote
1143 /// the block, so it is the half that has to survive them.
1144 #[test]
1145 fn a_point_yields_the_value_of_whichever_column_its_table_actually_set() {
1146 for (table, want) in [
1147 ("number_dp", "number"),
1148 ("hist_dp", "countsum"),
1149 ("exp_hist_dp", "countsum"),
1150 ("summary_dp", "countsum"),
1151 // Not a point table. `collect_block` only ever asks about the four,
1152 // so this arm is what stops a fifth table added later from being
1153 // charted as whatever its first two columns happen to be called.
1154 ("logs", "none"),
1155 ] {
1156 let got = match Values::for_table(table) {
1157 Values::Number => "number",
1158 Values::CountSum => "countsum",
1159 Values::None => "none",
1160 };
1161 assert_eq!(got, want, "{table}");
1162 }
1163
1164 /// `(suffix, value)` as text, because `Pt` is deliberately not `Eq` —
1165 /// comparing two charted doubles for equality is a bug everywhere else.
1166 fn shape(v: Vec<(&'static str, Pt)>) -> Vec<String> {
1167 v.into_iter()
1168 .map(|(s, p)| match p {
1169 Pt::Int(i) => format!("{s}=i{i}"),
1170 Pt::Double(d) => format!("{s}=d{d}"),
1171 })
1172 .collect()
1173 }
1174
1175 let num = RecordBatch::try_from_iter(vec![
1176 (
1177 "int",
1178 Arc::new(Int64Array::from(vec![Some(7), None, None])) as ArrayRef,
1179 ),
1180 (
1181 "double",
1182 Arc::new(Float64Array::from(vec![None, Some(0.5), None])) as ArrayRef,
1183 ),
1184 ])
1185 .unwrap();
1186 assert_eq!(shape(Values::Number.at(&num, 0)), ["=i7"]);
1187 assert_eq!(shape(Values::Number.at(&num, 1)), ["=d0.5"]);
1188 assert!(shape(Values::Number.at(&num, 2)).is_empty());
1189
1190 // A writer that never emitted the column at all, rather than emitting
1191 // it null: the reader must fall through to the other one, not index a
1192 // column that is not there.
1193 let only_double = RecordBatch::try_from_iter(vec![(
1194 "double",
1195 Arc::new(Float64Array::from(vec![1.25])) as ArrayRef,
1196 )])
1197 .unwrap();
1198 assert_eq!(shape(Values::Number.at(&only_double, 0)), ["=d1.25"]);
1199
1200 let cs = RecordBatch::try_from_iter(vec![
1201 (
1202 "count",
1203 Arc::new(UInt64Array::from(vec![None, Some(4), Some(4), None])) as ArrayRef,
1204 ),
1205 (
1206 "sum",
1207 Arc::new(Float64Array::from(vec![Some(1.5), None, Some(2.5), None])) as ArrayRef,
1208 ),
1209 ])
1210 .unwrap();
1211 assert_eq!(shape(Values::CountSum.at(&cs, 0)), [".sum=d1.5"]);
1212 assert_eq!(shape(Values::CountSum.at(&cs, 1)), [".count=i4"]);
1213 assert_eq!(
1214 shape(Values::CountSum.at(&cs, 2)),
1215 [".count=i4", ".sum=d2.5"]
1216 );
1217 assert!(shape(Values::CountSum.at(&cs, 3)).is_empty());
1218
1219 // The half of a histogram that is missing is missing, not zero: a
1220 // `sum`-less point charts a `.count` series and no `.sum` series at
1221 // all. `starts_with`, not `==`: `shape` renders the value into the
1222 // string, so an equality against the bare suffix could never match and
1223 // would hold however many `.sum` points came back.
1224 assert!(
1225 !shape(Values::CountSum.at(&cs, 1))
1226 .iter()
1227 .any(|s| s.starts_with(SUM))
1228 );
1229 assert!(shape(Values::None.at(&cs, 2)).is_empty());
1230 }
1231
1232 /// Every kind OTLP can declare gets a name a caller can read, including the
1233 /// two that were added last and the descriptor with no `data` at all.
1234 ///
1235 /// This is the dropdown an agent reads before writing its first query. A
1236 /// kind rendered as the wrong word — or an exponential histogram rendered
1237 /// as "unset" because the match arm was never added — is a metric the
1238 /// caller cannot tell apart from one that is not reporting.
1239 #[test]
1240 fn every_metric_kind_reports_the_name_a_dropdown_shows() {
1241 let dir = tmp("series-kinds");
1242 let point = NumberDataPoint {
1243 time_unix_nano: 1_000,
1244 value: Some(NumValue::AsInt(1)),
1245 ..Default::default()
1246 };
1247 let named = |name: &str, unit: &str, data: Option<Data>| Metric {
1248 name: name.into(),
1249 unit: unit.into(),
1250 data,
1251 ..Default::default()
1252 };
1253 publish(
1254 &dir,
1255 0,
1256 None,
1257 None,
1258 vec![
1259 // No `data` oneof: a descriptor somebody's exporter believes in
1260 // that owns no points. It still belongs in the listing.
1261 named("a.declared", "1", None),
1262 named(
1263 "b.gauge",
1264 "By",
1265 Some(Data::Gauge(Gauge {
1266 data_points: vec![point.clone()],
1267 })),
1268 ),
1269 named(
1270 "c.sum",
1271 "1",
1272 Some(Data::Sum(Sum {
1273 data_points: vec![point],
1274 ..Default::default()
1275 })),
1276 ),
1277 named(
1278 "d.hist",
1279 "s",
1280 Some(Data::Histogram(Histogram {
1281 data_points: vec![HistogramDataPoint {
1282 time_unix_nano: 1_000,
1283 count: 2,
1284 sum: Some(3.0),
1285 ..Default::default()
1286 }],
1287 ..Default::default()
1288 })),
1289 ),
1290 named(
1291 "e.exp",
1292 "s",
1293 Some(Data::ExponentialHistogram(ExponentialHistogram {
1294 data_points: vec![ExponentialHistogramDataPoint {
1295 time_unix_nano: 1_000,
1296 count: 2,
1297 sum: Some(3.0),
1298 ..Default::default()
1299 }],
1300 ..Default::default()
1301 })),
1302 ),
1303 named(
1304 "f.summary",
1305 "s",
1306 Some(Data::Summary(Summary {
1307 data_points: vec![SummaryDataPoint {
1308 time_unix_nano: 1_000,
1309 count: 2,
1310 sum: 3.0,
1311 ..Default::default()
1312 }],
1313 })),
1314 ),
1315 ],
1316 );
1317
1318 let r = names(&dir, 0, 10_000).unwrap();
1319 assert_eq!(
1320 r.json,
1321 concat!(
1322 r#"[{"name":"a.declared","unit":"1","kind":"unset"},"#,
1323 r#"{"name":"b.gauge","unit":"By","kind":"gauge"},"#,
1324 r#"{"name":"c.sum","unit":"1","kind":"sum"},"#,
1325 r#"{"name":"d.hist","unit":"s","kind":"histogram"},"#,
1326 r#"{"name":"e.exp","unit":"s","kind":"exponential_histogram"},"#,
1327 r#"{"name":"f.summary","unit":"s","kind":"summary"}]"#,
1328 )
1329 );
1330
1331 // The same words come back on the series themselves, and the three
1332 // aggregate kinds each split into the two derived halves.
1333 let json = query(&dir, &wide()).json;
1334 for (name, kind) in [
1335 ("b.gauge", "gauge"),
1336 ("c.sum", "sum"),
1337 ("d.hist.count", "histogram"),
1338 ("d.hist.sum", "histogram"),
1339 ("e.exp.count", "exponential_histogram"),
1340 ("e.exp.sum", "exponential_histogram"),
1341 ("f.summary.count", "summary"),
1342 ("f.summary.sum", "summary"),
1343 ] {
1344 assert!(
1345 json.contains(&format!(r#""name":"{name}","unit":"#))
1346 && json.contains(&format!(r#""kind":"{kind}""#)),
1347 "{name}/{kind}: {json}"
1348 );
1349 }
1350 // A descriptor with no points charts nothing, however loudly it is
1351 // declared.
1352 assert!(!json.contains("a.declared"), "{json}");
1353
1354 let _ = std::fs::remove_dir_all(&dir);
1355 }
1356
1357 /// A block outside the window is never opened, and a series over the point
1358 /// cap keeps its newest points and says how many it dropped.
1359 ///
1360 /// Both are the difference between a chart and a lie. Opening every block
1361 /// makes the directory name — the whole index — worthless; keeping an
1362 /// arbitrary subset of points and sorting it at render time draws a
1363 /// contiguous line through whichever rows the directory listing reached
1364 /// first, with no sign that anything is missing.
1365 #[test]
1366 fn a_series_is_bounded_by_the_window_asked_for_and_the_points_it_can_chart() {
1367 let dir = tmp("series-bounds");
1368 let pts: Vec<(u64, i64)> = (1..=6).map(|i| (i as u64 * 1_000, i)).collect();
1369 publish(&dir, 0, None, None, vec![gauge("m", &pts)]);
1370 // Hours later, so it lands in its own partition and its own directory
1371 // name. Nothing in the window below can reach it.
1372 publish(
1373 &dir,
1374 1,
1375 None,
1376 None,
1377 vec![gauge("m", &[(9_000_000_000_000, 99)])],
1378 );
1379
1380 let mut q = wide();
1381 q.to = 10_000;
1382 q.max_points = 4;
1383 let r = query(&dir, &q);
1384
1385 assert_eq!(r.stats.blocks_total, 2);
1386 assert_eq!(r.stats.blocks_scanned, 1, "the far block was opened");
1387 // Six points offered, four charted, and the two dropped are the two
1388 // oldest — not the two the scan happened to see last.
1389 assert!(r.json.contains(r#""dropped_points":2"#), "{}", r.json);
1390 assert!(
1391 r.json
1392 .contains(r#""points":[["3000","3"],["4000","4"],["5000","5"],["6000","6"]]"#),
1393 "{}",
1394 r.json
1395 );
1396
1397 // Under the cap there is no `dropped_points` member at all: a chart
1398 // must be able to treat its presence as "something is missing".
1399 let mut q = wide();
1400 q.to = 10_000;
1401 let r = query(&dir, &q);
1402 assert!(!r.json.contains("dropped_points"), "{}", r.json);
1403 assert!(r.json.contains(r#"["1000","1"]"#), "{}", r.json);
1404
1405 let _ = std::fs::remove_dir_all(&dir);
1406 }
1407
1408 /// A term is satisfied at whichever of the four levels carries the
1409 /// attribute, and a block missing a level is filtered by the levels it has.
1410 ///
1411 /// Whether `service.name` is a resource attribute or a data point one is a
1412 /// detail of whoever configured the SDK. A filter that only looked at the
1413 /// point level would return nothing for the single most common metrics
1414 /// query there is, with a 200 and an empty list.
1415 #[test]
1416 fn a_term_is_satisfied_at_whichever_level_carries_the_attribute() {
1417 let dir = tmp("series-levels");
1418 publish(
1419 &dir,
1420 0,
1421 Some(Resource {
1422 attributes: vec![kv("service.name", "api")],
1423 ..Default::default()
1424 }),
1425 Some(InstrumentationScope {
1426 attributes: vec![kv("otel.lib", "sdk")],
1427 ..Default::default()
1428 }),
1429 vec![
1430 Metric {
1431 metadata: vec![kv("tier", "gold")],
1432 ..gauge("m1", &[(1_000, 1)])
1433 },
1434 Metric {
1435 ..gauge("m2", &[(1_000, 2)])
1436 },
1437 ],
1438 );
1439 // No resource and no scope: both attribute tables are empty, so
1440 // `publish` never writes them. Filtering must fall back to the levels
1441 // this block does have rather than treating the absence as a match.
1442 publish(
1443 &dir,
1444 1,
1445 None,
1446 None,
1447 vec![Metric {
1448 metadata: vec![kv("tier", "gold")],
1449 ..gauge("m3", &[(1_000, 3)])
1450 }],
1451 );
1452
1453 // Point-level attributes, added after the fact so each metric above
1454 // stays readable.
1455 let with_pod = |name: &str, v: i64, pod: &str| Metric {
1456 data: Some(Data::Gauge(Gauge {
1457 data_points: vec![NumberDataPoint {
1458 time_unix_nano: 1_000,
1459 value: Some(NumValue::AsInt(v)),
1460 attributes: vec![kv("pod", pod)],
1461 ..Default::default()
1462 }],
1463 })),
1464 ..Metric {
1465 name: name.into(),
1466 ..Default::default()
1467 }
1468 };
1469 publish(
1470 &dir,
1471 2,
1472 Some(Resource {
1473 attributes: vec![kv("service.name", "db")],
1474 ..Default::default()
1475 }),
1476 None,
1477 vec![with_pod("m4", 4, "p4")],
1478 );
1479
1480 let ask = |key: &str, want: &str| {
1481 let mut q = wide();
1482 q.to = 10_000;
1483 q.terms = vec![Term {
1484 target: crate::query::Target::Attr(key.into()),
1485 op: crate::query::Op::Eq,
1486 value: crate::query::Value::Str(want.into()),
1487 }];
1488 let r = query(&dir, &q);
1489 let mut names: Vec<String> = Vec::new();
1490 for m in ["m1", "m2", "m3", "m4"] {
1491 if r.json.contains(&format!(r#""name":"{m}""#)) {
1492 names.push(m.to_owned());
1493 }
1494 }
1495 names
1496 };
1497
1498 // Resource level. The second block has no resource attributes at all
1499 // and must not match on the strength of not having them.
1500 assert_eq!(ask("service.name", "api"), ["m1", "m2"]);
1501 assert_eq!(ask("service.name", "db"), ["m4"]);
1502 // Scope level.
1503 assert_eq!(ask("otel.lib", "sdk"), ["m1", "m2"]);
1504 // Metric level — `Metric.metadata`, which is per descriptor row and so
1505 // reaches both blocks that declare it.
1506 assert_eq!(ask("tier", "gold"), ["m1", "m3"]);
1507 // Point level.
1508 assert_eq!(ask("pod", "p4"), ["m4"]);
1509 // A term nothing carries at any level matches nothing, rather than
1510 // falling back to "no opinion" and matching everything.
1511 assert!(ask("tier", "bronze").is_empty());
1512 assert!(ask("nope", "x").is_empty());
1513
1514 let _ = std::fs::remove_dir_all(&dir);
1515 }
1516
1517 /// The four attribute levels merge with the most specific winning, and the
1518 /// merged map is what identifies the series.
1519 ///
1520 /// Attributes are half the grouping key, so getting the precedence wrong
1521 /// does not just mislabel a line — it splits one series into two, or fuses
1522 /// two into one, and neither is visible in the response.
1523 #[test]
1524 fn attributes_from_every_level_merge_with_the_most_specific_winning() {
1525 let dir = tmp("series-merge");
1526 publish(
1527 &dir,
1528 0,
1529 Some(Resource {
1530 attributes: vec![kv("env", "prod"), kv("region", "eu")],
1531 ..Default::default()
1532 }),
1533 None,
1534 vec![Metric {
1535 // Same key as the resource carries, one level down.
1536 metadata: vec![kv("env", "staging")],
1537 data: Some(Data::Gauge(Gauge {
1538 data_points: vec![NumberDataPoint {
1539 time_unix_nano: 1_000,
1540 value: Some(NumValue::AsInt(1)),
1541 // And again, one level further down.
1542 attributes: vec![kv("env", "canary"), kv("pod", "x")],
1543 ..Default::default()
1544 }],
1545 })),
1546 ..Metric {
1547 name: "m".into(),
1548 ..Default::default()
1549 }
1550 }],
1551 );
1552
1553 let json = query(&dir, &wide()).json;
1554 // `env` resolves to the point's own value; the keys only an upper level
1555 // carries survive; and the losing values appear nowhere.
1556 assert!(
1557 json.contains(r#""attributes":{"env":"canary","pod":"x","region":"eu"}"#),
1558 "{json}"
1559 );
1560 assert!(
1561 !json.contains("prod") && !json.contains("staging"),
1562 "{json}"
1563 );
1564
1565 let _ = std::fs::remove_dir_all(&dir);
1566 }
1567
1568 /// Exemplars are capped per series, because nothing in OTLP bounds how many
1569 /// an exporter attaches to a point.
1570 ///
1571 /// Without the cap one misconfigured SDK inflates every chart response that
1572 /// touches its metric — and this is the response a chart polls on a timer.
1573 #[test]
1574 fn exemplars_are_capped_so_one_exporter_cannot_inflate_every_chart_response() {
1575 let dir = tmp("series-exemplars");
1576 publish(
1577 &dir,
1578 0,
1579 None,
1580 None,
1581 vec![Metric {
1582 data: Some(Data::Gauge(Gauge {
1583 data_points: vec![NumberDataPoint {
1584 time_unix_nano: 1_000,
1585 value: Some(NumValue::AsInt(1)),
1586 exemplars: (0..MAX_EXEMPLARS as i64 + 6)
1587 .map(|i| Exemplar {
1588 time_unix_nano: 1_000 + i as u64,
1589 value: Some(exemplar::Value::AsInt(i)),
1590 trace_id: vec![7u8; 16].into(),
1591 ..Default::default()
1592 })
1593 .collect(),
1594 ..Default::default()
1595 }],
1596 })),
1597 ..Metric {
1598 name: "m".into(),
1599 ..Default::default()
1600 }
1601 }],
1602 );
1603
1604 let json = query(&dir, &wide()).json;
1605 // Points carry their timestamp as a bare array element, so every
1606 // occurrence of the key is one exemplar object.
1607 assert_eq!(
1608 json.matches(r#""time_unix_nano""#).count(),
1609 MAX_EXEMPLARS,
1610 "{json}"
1611 );
1612 // Capped, not dropped: the ones that are there carry the trace id that
1613 // is the whole point of an exemplar.
1614 assert!(json.contains(r#""trace_id":"07070707"#), "{json}");
1615
1616 let _ = std::fs::remove_dir_all(&dir);
1617 }
1618
1619 /// A table shaped differently from the one this reader writes contributes
1620 /// nothing, rather than panicking or charting garbage.
1621 ///
1622 /// Two things produce that shape and neither is an error: retention
1623 /// unlinking a block while a query walks it, and a block written by another
1624 /// version of the binary — the block directory is the manifest, so there is
1625 /// no schema version to check against and no coordinator to ask.
1626 #[test]
1627 fn a_table_the_reader_did_not_write_contributes_nothing_instead_of_panicking() {
1628 let dir = tmp("series-ragged");
1629 publish(&dir, 0, None, None, vec![gauge("ok", &[(1_000, 1)])]);
1630 let unlinked = publish(&dir, 1, None, None, vec![gauge("gone", &[(1_000, 2)])]);
1631 let ragged = publish(&dir, 2, None, None, vec![gauge("ragged", &[(1_000, 3)])]);
1632
1633 // Retention got here first: the descriptor table is gone, but the
1634 // directory is still listed.
1635 std::fs::remove_file(unlinked.join("metrics.arrow")).unwrap();
1636
1637 // An older writer's point table, with no `id` column to join
1638 // attributes and exemplars on. Staged and renamed, because a mapping
1639 // over a file being truncated is a SIGBUS, not an error.
1640 let table = ragged.join("number_dp.arrow");
1641 let mut b = crate::block::open_table_opt(&table)
1642 .unwrap()
1643 .unwrap()
1644 .batches[0]
1645 .clone();
1646 b.remove_column(b.schema().index_of("id").unwrap());
1647 let staged = ragged.join("number_dp.arrow.staged");
1648 crate::block::write_table(&staged, &b).unwrap();
1649 std::fs::rename(&staged, &table).unwrap();
1650
1651 let r = query(&dir, &wide());
1652 assert_eq!(r.stats.blocks_scanned, 3, "all three were in the window");
1653 assert!(r.json.contains(r#""name":"ok""#), "{}", r.json);
1654 assert!(!r.json.contains("gone"), "{}", r.json);
1655 assert!(!r.json.contains("ragged"), "{}", r.json);
1656
1657 // The name listing reads only the descriptor table, so the ragged block
1658 // still names its metric and the unlinked one names nothing.
1659 let n = names(&dir, 0, 10_000).unwrap();
1660 assert_eq!(
1661 n.json,
1662 concat!(
1663 r#"[{"name":"ok","unit":"","kind":"gauge"},"#,
1664 r#"{"name":"ragged","unit":"","kind":"gauge"}]"#,
1665 )
1666 );
1667
1668 // The same tolerance one layer down, where the missing column is the
1669 // one the whole index is built on.
1670 let no_parent = RecordBatch::try_from_iter(vec![(
1671 "id",
1672 Arc::new(UInt32Array::from(vec![0u32, 1])) as ArrayRef,
1673 )])
1674 .unwrap();
1675 assert!(index_by_parent(&no_parent).is_empty());
1676
1677 // And on a descriptor this renderer did not render: the suffix is
1678 // dropped rather than spliced into the middle of some other member.
1679 // Returning the input unchanged keeps the string valid JSON, which a
1680 // `find`-and-insert on a miss would not.
1681 assert_eq!(
1682 with_suffix(r#""kind":"histogram""#, COUNT),
1683 r#""kind":"histogram""#
1684 );
1685 assert_eq!(
1686 with_suffix(r#""name":"x","unit":"s""#, ""),
1687 r#""name":"x","unit":"s""#
1688 );
1689 assert_eq!(
1690 with_suffix(r#""name":"x","unit":"s""#, COUNT),
1691 r#""name":"x.count","unit":"s""#
1692 );
1693
1694 let _ = std::fs::remove_dir_all(&dir);
1695 }
1696
1697 /// What one data point costs to render, at the point count that makes the
1698 /// attribute join visible.
1699 ///
1700 /// Scaled rather than `#[ignore]`d, exactly like
1701 /// [`crate::query::scan_cost_per_row`]: the default 2,000 points run in the
1702 /// normal suite as a correctness check on the merge, and the same code is
1703 /// the measurement at a real point count.
1704 ///
1705 /// ```sh
1706 /// MIRA_BENCH_POINTS=50000 cargo test --release -p miradb-core \
1707 /// --lib series_cost_per_point -- --nocapture
1708 /// ```
1709 ///
1710 /// The shape is the one a real exporter produces and the one the old
1711 /// [`collect_attrs`] was quadratic in: a few distinct attribute sets over
1712 /// many timestamps, so the series map stays small while `dp_attrs` grows
1713 /// with the point count. Scanning the whole table per point is `2n²`
1714 /// comparisons; the run search is two per point.
1715 #[test]
1716 fn series_cost_per_point() {
1717 let n: usize = std::env::var("MIRA_BENCH_POINTS")
1718 .ok()
1719 .and_then(|s| s.parse().ok())
1720 .unwrap_or(2_000);
1721 let dir = tmp("series-bench");
1722 const PODS: usize = 20;
1723 let points: Vec<NumberDataPoint> = (0..n)
1724 .map(|i| NumberDataPoint {
1725 time_unix_nano: 1_000 + i as u64,
1726 value: Some(NumValue::AsInt(i as i64)),
1727 attributes: vec![
1728 kv("k8s.pod.name", &format!("checkout-{}", i % PODS)),
1729 kv("cloud.availability_zone", &format!("z{}", i % 3)),
1730 ],
1731 ..Default::default()
1732 })
1733 .collect();
1734 publish(
1735 &dir,
1736 0,
1737 Some(Resource {
1738 attributes: vec![kv("service.name", "checkout")],
1739 ..Default::default()
1740 }),
1741 None,
1742 vec![Metric {
1743 name: "http.server.requests".into(),
1744 data: Some(Data::Gauge(Gauge {
1745 data_points: points,
1746 })),
1747 ..Default::default()
1748 }],
1749 );
1750 let q = SeriesQuery {
1751 name: Some("http.server.requests".into()),
1752 max_points: n,
1753 ..wide()
1754 };
1755
1756 // Once to fault the block in, so the number below is the join and not
1757 // the first touch of every page.
1758 let warm = query(&dir, &q);
1759 assert_eq!(warm.stats.rows_matched, n);
1760
1761 let t = std::time::Instant::now();
1762 let r = query(&dir, &q);
1763 let el = t.elapsed();
1764
1765 // The correctness half, which is why this is not `#[ignore]`d: every
1766 // point kept its own pod and inherited the resource's service, so the
1767 // binary search found the right run and not its neighbour.
1768 let series = r.json.matches(r#""name":"#).count();
1769 assert_eq!(series, PODS * 3, "one series per distinct attribute set");
1770 for pod in 0..PODS {
1771 assert!(
1772 r.json
1773 .contains(&format!(r#""k8s.pod.name":"checkout-{pod}""#)),
1774 "pod {pod} lost its own attributes"
1775 );
1776 }
1777 assert!(r.json.contains(r#""service.name":"checkout""#));
1778
1779 println!(
1780 "series: {n} points in {el:?} {:.3} us/point {series} series",
1781 el.as_secs_f64() * 1e6 / n as f64,
1782 );
1783 let _ = std::fs::remove_dir_all(&dir);
1784 }
1785}