netget/server/zipkin/
wire.rs

1//! Zipkin v2 JSON spans: validation and normalisation shared by the collector and the
2//! reporter, plus the query API's endpoint table and the shape each endpoint answers with.
3//!
4//! The rules follow what the official Zipkin server accepts, measured against
5//! zipkin-server 3.5.1: ids are lower-case hex without a prefix, left-padded to 16 (or, for a
6//! trace id longer than 16, to 32) characters, never all zero; `kind` is one of four names;
7//! unknown keys are ignored; a scalar tag value is kept as its text; an unparseable endpoint
8//! address and a non-positive timestamp or duration are dropped rather than refused.
9use anyhow::{bail, ensure, Context, Result};
10use serde_json::{json, Map, Value};
11use std::io::Read;
12
13/// Largest request or response body, after decompression as well as before.
14pub const MAX_BODY_BYTES: usize = 1024 * 1024;
15/// Most spans in one report or one query answer.
16pub const MAX_SPANS: usize = 1000;
17/// Most annotations, and most tags, on one span.
18pub const MAX_ANNOTATIONS: usize = 256;
19pub const MAX_TAGS: usize = 256;
20/// Longest name, service name, tag key or value, or annotation value.
21pub const MAX_TEXT: usize = 64 * 1024;
22
23pub const KINDS: [&str; 4] = ["CLIENT", "SERVER", "PRODUCER", "CONSUMER"];
24
25fn hex_id(value: &Value, field: &str, max: usize) -> Result<String> {
26    let s = value
27        .as_str()
28        .with_context(|| format!("{field} must be a string"))?;
29    ensure!(
30        !s.is_empty() && s.len() <= max,
31        "{field} must be 1..={max} hex characters"
32    );
33    ensure!(
34        s.bytes()
35            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)),
36        "{field} should be lower-hex encoded with no prefix"
37    );
38    ensure!(s.bytes().any(|b| b != b'0'), "{field} must not be zero");
39    let width = if s.len() > 16 { 32 } else { 16 };
40    Ok(format!("{s:0>width$}"))
41}
42
43/// A trace id as Zipkin stores it: 16 or 32 lower-case hex characters.
44pub fn trace_id(value: &Value) -> Result<String> {
45    hex_id(value, "traceId", 32)
46}
47
48fn text(value: &Value, field: &str) -> Result<Option<String>> {
49    match value {
50        Value::Null => Ok(None),
51        Value::String(s) => {
52            ensure!(s.len() <= MAX_TEXT, "{field} exceeds {MAX_TEXT} bytes");
53            Ok((!s.is_empty()).then(|| s.clone()))
54        }
55        _ => bail!("{field} must be a string"),
56    }
57}
58
59fn micros(value: &Value) -> Option<u64> {
60    value.as_u64().filter(|n| *n > 0)
61}
62
63fn endpoint(value: &Value, field: &str) -> Result<Option<Value>> {
64    let Some(map) = value.as_object() else {
65        ensure!(value.is_null(), "{field} must be an object");
66        return Ok(None);
67    };
68    let mut out = Map::new();
69    if let Some(name) = text(
70        map.get("serviceName").unwrap_or(&Value::Null),
71        "serviceName",
72    )? {
73        out.insert("serviceName".into(), json!(name));
74    }
75    if let Some(ip) = map
76        .get("ipv4")
77        .and_then(Value::as_str)
78        .and_then(|s| s.parse::<std::net::Ipv4Addr>().ok())
79    {
80        out.insert("ipv4".into(), json!(ip.to_string()));
81    }
82    if let Some(ip) = map
83        .get("ipv6")
84        .and_then(Value::as_str)
85        .and_then(|s| s.parse::<std::net::Ipv6Addr>().ok())
86    {
87        out.insert("ipv6".into(), json!(ip.to_string()));
88    }
89    if let Some(port) = map
90        .get("port")
91        .and_then(Value::as_u64)
92        .filter(|p| (1..=65535).contains(p))
93    {
94        out.insert("port".into(), json!(port));
95    }
96    Ok((!out.is_empty()).then_some(Value::Object(out)))
97}
98
99/// Validate one v2 span and return it in canonical form, keys in Zipkin's own order.
100pub fn span(value: &Value) -> Result<Value> {
101    let map = value.as_object().context("a span must be a JSON object")?;
102    let mut out = Map::new();
103    out.insert(
104        "traceId".into(),
105        json!(trace_id(map.get("traceId").context("traceId required")?)?),
106    );
107    if let Some(parent) = map.get("parentId").filter(|v| !v.is_null()) {
108        out.insert("parentId".into(), json!(hex_id(parent, "parentId", 16)?));
109    }
110    out.insert(
111        "id".into(),
112        json!(hex_id(map.get("id").context("id required")?, "id", 16)?),
113    );
114    if let Some(kind) = map.get("kind").filter(|v| !v.is_null()) {
115        let kind = kind.as_str().context("kind must be a string")?;
116        ensure!(
117            KINDS.contains(&kind),
118            "kind must be CLIENT, SERVER, PRODUCER or CONSUMER"
119        );
120        out.insert("kind".into(), json!(kind));
121    }
122    if let Some(name) = text(map.get("name").unwrap_or(&Value::Null), "name")? {
123        out.insert("name".into(), json!(name));
124    }
125    for field in ["timestamp", "duration"] {
126        if let Some(n) = map.get(field).and_then(micros) {
127            out.insert(field.into(), json!(n));
128        }
129    }
130    for field in ["localEndpoint", "remoteEndpoint"] {
131        if let Some(e) = endpoint(map.get(field).unwrap_or(&Value::Null), field)? {
132            out.insert(field.into(), e);
133        }
134    }
135    if let Some(list) = map.get("annotations").filter(|v| !v.is_null()) {
136        let list = list.as_array().context("annotations must be an array")?;
137        ensure!(
138            list.len() <= MAX_ANNOTATIONS,
139            "more than {MAX_ANNOTATIONS} annotations"
140        );
141        let mut annotations = Vec::new();
142        for a in list {
143            let timestamp = a["timestamp"]
144                .as_u64()
145                .context("annotation timestamp must be a positive integer")?;
146            let value = text(&a["value"], "annotation value")?
147                .context("annotation value must not be empty")?;
148            annotations.push(json!({"timestamp": timestamp, "value": value}));
149        }
150        if !annotations.is_empty() {
151            out.insert("annotations".into(), Value::Array(annotations));
152        }
153    }
154    if let Some(tags) = map.get("tags").filter(|v| !v.is_null()) {
155        let tags = tags.as_object().context("tags must be an object")?;
156        ensure!(tags.len() <= MAX_TAGS, "more than {MAX_TAGS} tags");
157        let mut kept = Map::new();
158        for (k, v) in tags {
159            ensure!(
160                !k.is_empty() && k.len() <= MAX_TEXT,
161                "tag keys must be 1..={MAX_TEXT} bytes"
162            );
163            let v = match v {
164                Value::String(s) => s.clone(),
165                Value::Number(n) => n.to_string(),
166                Value::Bool(b) => b.to_string(),
167                _ => bail!("tag {k} must be a string"),
168            };
169            ensure!(v.len() <= MAX_TEXT, "tag {k} exceeds {MAX_TEXT} bytes");
170            kept.insert(k.clone(), json!(v));
171        }
172        if !kept.is_empty() {
173            out.insert("tags".into(), Value::Object(kept));
174        }
175    }
176    for field in ["debug", "shared"] {
177        if map.get(field).and_then(Value::as_bool) == Some(true) {
178            out.insert(field.into(), json!(true));
179        }
180    }
181    Ok(Value::Object(out))
182}
183
184/// Validate a list of spans: at most `MAX_SPANS`, each canonicalised.
185pub fn spans(value: &Value) -> Result<Vec<Value>> {
186    let list = value.as_array().context("spans must be a JSON array")?;
187    ensure!(list.len() <= MAX_SPANS, "more than {MAX_SPANS} spans");
188    list.iter()
189        .enumerate()
190        .map(|(i, s)| span(s).with_context(|| format!("span {i}")))
191        .collect()
192}
193
194/// Decode a request body by its `Content-Encoding`, bounded after decompression too.
195pub fn decode_body(bytes: &[u8], encoding: &str) -> Result<Vec<u8>> {
196    match encoding {
197        "identity" => Ok(bytes.to_vec()),
198        "gzip" => {
199            let mut out = Vec::new();
200            flate2::read::GzDecoder::new(bytes)
201                .take(MAX_BODY_BYTES as u64 + 1)
202                .read_to_end(&mut out)
203                .context("malformed gzip body")?;
204            ensure!(
205                out.len() <= MAX_BODY_BYTES,
206                "decompressed body exceeds byte limit"
207            );
208            Ok(out)
209        }
210        _ => bail!("unsupported content encoding"),
211    }
212}
213
214pub fn gzip(bytes: &[u8]) -> Result<Vec<u8>> {
215    use std::io::Write;
216    let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
217    encoder.write_all(bytes)?;
218    Ok(encoder.finish()?)
219}
220
221/// What a query endpoint answers with.
222#[derive(Clone, Copy, Debug, PartialEq, Eq)]
223pub enum Shape {
224    /// `["frontend", "backend"]`
225    Names,
226    /// `[span, ...]`, one trace
227    Trace,
228    /// `[[span, ...], ...]`
229    Traces,
230    /// `[{"parent": "a", "child": "b", "callCount": 1}, ...]`
231    Links,
232}
233
234/// The read API's endpoints, the query parameters each takes and the shape it answers with.
235/// `trace` takes its id in the path, `/api/v2/trace/{traceId}`.
236pub const ENDPOINTS: [(&str, &[&str], Shape); 9] = [
237    ("services", &[], Shape::Names),
238    ("spans", &["serviceName"], Shape::Names),
239    ("remoteServices", &["serviceName"], Shape::Names),
240    (
241        "traces",
242        &[
243            "serviceName",
244            "remoteServiceName",
245            "spanName",
246            "annotationQuery",
247            "minDuration",
248            "maxDuration",
249            "endTs",
250            "lookback",
251            "limit",
252        ],
253        Shape::Traces,
254    ),
255    ("trace", &[], Shape::Trace),
256    ("traceMany", &["traceIds"], Shape::Traces),
257    ("dependencies", &["endTs", "lookback"], Shape::Links),
258    ("autocompleteKeys", &[], Shape::Names),
259    ("autocompleteValues", &["key"], Shape::Names),
260];
261
262pub fn endpoint_shape(name: &str) -> Option<(&'static [&'static str], Shape)> {
263    ENDPOINTS
264        .iter()
265        .find(|(n, _, _)| *n == name)
266        .map(|(_, params, shape)| (*params, *shape))
267}
268
269/// Check (and canonicalise) a query answer against the shape its endpoint promises.
270pub fn result(shape: Shape, value: &Value) -> Result<Value> {
271    let list = value
272        .as_array()
273        .context("a query result must be an array")?;
274    match shape {
275        Shape::Names => {
276            ensure!(list.len() <= MAX_SPANS, "more than {MAX_SPANS} names");
277            for n in list {
278                let n = n.as_str().context("names must be strings")?;
279                ensure!(n.len() <= MAX_TEXT, "name exceeds {MAX_TEXT} bytes");
280            }
281            Ok(value.clone())
282        }
283        Shape::Trace => Ok(Value::Array(spans(value)?)),
284        Shape::Traces => {
285            let total: usize = list.iter().map(|t| t.as_array().map_or(0, Vec::len)).sum();
286            ensure!(total <= MAX_SPANS, "more than {MAX_SPANS} spans in all");
287            Ok(Value::Array(
288                list.iter()
289                    .map(|t| spans(t).map(Value::Array))
290                    .collect::<Result<_>>()?,
291            ))
292        }
293        Shape::Links => {
294            ensure!(list.len() <= MAX_SPANS, "more than {MAX_SPANS} links");
295            let mut out = Vec::new();
296            for l in list {
297                let parent = l["parent"].as_str().context("link parent required")?;
298                let child = l["child"].as_str().context("link child required")?;
299                let calls = l["callCount"].as_u64().unwrap_or(0);
300                let mut link = json!({"parent": parent, "child": child, "callCount": calls});
301                if let Some(errors) = l["errorCount"].as_u64().filter(|n| *n > 0) {
302                    link["errorCount"] = json!(errors);
303                }
304                out.push(link);
305            }
306            Ok(Value::Array(out))
307        }
308    }
309}
310
311/// Decode one `application/x-www-form-urlencoded` component.
312pub fn unescape(s: &str) -> Result<String> {
313    let b = s.as_bytes();
314    let mut out = Vec::with_capacity(b.len());
315    let mut at = 0;
316    while at < b.len() {
317        match b[at] {
318            b'+' => {
319                out.push(b' ');
320                at += 1;
321            }
322            b'%' => {
323                let pair = b.get(at + 1..at + 3).context("truncated escape")?;
324                out.push(u8::from_str_radix(std::str::from_utf8(pair)?, 16)?);
325                at += 3;
326            }
327            c => {
328                out.push(c);
329                at += 1;
330            }
331        }
332    }
333    Ok(String::from_utf8(out)?)
334}
335
336pub fn escape(value: &str) -> String {
337    let mut out = String::new();
338    for b in value.bytes() {
339        if b.is_ascii_alphanumeric() || b"-._~".contains(&b) {
340            out.push(b as char);
341        } else {
342            out.push_str(&format!("%{b:02X}"));
343        }
344    }
345    out
346}
netget/server/zipkin/wire.rs

netget/server/zipkin/
wire.rs
