netget/client/prometheus/
exposition.rs

1//! Bounded ingestion of classic text 0.0.4 and OpenMetrics text 1.0.0.
2//! Samples retain their wire names; family names differ for counters between the formats.
3use anyhow::{bail, ensure, Context, Result};
4use serde_json::{json, Map, Value};
5use std::collections::{BTreeMap, BTreeSet};
6
7pub const MAX_BODY: usize = 4 * 1024 * 1024;
8pub const MAX_SAMPLES: usize = 20_000;
9pub const MAX_FAMILIES: usize = 4096;
10pub const MAX_LABELS: usize = 64;
11pub const MAX_NAME: usize = 256;
12pub const MAX_TEXT: usize = 16 * 1024;
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum Format {
16    Text,
17    OpenMetrics,
18}
19impl Format {
20    pub fn name(self) -> &'static str {
21        match self {
22            Self::Text => "text",
23            Self::OpenMetrics => "openmetrics",
24        }
25    }
26}
27pub fn content_type(header: &str) -> Result<Format> {
28    let mut parts = header.split(';');
29    let format = match parts
30        .next()
31        .unwrap_or("")
32        .trim()
33        .to_ascii_lowercase()
34        .as_str()
35    {
36        "text/plain" => Format::Text,
37        "application/openmetrics-text" => Format::OpenMetrics,
38        _ => bail!("unsupported metrics Content-Type"),
39    };
40    let mut seen = BTreeSet::new();
41    for part in parts {
42        let (key, value) = part
43            .trim()
44            .split_once('=')
45            .context("malformed Content-Type parameter")?;
46        let key = key.trim().to_ascii_lowercase();
47        let value = value.trim().trim_matches('"');
48        ensure!(seen.insert(key.clone()), "duplicate Content-Type parameter");
49        match key.as_str() {
50            "version" => ensure!(
51                value
52                    == if format == Format::Text {
53                        "0.0.4"
54                    } else {
55                        "1.0.0"
56                    },
57                "unsupported metrics format version"
58            ),
59            "charset" => ensure!(value.eq_ignore_ascii_case("utf-8"), "metrics must be UTF-8"),
60            "escaping" => ensure!(value == "underscores", "unsupported metrics name escaping"),
61            _ => bail!("unsupported metrics Content-Type parameter {key}"),
62        }
63    }
64    Ok(format)
65}
66fn name(s: &str, metric: bool) -> Result<()> {
67    ensure!(
68        !s.is_empty() && s.len() <= MAX_NAME,
69        "metric/label name length"
70    );
71    ensure!(
72        s.bytes().enumerate().all(|(i, c)| c.is_ascii_alphabetic()
73            || c == b'_'
74            || (metric && c == b':')
75            || (i > 0 && c.is_ascii_digit())),
76        "unsupported metric/label name {s:?}"
77    );
78    Ok(())
79}
80fn number(s: &str) -> Result<(f64, Value)> {
81    let n = match s.to_ascii_lowercase().as_str() {
82        "nan" => f64::NAN,
83        "inf" | "+inf" | "infinity" | "+infinity" => f64::INFINITY,
84        "-inf" | "-infinity" => f64::NEG_INFINITY,
85        _ => {
86            ensure!(
87                !s.is_empty()
88                    && s.bytes()
89                        .all(|c| c.is_ascii_digit() || b"+-.eE".contains(&c)),
90                "invalid metric number {s:?}"
91            );
92            let n = s.parse::<f64>().context("invalid metric number")?;
93            ensure!(
94                n.is_finite(),
95                "metric number overflow; use explicit infinity"
96            );
97            n
98        }
99    };
100    Ok((
101        n,
102        if n.is_nan() {
103            json!("NaN")
104        } else if n == f64::INFINITY {
105            json!("+Inf")
106        } else if n == f64::NEG_INFINITY {
107            json!("-Inf")
108        } else {
109            json!(n)
110        },
111    ))
112}
113fn unescape(s: &str, quote: bool) -> Result<String> {
114    let mut result = String::new();
115    let mut chars = s.chars();
116    while let Some(c) = chars.next() {
117        if c == '\\' {
118            result.push(match chars.next() {
119                Some('\\') => '\\',
120                Some('n') => '\n',
121                Some('"') if quote => '"',
122                _ => bail!("invalid exposition escape"),
123            });
124        } else {
125            result.push(c);
126        }
127        ensure!(result.len() <= MAX_TEXT, "exposition text limit");
128    }
129    Ok(result)
130}
131struct Cursor<'a> {
132    rest: &'a str,
133}
134impl<'a> Cursor<'a> {
135    fn whitespace(&mut self) {
136        self.rest = self.rest.trim_start_matches([' ', '\t']);
137    }
138    fn token(&mut self) -> Result<&'a str> {
139        let n = self.rest.find([' ', '\t']).unwrap_or(self.rest.len());
140        let token = &self.rest[..n];
141        ensure!(!token.is_empty(), "missing exposition token");
142        self.rest = &self.rest[n..];
143        self.whitespace();
144        Ok(token)
145    }
146    fn labels(&mut self, format: Format) -> Result<Map<String, Value>> {
147        let mut result = Map::new();
148        ensure!(self.rest.starts_with('{'), "expected labels");
149        self.rest = &self.rest[1..];
150        self.whitespace();
151        if self.rest.starts_with('}') {
152            self.rest = &self.rest[1..];
153            return Ok(result);
154        }
155        loop {
156            ensure!(result.len() < MAX_LABELS, "label count limit");
157            let n = self.rest.find('=').context("missing label equals")?;
158            let key = self.rest[..n].trim();
159            name(key, false)?;
160            ensure!(key != "__name__", "reserved __name__ label");
161            self.rest = self.rest[n + 1..].trim_start_matches([' ', '\t']);
162            ensure!(self.rest.starts_with('"'), "label value must be quoted");
163            self.rest = &self.rest[1..];
164            let mut escaped = false;
165            let mut end = None;
166            for (i, c) in self.rest.char_indices() {
167                if c == '"' && !escaped {
168                    end = Some(i);
169                    break;
170                }
171                if c == '\\' && !escaped {
172                    escaped = true;
173                } else {
174                    escaped = false;
175                }
176            }
177            let end = end.context("unterminated label value")?;
178            let value = unescape(&self.rest[..end], true)?;
179            ensure!(
180                result.insert(key.into(), json!(value)).is_none(),
181                "duplicate label {key}"
182            );
183            self.rest = &self.rest[end + 1..];
184            self.whitespace();
185            if self.rest.starts_with('}') {
186                self.rest = &self.rest[1..];
187                break;
188            }
189            ensure!(self.rest.starts_with(','), "missing label comma");
190            self.rest = &self.rest[1..];
191            self.whitespace();
192            if self.rest.starts_with('}') {
193                ensure!(format == Format::Text, "OpenMetrics trailing label comma");
194                self.rest = &self.rest[1..];
195                break;
196            }
197        }
198        Ok(result)
199    }
200}
201#[derive(Default)]
202struct Family {
203    kind: Option<String>,
204    help: Option<String>,
205    unit: Option<String>,
206    samples: Vec<Value>,
207}
208fn suffixes(kind: &str, format: Format) -> &'static [&'static str] {
209    match (kind, format) {
210        ("counter", Format::OpenMetrics) => &["_total", "_created"],
211        ("histogram", Format::OpenMetrics) => &["_bucket", "_sum", "_count", "_created"],
212        ("histogram", _) => &["_bucket", "_sum", "_count"],
213        ("gaugehistogram", _) => &["_bucket", "_gsum", "_gcount"],
214        ("summary", Format::OpenMetrics) => &["", "_sum", "_count", "_created"],
215        ("summary", _) => &["", "_sum", "_count"],
216        ("info", _) => &["_info"],
217        _ => &[""],
218    }
219}
220#[derive(Default)]
221struct HistogramPoint {
222    last_bucket: Option<(f64, f64)>,
223    count: Option<f64>,
224}
225fn validate_histogram(family: &Family) -> Result<()> {
226    let mut points: BTreeMap<String, HistogramPoint> = BTreeMap::new();
227    for sample in &family.samples {
228        let suffix = sample["suffix"].as_str().unwrap();
229        let mut labels = sample["labels"].as_object().unwrap().clone();
230        let bound = labels.remove("le");
231        let key = serde_json::to_string(&labels)?;
232        let point = points.entry(key).or_default();
233        let value = sample["value"]
234            .as_f64()
235            .or_else(|| {
236                sample["value"]
237                    .as_str()
238                    .and_then(|s| number(s).ok().map(|v| v.0))
239            })
240            .context("invalid histogram value")?;
241        if suffix == "_bucket" {
242            let (bound, _) = number(
243                bound
244                    .as_ref()
245                    .and_then(Value::as_str)
246                    .context("bucket missing le")?,
247            )?;
248            ensure!(
249                !bound.is_nan() && value.is_finite() && value >= 0.0 && value.fract() == 0.0,
250                "histogram bucket count must be a finite nonnegative integer"
251            );
252            if let Some((previous, count)) = point.last_bucket {
253                ensure!(
254                    bound > previous && value >= count,
255                    "histogram buckets must increase and counts be cumulative"
256                );
257            }
258            point.last_bucket = Some((bound, value));
259        } else {
260            ensure!(bound.is_none(), "le label on a non-bucket histogram sample");
261            if suffix == "_count" || suffix == "_gcount" {
262                ensure!(
263                    value.is_finite() && value >= 0.0 && value.fract() == 0.0,
264                    "histogram count must be a finite nonnegative integer"
265                );
266                point.count = Some(value);
267            }
268        }
269    }
270    for point in points.values() {
271        let (bound, count) = point
272            .last_bucket
273            .context("histogram point has no buckets")?;
274        ensure!(
275            bound == f64::INFINITY,
276            "histogram point missing +Inf bucket"
277        );
278        if let Some(total) = point.count {
279            ensure!(total == count, "histogram +Inf bucket disagrees with count");
280        }
281    }
282    Ok(())
283}
284pub fn parse(body: &[u8], format: Format) -> Result<Value> {
285    ensure!(body.len() <= MAX_BODY, "metrics body limit");
286    let text = std::str::from_utf8(body).context("metrics body is not UTF-8")?;
287    ensure!(!text.starts_with('\u{feff}'), "metrics byte order mark");
288    if format == Format::Text {
289        ensure!(
290            text.is_empty() || text.ends_with('\n'),
291            "text exposition missing final newline"
292        );
293    }
294    if format == Format::OpenMetrics {
295        ensure!(!text.contains('\r'), "OpenMetrics carriage return");
296    }
297    let mut families: BTreeMap<String, Family> = BTreeMap::new();
298    let mut aliases: BTreeMap<String, (String, &'static str)> = BTreeMap::new();
299    let mut series = BTreeSet::new();
300    let mut count = 0;
301    let mut eof = false;
302    let mut current: Option<String> = None;
303    let mut closed = BTreeSet::new();
304    for (line_index, line) in text.split_terminator('\n').enumerate() {
305        let line = line.trim_matches([' ', '\t']);
306        ensure!(!eof, "data after OpenMetrics EOF");
307        if line.is_empty() {
308            continue;
309        }
310        if line == "# EOF" {
311            ensure!(
312                format == Format::OpenMetrics,
313                "unexpected EOF in classic text"
314            );
315            eof = true;
316            continue;
317        }
318        if let Some(comment) = line.strip_prefix('#') {
319            let mut cursor = Cursor {
320                rest: comment.trim_start_matches([' ', '\t']),
321            };
322            if cursor.rest.is_empty() {
323                ensure!(format == Format::Text, "empty OpenMetrics comment");
324                continue;
325            }
326            let key = cursor.token()?;
327            if !["HELP", "TYPE", "UNIT"].contains(&key) {
328                ensure!(format == Format::Text, "unsupported OpenMetrics comment");
329                continue;
330            }
331            let metric = cursor.token()?;
332            name(metric, true)?;
333            ensure!(
334                families.contains_key(metric) || families.len() < MAX_FAMILIES,
335                "metric family limit"
336            );
337            let f = families.entry(metric.into()).or_default();
338            ensure!(f.samples.is_empty(), "metadata after samples for {metric}");
339            match key {
340                "TYPE" => {
341                    ensure!(f.kind.is_none(), "duplicate TYPE {metric}");
342                    let kind = cursor.token()?;
343                    let allowed = if format == Format::Text {
344                        &["counter", "gauge", "histogram", "summary", "untyped"][..]
345                    } else {
346                        &[
347                            "counter",
348                            "gauge",
349                            "histogram",
350                            "summary",
351                            "unknown",
352                            "gaugehistogram",
353                            "info",
354                            "stateset",
355                        ][..]
356                    };
357                    ensure!(
358                        allowed.contains(&kind) && cursor.rest.is_empty(),
359                        "invalid metric TYPE"
360                    );
361                    f.kind = Some(kind.into());
362                    for suffix in suffixes(kind, format) {
363                        ensure!(
364                            aliases
365                                .insert(format!("{metric}{suffix}"), (metric.into(), suffix))
366                                .is_none(),
367                            "ambiguous metric family/sample names"
368                        );
369                    }
370                }
371                "HELP" => {
372                    ensure!(f.help.is_none(), "duplicate HELP {metric}");
373                    f.help = Some(unescape(cursor.rest, format == Format::OpenMetrics)?);
374                }
375                _ => {
376                    ensure!(
377                        format == Format::OpenMetrics && f.unit.is_none(),
378                        "unsupported or duplicate UNIT"
379                    );
380                    let unit = cursor.rest;
381                    if !unit.is_empty() {
382                        name(unit, false)?;
383                        ensure!(
384                            metric.ends_with(&format!("_{unit}")),
385                            "UNIT must suffix family name"
386                        );
387                    }
388                    f.unit = Some(unit.into());
389                }
390            }
391            continue;
392        }
393        count += 1;
394        ensure!(count <= MAX_SAMPLES, "metric sample limit");
395        let mut cursor = Cursor { rest: line };
396        let end = line
397            .find(['{', ' ', '\t'])
398            .context("sample missing value")?;
399        let sample_name = &line[..end];
400        name(sample_name, true)?;
401        cursor.rest = &line[end..];
402        let labels = if cursor.rest.starts_with('{') {
403            cursor.labels(format)?
404        } else {
405            Map::new()
406        };
407        ensure!(
408            cursor.rest.starts_with([' ', '\t']),
409            "sample value must be separated"
410        );
411        cursor.whitespace();
412        let (value, value_json) =
413            number(cursor.token()?).with_context(|| format!("line {}", line_index + 1))?;
414        let mut sample = json!({"name":sample_name,"labels":labels,"value":value_json});
415        if !cursor.rest.is_empty() && !cursor.rest.starts_with('#') {
416            let timestamp = cursor.token()?;
417            if format == Format::Text {
418                sample["timestamp_ms"] = json!(timestamp
419                    .parse::<i64>()
420                    .context("text timestamp must be int64 milliseconds")?);
421            } else {
422                let (n, v) = number(timestamp)?;
423                ensure!(n.is_finite(), "nonfinite OpenMetrics timestamp");
424                sample["timestamp_seconds"] = v;
425            }
426        }
427        if !cursor.rest.is_empty() {
428            ensure!(
429                format == Format::OpenMetrics && cursor.rest.starts_with("# "),
430                "unexpected trailing sample content"
431            );
432            cursor.rest = &cursor.rest[2..];
433            let labels = cursor.labels(format)?;
434            ensure!(!labels.is_empty(), "empty exemplar labels");
435            ensure!(
436                labels
437                    .iter()
438                    .map(|(k, v)| k.len() + v.as_str().unwrap().chars().count())
439                    .sum::<usize>()
440                    <= 128,
441                "exemplar label length"
442            );
443            ensure!(
444                cursor.rest.starts_with(' '),
445                "exemplar value must be separated"
446            );
447            cursor.whitespace();
448            let (_, v) = number(cursor.token()?)?;
449            let mut exemplar = json!({"labels":labels,"value":v});
450            if !cursor.rest.is_empty() {
451                let (n, v) = number(cursor.token()?)?;
452                ensure!(n.is_finite(), "nonfinite exemplar timestamp");
453                exemplar["timestamp_seconds"] = v;
454            }
455            ensure!(cursor.rest.is_empty(), "trailing exemplar content");
456            sample["exemplar"] = exemplar;
457        }
458        ensure!(
459            series.insert(serde_json::to_string(&(sample_name, &sample["labels"]))?),
460            "duplicate metric series"
461        );
462        let (family, suffix) = aliases
463            .get(sample_name)
464            .cloned()
465            .unwrap_or((sample_name.into(), ""));
466        ensure!(!closed.contains(&family), "interleaved metric families");
467        if current.as_ref() != Some(&family) {
468            if let Some(previous) = current.replace(family.clone()) {
469                closed.insert(previous);
470            }
471        }
472        ensure!(
473            families.contains_key(&family) || families.len() < MAX_FAMILIES,
474            "metric family limit"
475        );
476        let f = families.entry(family).or_default();
477        let kind = f.kind.as_deref().unwrap_or(if format == Format::Text {
478            "untyped"
479        } else {
480            "unknown"
481        });
482        ensure!(
483            suffixes(kind, format).contains(&suffix),
484            "sample does not match declared metric type"
485        );
486        sample["suffix"] = json!(suffix);
487        if suffix == "_bucket" {
488            let le = sample["labels"]["le"]
489                .as_str()
490                .context("histogram bucket missing le")?;
491            let (n, _) = number(le)?;
492            ensure!(!n.is_nan(), "NaN bucket bound");
493        }
494        if kind == "summary" && suffix.is_empty() {
495            let q = sample["labels"]["quantile"]
496                .as_str()
497                .context("summary missing quantile")?;
498            let (q, _) = number(q)?;
499            ensure!((0.0..=1.0).contains(&q), "quantile outside 0..1");
500        }
501        if format == Format::OpenMetrics {
502            if kind == "info" {
503                ensure!(value == 1.0, "info value must be 1");
504            }
505            if kind == "stateset" {
506                ensure!(
507                    value == 0.0 || value == 1.0,
508                    "stateset value must be boolean"
509                );
510            }
511            if (kind == "counter" && suffix == "_total")
512                || ["_bucket", "_count", "_gcount", "_created"].contains(&suffix)
513            {
514                ensure!(
515                    !value.is_nan() && value >= 0.0,
516                    "negative/NaN counter/count/created"
517                );
518            }
519            if ["_bucket", "_count", "_gcount"].contains(&suffix) {
520                ensure!(
521                    value.is_finite() && value.fract() == 0.0,
522                    "count must be a finite integer"
523                );
524            }
525            ensure!(
526                sample.get("exemplar").is_none()
527                    || (kind == "counter" && suffix == "_total")
528                    || suffix == "_bucket",
529                "exemplar on unsupported sample"
530            );
531        }
532        f.samples.push(sample);
533    }
534    ensure!(format == Format::Text || eof, "OpenMetrics missing EOF");
535    for f in families.values() {
536        if matches!(f.kind.as_deref(), Some("histogram" | "gaugehistogram")) {
537            validate_histogram(f)?;
538        }
539        if matches!(f.kind.as_deref(), Some("info" | "stateset")) {
540            ensure!(
541                f.unit.as_deref().unwrap_or("").is_empty(),
542                "info/stateset must have no unit"
543            );
544        }
545    }
546    let metrics:Vec<Value>=families.into_iter().map(|(name,f)|json!({"name":name,"type":f.kind.unwrap_or_else(||if format==Format::Text {"untyped"} else {"unknown"}.into()),"help":f.help,"unit":f.unit,"samples":f.samples})).collect();
547    Ok(json!({"format":format.name(),"sample_count":count,"metrics":metrics}))
548}
netget/client/prometheus/exposition.rs

netget/client/prometheus/
exposition.rs
