netget/server/graphite/
codec.rs

1//! Carbon plaintext: UTF-8 metric path, finite value and timestamp, terminated by LF.
2use anyhow::{ensure, Context, Result};
3use serde::{Deserialize, Serialize};
4
5pub const MAX_LINE_BYTES: usize = 4096;
6pub const MAX_BATCH_BYTES: usize = 64 * 1024;
7pub const MAX_BATCH_RECORDS: usize = 256;
8pub const READ_BYTES: usize = 8192;
9pub const DEFAULT_LLM_FALLBACK: bool = false;
10
11#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
12#[serde(deny_unknown_fields)]
13pub struct Metric {
14    pub path: String,
15    pub value: f64,
16    pub timestamp: f64,
17}
18impl Metric {
19    pub fn validate(&self) -> Result<()> {
20        ensure!(
21            !self.path.is_empty()
22                && !self
23                    .path
24                    .chars()
25                    .any(|c| c.is_whitespace() || c.is_control()),
26            "metric path must be nonempty without whitespace or control characters"
27        );
28        ensure!(self.value.is_finite(), "metric value must be finite");
29        ensure!(
30            self.timestamp.is_finite() && (self.timestamp >= 0.0 || self.timestamp == -1.0),
31            "timestamp must be nonnegative UNIX seconds, or -1 for receiver time"
32        );
33        Ok(())
34    }
35}
36pub fn parse_line(line: &[u8]) -> Result<Metric> {
37    ensure!(
38        !line.is_empty() && line.len() <= MAX_LINE_BYTES,
39        "invalid metric line length"
40    );
41    let text = std::str::from_utf8(line).context("metric line must be UTF-8")?;
42    let mut fields = text.split_ascii_whitespace();
43    let metric = Metric {
44        path: fields.next().context("missing metric path")?.into(),
45        value: fields
46            .next()
47            .context("missing value")?
48            .parse()
49            .context("invalid metric value")?,
50        timestamp: fields
51            .next()
52            .context("missing timestamp")?
53            .parse()
54            .context("invalid timestamp")?,
55    };
56    ensure!(fields.next().is_none(), "extra fields in metric line");
57    metric.validate()?;
58    Ok(metric)
59}
60pub fn encode_batch(metrics: &[Metric]) -> Result<Vec<u8>> {
61    ensure!(
62        !metrics.is_empty() && metrics.len() <= MAX_BATCH_RECORDS,
63        "metric batch must contain 1..256 records"
64    );
65    let mut out = Vec::new();
66    for metric in metrics {
67        metric.validate()?;
68        let line = format!("{} {} {}\n", metric.path, metric.value, metric.timestamp);
69        ensure!(
70            line.len() - 1 <= MAX_LINE_BYTES,
71            "metric line exceeds bound"
72        );
73        ensure!(
74            out.len() + line.len() <= MAX_BATCH_BYTES,
75            "metric batch exceeds bound"
76        );
77        out.extend_from_slice(line.as_bytes());
78    }
79    Ok(out)
80}
81
82/// TCP has no batch boundaries. Drain at most 256 complete lines from one bounded
83/// read before dispatching; retain the partial next line for the following read.
84#[derive(Default)]
85pub struct Decoder {
86    pending: Vec<u8>,
87}
88impl Decoder {
89    pub fn feed(&mut self, bytes: &[u8]) -> Result<()> {
90        ensure!(
91            self.pending.len() + bytes.len() <= MAX_LINE_BYTES + READ_BYTES,
92            "pending stream buffer exceeds bound"
93        );
94        self.pending.extend_from_slice(bytes);
95        Ok(())
96    }
97    pub fn next_batch(&mut self) -> Result<Vec<Metric>> {
98        let mut records = Vec::new();
99        let mut consumed = 0;
100        for line in self.pending.split_inclusive(|b| *b == b'\n') {
101            if !line.ends_with(b"\n") {
102                ensure!(
103                    line.len() <= MAX_LINE_BYTES,
104                    "unterminated metric line exceeds bound"
105                );
106                break;
107            }
108            records.push(parse_line(&line[..line.len() - 1])?);
109            consumed += line.len();
110            if records.len() == MAX_BATCH_RECORDS {
111                break;
112            }
113        }
114        self.pending.drain(..consumed);
115        Ok(records)
116    }
117    pub fn finish(&self) -> Result<()> {
118        ensure!(self.pending.is_empty(), "EOF before metric line terminator");
119        Ok(())
120    }
121}
netget/server/graphite/codec.rs

netget/server/graphite/
codec.rs
