netget/server/influxdb/
codec.rs

1//! Typed InfluxDB v2 write line protocol. No time-series storage or schema state.
2use anyhow::{bail, ensure, Context, Result};
3use serde::{Deserialize, Serialize};
4use std::{
5    collections::BTreeMap,
6    io::{Read, Write},
7};
8
9pub const MAX_BODY_BYTES: usize = 256 * 1024;
10pub const MAX_LINE_BYTES: usize = 16 * 1024;
11pub const MAX_POINTS: usize = 256;
12pub const MAX_LINES: usize = 1024;
13pub const MAX_TAGS: usize = 64;
14pub const MAX_FIELDS: usize = 64;
15pub const MAX_NAME_BYTES: usize = 1024;
16pub const MIN_TIMESTAMP_NS: i64 = i64::MIN + 2;
17pub const MAX_TIMESTAMP_NS: i64 = i64::MAX - 1;
18pub const DEFAULT_LLM_FALLBACK: bool = false;
19
20#[derive(Clone, Copy, Debug, Default, PartialEq, Serialize, Deserialize)]
21pub enum Precision {
22    #[default]
23    #[serde(rename = "ns")]
24    Nanoseconds,
25    #[serde(rename = "us")]
26    Microseconds,
27    #[serde(rename = "ms")]
28    Milliseconds,
29    #[serde(rename = "s")]
30    Seconds,
31}
32impl Precision {
33    pub fn parse(s: &str) -> Result<Self> {
34        Ok(match s {
35            "ns" => Self::Nanoseconds,
36            "us" => Self::Microseconds,
37            "ms" => Self::Milliseconds,
38            "s" => Self::Seconds,
39            _ => bail!("precision must be ns, us, ms or s"),
40        })
41    }
42    pub fn as_str(self) -> &'static str {
43        match self {
44            Self::Nanoseconds => "ns",
45            Self::Microseconds => "us",
46            Self::Milliseconds => "ms",
47            Self::Seconds => "s",
48        }
49    }
50    pub fn to_nanoseconds(self, timestamp: i64) -> Result<i64> {
51        let factor = match self {
52            Self::Nanoseconds => 1,
53            Self::Microseconds => 1000,
54            Self::Milliseconds => 1_000_000,
55            Self::Seconds => 1_000_000_000,
56        };
57        let ns = timestamp
58            .checked_mul(factor)
59            .context("timestamp precision overflow")?;
60        ensure!(
61            (MIN_TIMESTAMP_NS..=MAX_TIMESTAMP_NS).contains(&ns),
62            "timestamp outside InfluxDB v2 range"
63        );
64        Ok(ns)
65    }
66}
67#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
68#[serde(tag = "type", content = "value", rename_all = "snake_case")]
69pub enum FieldValue {
70    Float(f64),
71    Integer(i64),
72    Unsigned(u64),
73    Boolean(bool),
74    String(String),
75}
76#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
77#[serde(deny_unknown_fields)]
78pub struct Point {
79    pub measurement: String,
80    #[serde(default)]
81    pub tags: BTreeMap<String, String>,
82    pub fields: BTreeMap<String, FieldValue>,
83    #[serde(default, skip_serializing_if = "Option::is_none")]
84    pub timestamp: Option<i64>,
85}
86#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
87#[serde(deny_unknown_fields)]
88pub struct WriteBatch {
89    pub org: String,
90    pub bucket: String,
91    #[serde(default)]
92    pub precision: Precision,
93    pub points: Vec<Point>,
94    #[serde(default)]
95    pub gzip: bool,
96}
97#[derive(Clone, Debug, PartialEq, Serialize)]
98pub struct ObservedPoint {
99    pub line: usize,
100    #[serde(flatten)]
101    pub point: Point,
102    pub timestamp_ns: i64,
103}
104#[derive(Clone, Debug, PartialEq, Serialize)]
105pub struct LineError {
106    pub line: usize,
107    pub message: String,
108}
109#[derive(Clone, Debug, PartialEq, Serialize)]
110pub struct ParsedBatch {
111    pub points: Vec<ObservedPoint>,
112    pub errors: Vec<LineError>,
113}
114
115pub fn validate_token(value: &str) -> Result<()> {
116    ensure!(
117        !value.is_empty()
118            && value.len() <= MAX_NAME_BYTES
119            && value.bytes().all(|b| b.is_ascii_graphic()),
120        "token must be nonempty printable ASCII, <=1024 bytes"
121    );
122    Ok(())
123}
124pub fn validate_target(value: &str) -> Result<()> {
125    ensure!(
126        !value.is_empty()
127            && value.len() <= MAX_NAME_BYTES
128            && !crate::utils::sanitize::has_controls(&value),
129        "org/bucket must be nonempty, <=1024 bytes, without controls"
130    );
131    Ok(())
132}
133fn name(value: &str, namespace: bool) -> Result<()> {
134    ensure!(
135        !value.is_empty()
136            && value.len() <= MAX_NAME_BYTES
137            && !value.bytes().any(|b| b < 32 || b == 127)
138            && !value.ends_with('\\'),
139        "empty, oversized, control or trailing-backslash name"
140    );
141    if namespace {
142        ensure!(
143            !value.starts_with('_'),
144            "reserved InfluxDB underscore namespace"
145        );
146    }
147    Ok(())
148}
149fn validate_point(p: &Point, precision: Precision) -> Result<()> {
150    name(&p.measurement, true)?;
151    ensure!(
152        !p.measurement.starts_with('#'),
153        "measurement cannot begin comment marker"
154    );
155    ensure!(p.tags.len() <= MAX_TAGS, "tag count limit");
156    ensure!(
157        !p.fields.is_empty() && p.fields.len() <= MAX_FIELDS,
158        "point requires 1..64 fields"
159    );
160    for (key, value) in &p.tags {
161        name(key, true)?;
162        ensure!(key != "time", "reserved time tag key");
163        name(value, false)?;
164    }
165    for (key, value) in &p.fields {
166        name(key, true)?;
167        ensure!(key != "time", "reserved time field key");
168        match value {
169            FieldValue::Float(n) => ensure!(n.is_finite(), "nonfinite float field"),
170            FieldValue::String(s) => ensure!(
171                s.len() <= MAX_LINE_BYTES && !s.contains(['\r', '\n', '\0']),
172                "string field byte/newline/control limit"
173            ),
174            _ => {}
175        }
176    }
177    if let Some(ts) = p.timestamp {
178        precision.to_nanoseconds(ts)?;
179    }
180    Ok(())
181}
182fn escaped(value: &str, special: &[u8]) -> String {
183    let mut out = String::new();
184    for c in value.chars() {
185        if c.is_ascii() && special.contains(&(c as u8)) {
186            out.push('\\');
187        }
188        out.push(c);
189    }
190    out
191}
192fn encode_point(p: &Point, precision: Precision) -> Result<String> {
193    validate_point(p, precision)?;
194    let mut out = escaped(&p.measurement, b", ");
195    for (key, value) in &p.tags {
196        out.push(',');
197        out.push_str(&escaped(key, b",= "));
198        out.push('=');
199        out.push_str(&escaped(value, b",= "));
200    }
201    out.push(' ');
202    for (index, (key, value)) in p.fields.iter().enumerate() {
203        if index > 0 {
204            out.push(',');
205        }
206        out.push_str(&escaped(key, b",= "));
207        out.push('=');
208        match value {
209            FieldValue::Float(n) => out.push_str(&n.to_string()),
210            FieldValue::Integer(n) => {
211                out.push_str(&n.to_string());
212                out.push('i');
213            }
214            FieldValue::Unsigned(n) => {
215                out.push_str(&n.to_string());
216                out.push('u');
217            }
218            FieldValue::Boolean(b) => out.push_str(if *b { "true" } else { "false" }),
219            FieldValue::String(s) => {
220                out.push('"');
221                out.push_str(&escaped(s, b"\\\""));
222                out.push('"');
223            }
224        }
225        ensure!(out.len() <= MAX_LINE_BYTES, "encoded line byte limit");
226    }
227    if let Some(ts) = p.timestamp {
228        out.push(' ');
229        out.push_str(&ts.to_string());
230    }
231    ensure!(out.len() <= MAX_LINE_BYTES, "encoded line byte limit");
232    Ok(out)
233}
234pub fn encode_batch(batch: &WriteBatch) -> Result<Vec<u8>> {
235    validate_target(&batch.org)?;
236    validate_target(&batch.bucket)?;
237    ensure!(
238        !batch.points.is_empty() && batch.points.len() <= MAX_POINTS,
239        "batch requires 1..256 points"
240    );
241    let mut out = Vec::new();
242    for p in &batch.points {
243        let line = encode_point(p, batch.precision)?;
244        ensure!(
245            out.len() + line.len() < MAX_BODY_BYTES,
246            "encoded body byte limit"
247        );
248        out.extend_from_slice(line.as_bytes());
249        out.push(b'\n');
250    }
251    if batch.gzip {
252        let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
253        e.write_all(&out)?;
254        out = e.finish()?;
255        ensure!(out.len() <= MAX_BODY_BYTES, "compressed body byte limit");
256    }
257    Ok(out)
258}
259pub fn decode_body(bytes: &[u8], encoding: &str) -> Result<Vec<u8>> {
260    ensure!(bytes.len() <= MAX_BODY_BYTES, "body byte limit");
261    match encoding {
262        "" | "identity" => Ok(bytes.to_vec()),
263        "gzip" => {
264            let mut d = flate2::bufread::MultiGzDecoder::new(bytes);
265            let mut out = Vec::new();
266            d.by_ref()
267                .take((MAX_BODY_BYTES + 1) as u64)
268                .read_to_end(&mut out)
269                .context("malformed gzip body")?;
270            ensure!(
271                out.len() <= MAX_BODY_BYTES && d.get_ref().is_empty(),
272                "decompressed body byte limit or trailing bytes"
273            );
274            Ok(out)
275        }
276        _ => bail!("unsupported content encoding"),
277    }
278}
279fn token(bytes: &[u8], at: &mut usize, delimiters: &[u8], escapes: &[u8]) -> Result<String> {
280    let mut out = Vec::new();
281    while *at < bytes.len() {
282        let b = bytes[*at];
283        if b == b'\\'
284            && bytes
285                .get(*at + 1)
286                .is_some_and(|next| escapes.contains(next))
287        {
288            out.push(bytes[*at + 1]);
289            *at += 2;
290        } else if delimiters.contains(&b) {
291            break;
292        } else {
293            out.push(b);
294            *at += 1;
295        }
296    }
297    Ok(String::from_utf8(out)?)
298}
299fn expect(bytes: &[u8], at: &mut usize, b: u8) -> Result<()> {
300    ensure!(bytes.get(*at) == Some(&b), "expected '{}'", b as char);
301    *at += 1;
302    Ok(())
303}
304fn field(bytes: &[u8], at: &mut usize) -> Result<FieldValue> {
305    if bytes.get(*at) == Some(&b'"') {
306        *at += 1;
307        let mut string = Vec::new();
308        while *at < bytes.len() && bytes[*at] != b'"' {
309            let b = bytes[*at];
310            let replacement = if b == b'\\' {
311                bytes.get(*at + 1).and_then(|c| match c {
312                    b'\\' => Some(b'\\'),
313                    b'"' => Some(b'"'),
314                    b'n' => Some(b'\n'),
315                    b'r' => Some(b'\r'),
316                    b't' => Some(b'\t'),
317                    _ => None,
318                })
319            } else {
320                None
321            };
322            if let Some(c) = replacement {
323                string.push(c);
324                *at += 2;
325            } else {
326                string.push(b);
327                *at += 1;
328            }
329        }
330        let s = String::from_utf8(string)?;
331        expect(bytes, at, b'"')?;
332        ensure!(
333            *at == bytes.len() || matches!(bytes[*at], b',' | b' '),
334            "unexpected characters after string field"
335        );
336        return Ok(FieldValue::String(s));
337    }
338    let s = token(bytes, at, b", ", b"")?;
339    ensure!(!s.is_empty(), "empty field value");
340    Ok(match s.as_str() {
341        "t" | "T" | "true" | "True" | "TRUE" => FieldValue::Boolean(true),
342        "f" | "F" | "false" | "False" | "FALSE" => FieldValue::Boolean(false),
343        _ if s.ends_with('i') => {
344            FieldValue::Integer(s[..s.len() - 1].parse().context("invalid int64 field")?)
345        }
346        _ if s.ends_with('u') => {
347            FieldValue::Unsigned(s[..s.len() - 1].parse().context("invalid uint64 field")?)
348        }
349        _ => {
350            let n: f64 = s.parse().context("invalid float field")?;
351            ensure!(n.is_finite(), "nonfinite float field");
352            FieldValue::Float(n)
353        }
354    })
355}
356fn parse_point(line: &str, precision: Precision) -> Result<Point> {
357    ensure!(line.len() <= MAX_LINE_BYTES, "line byte limit");
358    let bytes = line.as_bytes();
359    let mut at = 0;
360    let measurement = token(bytes, &mut at, b", ", b", ")?;
361    let mut tags = BTreeMap::new();
362    while bytes.get(at) == Some(&b',') {
363        at += 1;
364        ensure!(tags.len() < MAX_TAGS, "tag count limit");
365        let key = token(bytes, &mut at, b"=, ", b",= ")?;
366        expect(bytes, &mut at, b'=')?;
367        let value = token(bytes, &mut at, b",= ", b",= ")?;
368        ensure!(tags.insert(key, value).is_none(), "duplicate tag key");
369    }
370    expect(bytes, &mut at, b' ')?;
371    let mut fields = BTreeMap::new();
372    loop {
373        ensure!(fields.len() < MAX_FIELDS, "field count limit");
374        let key = token(bytes, &mut at, b"=, ", b",= ")?;
375        expect(bytes, &mut at, b'=')?;
376        let value = field(bytes, &mut at)?;
377        ensure!(fields.insert(key, value).is_none(), "duplicate field key");
378        if bytes.get(at) != Some(&b',') {
379            break;
380        }
381        at += 1;
382    }
383    let timestamp = if at == bytes.len() {
384        None
385    } else {
386        expect(bytes, &mut at, b' ')?;
387        let ts = &line[at..];
388        ensure!(
389            !ts.is_empty()
390                && ts
391                    .strip_prefix('-')
392                    .unwrap_or(ts)
393                    .bytes()
394                    .all(|b| b.is_ascii_digit()),
395            "invalid timestamp syntax"
396        );
397        Some(ts.parse().context("timestamp int64 overflow")?)
398    };
399    let point = Point {
400        measurement,
401        tags,
402        fields,
403        timestamp,
404    };
405    validate_point(&point, precision)?;
406    Ok(point)
407}
408pub fn parse_batch(bytes: &[u8], precision: Precision, received_ns: i64) -> Result<ParsedBatch> {
409    ensure!(bytes.len() <= MAX_BODY_BYTES, "body byte limit");
410    ensure!(
411        (MIN_TIMESTAMP_NS..=MAX_TIMESTAMP_NS).contains(&received_ns),
412        "receiver timestamp outside range"
413    );
414    let body = std::str::from_utf8(bytes).context("line protocol must be UTF-8")?;
415    let mut out = ParsedBatch {
416        points: Vec::new(),
417        errors: Vec::new(),
418    };
419    for (index, line) in body.lines().enumerate() {
420        ensure!(index < MAX_LINES, "line count limit");
421        if line.is_empty() || line.starts_with('#') {
422            continue;
423        }
424        ensure!(
425            out.points.len() + out.errors.len() < MAX_POINTS,
426            "point count limit"
427        );
428        match parse_point(line, precision) {
429            Ok(point) => {
430                let timestamp_ns = point
431                    .timestamp
432                    .map(|ts| precision.to_nanoseconds(ts))
433                    .transpose()?
434                    .unwrap_or(received_ns);
435                out.points.push(ObservedPoint {
436                    line: index + 1,
437                    point,
438                    timestamp_ns,
439                });
440            }
441            Err(error) => out.errors.push(LineError {
442                line: index + 1,
443                message: error.to_string(),
444            }),
445        }
446    }
447    Ok(out)
448}
netget/server/influxdb/codec.rs

netget/server/influxdb/
codec.rs
