netget/server/otlp/
codec.rs

1//! OTLP/HTTP payloads: decoding an export request into the summary the model sees, and encoding
2//! every response the receiver sends.
3//!
4//! Pure functions, shared by the server, the tests and nothing else. **The model never sees a
5//! payload**: it gets counts, the service name, a few resource attributes and the first few span
6//! names, metric names or log bodies, each truncated. It never writes one either: every
7//! response body — the empty full-success message, a partial success, a `google.rpc.Status` — is
8//! encoded here, in the encoding the request used.
9//!
10//! Protobuf is decoded with the OpenTelemetry project's own generated types
11//! (`opentelemetry-proto`, prost). prost stops at 100 levels of nesting, so a depth bomb in an
12//! `AnyValue` is a decode error, not a stack overflow. JSON is parsed as a `serde_json::Value`
13//! (recursion limit 128) and walked by hand rather than through the crate's serde types, because
14//! OTLP/JSON senders differ in ways a strict typed decode would refuse (64-bit integers as
15//! numbers or strings, enum values as numbers or names); keys are accepted in lowerCamelCase, as
16//! the specification writes them, and in the proto's own snake_case. Neither walk recurses into
17//! attribute values: an array or map is summarised by its length.
18
19use opentelemetry_proto::tonic::collector::logs::v1::{
20    ExportLogsPartialSuccess, ExportLogsServiceRequest, ExportLogsServiceResponse,
21};
22use opentelemetry_proto::tonic::collector::metrics::v1::{
23    ExportMetricsPartialSuccess, ExportMetricsServiceRequest, ExportMetricsServiceResponse,
24};
25use opentelemetry_proto::tonic::collector::trace::v1::{
26    ExportTracePartialSuccess, ExportTraceServiceRequest, ExportTraceServiceResponse,
27};
28use opentelemetry_proto::tonic::common::v1::{any_value, AnyValue, KeyValue};
29use opentelemetry_proto::tonic::metrics::v1::metric::Data;
30use prost::Message;
31use serde_json::{json, Map, Value};
32use std::io::Read;
33
34/// The largest request body the receiver reads, after decompression: 4 MiB, the default
35/// message limit of the gRPC servers OTLP receivers are built on and what SDK exporters batch
36/// under. A gzip body is held to the same number once inflated, so a small compressed body
37/// cannot expand past it.
38pub const MAX_BODY_BYTES: usize = 4 * 1024 * 1024;
39
40/// How much of an export the model is shown.
41pub const MAX_NAMES: usize = 10;
42pub const MAX_LOG_BODIES: usize = 5;
43pub const MAX_ATTRIBUTES: usize = 20;
44pub const MAX_SERVICES: usize = 5;
45const NAME_BYTES: usize = 100;
46const VALUE_BYTES: usize = 200;
47
48/// The three signals and their paths.
49#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum Signal {
51    Traces,
52    Metrics,
53    Logs,
54}
55
56impl Signal {
57    pub fn from_path(path: &str) -> Option<Signal> {
58        match path {
59            "/v1/traces" => Some(Signal::Traces),
60            "/v1/metrics" => Some(Signal::Metrics),
61            "/v1/logs" => Some(Signal::Logs),
62            _ => None,
63        }
64    }
65
66    pub fn as_str(self) -> &'static str {
67        match self {
68            Signal::Traces => "traces",
69            Signal::Metrics => "metrics",
70            Signal::Logs => "logs",
71        }
72    }
73
74    /// What a partial success counts as rejected: spans, data points or log records.
75    pub fn items(self) -> &'static str {
76        match self {
77            Signal::Traces => "spans",
78            Signal::Metrics => "data points",
79            Signal::Logs => "log records",
80        }
81    }
82}
83
84/// The two encodings OTLP/HTTP defines.
85#[derive(Debug, Clone, Copy, PartialEq, Eq)]
86pub enum Encoding {
87    Protobuf,
88    Json,
89}
90
91impl Encoding {
92    /// From a `Content-Type` header, parameters ignored.
93    pub fn from_content_type(value: &str) -> Option<Encoding> {
94        match value
95            .split(';')
96            .next()
97            .unwrap_or("")
98            .trim()
99            .to_ascii_lowercase()
100            .as_str()
101        {
102            "application/x-protobuf" => Some(Encoding::Protobuf),
103            "application/json" => Some(Encoding::Json),
104            _ => None,
105        }
106    }
107
108    pub fn content_type(self) -> &'static str {
109        match self {
110            Encoding::Protobuf => "application/x-protobuf",
111            Encoding::Json => "application/json",
112        }
113    }
114
115    pub fn as_str(self) -> &'static str {
116        match self {
117            Encoding::Protobuf => "protobuf",
118            Encoding::Json => "json",
119        }
120    }
121}
122
123/// Why a body could not be read.
124#[derive(Debug, Clone, PartialEq, Eq)]
125pub enum BodyError {
126    /// Inflated past [`MAX_BODY_BYTES`].
127    TooLarge,
128    /// Not valid gzip.
129    BadGzip,
130}
131
132/// Inflate a gzip body, refusing it as soon as it passes `limit` bytes. Never allocates more than
133/// `limit + 1` bytes of output.
134pub fn gunzip_bounded(body: &[u8], limit: usize) -> Result<Vec<u8>, BodyError> {
135    let mut out = Vec::new();
136    let mut decoder = flate2::read::MultiGzDecoder::new(body).take(limit as u64 + 1);
137    decoder
138        .read_to_end(&mut out)
139        .map_err(|_| BodyError::BadGzip)?;
140    if out.len() > limit {
141        return Err(BodyError::TooLarge);
142    }
143    Ok(out)
144}
145
146/// What the model is told about one export.
147#[derive(Debug, Clone, Default, PartialEq)]
148pub struct Summary {
149    pub resource_count: usize,
150    pub services: Vec<String>,
151    pub resource_attributes: Map<String, Value>,
152    /// Spans, data points or log records.
153    pub item_count: usize,
154    /// Metrics, for the metrics signal.
155    pub metric_count: usize,
156    /// Spans whose status is ERROR, or log records at severity ERROR or above.
157    pub error_count: usize,
158    /// Span names, metric names or log bodies, in order.
159    pub names: Vec<String>,
160}
161
162impl Summary {
163    fn add_resource(&mut self, attributes: Vec<(String, String)>) {
164        self.resource_count += 1;
165        for (key, value) in &attributes {
166            if key == "service.name"
167                && !self.services.contains(value)
168                && self.services.len() < MAX_SERVICES
169            {
170                self.services.push(value.clone());
171            }
172        }
173        if self.resource_count == 1 {
174            for (key, value) in attributes.into_iter().take(MAX_ATTRIBUTES) {
175                self.resource_attributes.insert(
176                    crate::utils::truncate_for_llm(&key, NAME_BYTES),
177                    json!(value),
178                );
179            }
180        }
181    }
182
183    fn add_name(&mut self, name: &str, max: usize, bytes: usize) {
184        if self.names.len() < max {
185            self.names.push(crate::utils::truncate_for_llm(
186                &crate::utils::sanitize::line_field(name),
187                bytes,
188            ));
189        }
190    }
191
192    /// The event data for `otlp_export`, without `answer_with`.
193    pub fn to_event(
194        &self,
195        signal: Signal,
196        encoding: Encoding,
197        compressed: bool,
198        body_bytes: usize,
199    ) -> Value {
200        let mut data = json!({
201            "signal": signal.as_str(),
202            "encoding": encoding.as_str(),
203            "compressed": compressed,
204            "body_bytes": body_bytes,
205            "resource_count": self.resource_count,
206            "service_name": self.services.first().cloned().unwrap_or_default(),
207            "resource_attributes": Value::Object(self.resource_attributes.clone()),
208        });
209        if self.services.len() > 1 {
210            data["service_names"] = json!(self.services);
211        }
212        match signal {
213            Signal::Traces => {
214                data["span_count"] = json!(self.item_count);
215                data["error_span_count"] = json!(self.error_count);
216                data["span_names"] = json!(self.names);
217            }
218            Signal::Metrics => {
219                data["metric_count"] = json!(self.metric_count);
220                data["data_point_count"] = json!(self.item_count);
221                data["metric_names"] = json!(self.names);
222            }
223            Signal::Logs => {
224                data["log_record_count"] = json!(self.item_count);
225                data["error_log_count"] = json!(self.error_count);
226                data["log_bodies"] = json!(self.names);
227            }
228        }
229        data
230    }
231}
232
233/// An `AnyValue` as one line of text. Arrays, maps and bytes are described, not expanded, so
234/// nothing here recurses and no raw bytes reach the model.
235fn any_value_text(value: Option<&AnyValue>) -> String {
236    match value.and_then(|v| v.value.as_ref()) {
237        None => String::new(),
238        Some(any_value::Value::StringValue(s)) => s.clone(),
239        Some(any_value::Value::BoolValue(b)) => b.to_string(),
240        Some(any_value::Value::IntValue(i)) => i.to_string(),
241        Some(any_value::Value::DoubleValue(d)) => d.to_string(),
242        Some(any_value::Value::ArrayValue(a)) => format!("<array of {}>", a.values.len()),
243        Some(any_value::Value::KvlistValue(k)) => format!("<map of {}>", k.values.len()),
244        Some(any_value::Value::BytesValue(b)) => format!("<{} bytes>", b.len()),
245    }
246}
247
248fn attributes(list: &[KeyValue]) -> Vec<(String, String)> {
249    list.iter()
250        .map(|kv| {
251            (
252                kv.key.clone(),
253                crate::utils::truncate_for_llm(
254                    &crate::utils::sanitize::line_field(&any_value_text(kv.value.as_ref())),
255                    VALUE_BYTES,
256                ),
257            )
258        })
259        .collect()
260}
261
262/// Decode a protobuf export request.
263pub fn summarize_protobuf(signal: Signal, body: &[u8]) -> Result<Summary, String> {
264    let mut summary = Summary::default();
265    match signal {
266        Signal::Traces => {
267            let request = ExportTraceServiceRequest::decode(body).map_err(|e| e.to_string())?;
268            for rs in &request.resource_spans {
269                summary.add_resource(
270                    rs.resource
271                        .as_ref()
272                        .map(|r| attributes(&r.attributes))
273                        .unwrap_or_default(),
274                );
275                for ss in &rs.scope_spans {
276                    for span in &ss.spans {
277                        summary.item_count += 1;
278                        if span.status.as_ref().is_some_and(|s| s.code == 2) {
279                            summary.error_count += 1;
280                        }
281                        summary.add_name(&span.name, MAX_NAMES, NAME_BYTES);
282                    }
283                }
284            }
285        }
286        Signal::Metrics => {
287            let request = ExportMetricsServiceRequest::decode(body).map_err(|e| e.to_string())?;
288            for rm in &request.resource_metrics {
289                summary.add_resource(
290                    rm.resource
291                        .as_ref()
292                        .map(|r| attributes(&r.attributes))
293                        .unwrap_or_default(),
294                );
295                for sm in &rm.scope_metrics {
296                    for metric in &sm.metrics {
297                        summary.metric_count += 1;
298                        summary.item_count += match &metric.data {
299                            Some(Data::Gauge(g)) => g.data_points.len(),
300                            Some(Data::Sum(s)) => s.data_points.len(),
301                            Some(Data::Histogram(h)) => h.data_points.len(),
302                            Some(Data::ExponentialHistogram(h)) => h.data_points.len(),
303                            Some(Data::Summary(s)) => s.data_points.len(),
304                            None => 0,
305                        };
306                        summary.add_name(&metric.name, MAX_NAMES, NAME_BYTES);
307                    }
308                }
309            }
310        }
311        Signal::Logs => {
312            let request = ExportLogsServiceRequest::decode(body).map_err(|e| e.to_string())?;
313            for rl in &request.resource_logs {
314                summary.add_resource(
315                    rl.resource
316                        .as_ref()
317                        .map(|r| attributes(&r.attributes))
318                        .unwrap_or_default(),
319                );
320                for sl in &rl.scope_logs {
321                    for record in &sl.log_records {
322                        summary.item_count += 1;
323                        if record.severity_number >= 17 {
324                            summary.error_count += 1;
325                        }
326                        summary.add_name(
327                            &any_value_text(record.body.as_ref()),
328                            MAX_LOG_BODIES,
329                            VALUE_BYTES,
330                        );
331                    }
332                }
333            }
334        }
335    }
336    Ok(summary)
337}
338
339/// `obj[camel]`, else `obj[snake]`.
340fn field<'a>(obj: &'a Value, camel: &str, snake: &str) -> Option<&'a Value> {
341    obj.get(camel).or_else(|| obj.get(snake))
342}
343
344/// A list field: absent or null is empty; anything but an array is an error.
345fn list<'a>(obj: &'a Value, camel: &str, snake: &str) -> Result<&'a [Value], String> {
346    match field(obj, camel, snake) {
347        None | Some(Value::Null) => Ok(&[]),
348        Some(Value::Array(a)) => Ok(a),
349        Some(_) => Err(format!("{camel} is not an array")),
350    }
351}
352
353fn object<'a>(value: &'a Value, what: &str) -> Result<&'a Value, String> {
354    if value.is_object() {
355        Ok(value)
356    } else {
357        Err(format!("{what} is not an object"))
358    }
359}
360
361/// An OTLP/JSON `AnyValue` as one line of text, without recursing.
362fn json_any_value_text(value: Option<&Value>) -> String {
363    let Some(value) = value.and_then(Value::as_object) else {
364        return String::new();
365    };
366    let scalar = |v: &Value| match v {
367        Value::String(s) => s.clone(),
368        other => other.to_string(),
369    };
370    if let Some(v) = value
371        .get("stringValue")
372        .or_else(|| value.get("string_value"))
373    {
374        return scalar(v);
375    }
376    for key in [
377        "boolValue",
378        "bool_value",
379        "intValue",
380        "int_value",
381        "doubleValue",
382        "double_value",
383    ] {
384        if let Some(v) = value.get(key) {
385            return scalar(v);
386        }
387    }
388    let len = |v: &Value| {
389        v.get("values")
390            .and_then(Value::as_array)
391            .map(Vec::len)
392            .unwrap_or(0)
393    };
394    if let Some(v) = value.get("arrayValue").or_else(|| value.get("array_value")) {
395        return format!("<array of {}>", len(v));
396    }
397    if let Some(v) = value
398        .get("kvlistValue")
399        .or_else(|| value.get("kvlist_value"))
400    {
401        return format!("<map of {}>", len(v));
402    }
403    if let Some(Value::String(b)) = value.get("bytesValue").or_else(|| value.get("bytes_value")) {
404        // Base64: three bytes per four characters.
405        return format!("<{} bytes>", b.trim_end_matches('=').len() * 3 / 4);
406    }
407    String::new()
408}
409
410fn json_attributes(resource: Option<&Value>) -> Result<Vec<(String, String)>, String> {
411    let Some(resource) = resource.filter(|r| !r.is_null()) else {
412        return Ok(Vec::new());
413    };
414    let resource = object(resource, "resource")?;
415    list(resource, "attributes", "attributes")?
416        .iter()
417        .map(|kv| {
418            let kv = object(kv, "attribute")?;
419            let key = kv
420                .get("key")
421                .and_then(Value::as_str)
422                .unwrap_or("")
423                .to_string();
424            let value = crate::utils::truncate_for_llm(
425                &crate::utils::sanitize::line_field(&json_any_value_text(kv.get("value"))),
426                VALUE_BYTES,
427            );
428            Ok((key, value))
429        })
430        .collect()
431}
432
433fn json_number(value: Option<&Value>) -> Option<i64> {
434    match value? {
435        Value::Number(n) => n.as_i64(),
436        Value::String(s) => s.parse().ok(),
437        _ => None,
438    }
439}
440
441/// Decode an OTLP/JSON export request.
442pub fn summarize_json(signal: Signal, body: &[u8]) -> Result<Summary, String> {
443    let root: Value = serde_json::from_slice(body).map_err(|e| e.to_string())?;
444    let root = object(&root, "the request")?;
445    let mut summary = Summary::default();
446    match signal {
447        Signal::Traces => {
448            for rs in list(root, "resourceSpans", "resource_spans")? {
449                let rs = object(rs, "resourceSpans entry")?;
450                summary.add_resource(json_attributes(rs.get("resource"))?);
451                for ss in list(rs, "scopeSpans", "scope_spans")? {
452                    for span in list(object(ss, "scopeSpans entry")?, "spans", "spans")? {
453                        let span = object(span, "span")?;
454                        summary.item_count += 1;
455                        let code = span.get("status").and_then(|s| s.get("code"));
456                        if json_number(code) == Some(2)
457                            || code.and_then(Value::as_str) == Some("STATUS_CODE_ERROR")
458                        {
459                            summary.error_count += 1;
460                        }
461                        let name = span.get("name").and_then(Value::as_str).unwrap_or("");
462                        summary.add_name(name, MAX_NAMES, NAME_BYTES);
463                    }
464                }
465            }
466        }
467        Signal::Metrics => {
468            for rm in list(root, "resourceMetrics", "resource_metrics")? {
469                let rm = object(rm, "resourceMetrics entry")?;
470                summary.add_resource(json_attributes(rm.get("resource"))?);
471                for sm in list(rm, "scopeMetrics", "scope_metrics")? {
472                    for metric in list(object(sm, "scopeMetrics entry")?, "metrics", "metrics")? {
473                        let metric = object(metric, "metric")?;
474                        summary.metric_count += 1;
475                        for (camel, snake) in [
476                            ("gauge", "gauge"),
477                            ("sum", "sum"),
478                            ("histogram", "histogram"),
479                            ("exponentialHistogram", "exponential_histogram"),
480                            ("summary", "summary"),
481                        ] {
482                            if let Some(data) = field(metric, camel, snake).filter(|d| !d.is_null())
483                            {
484                                summary.item_count +=
485                                    list(object(data, camel)?, "dataPoints", "data_points")?.len();
486                            }
487                        }
488                        let name = metric.get("name").and_then(Value::as_str).unwrap_or("");
489                        summary.add_name(name, MAX_NAMES, NAME_BYTES);
490                    }
491                }
492            }
493        }
494        Signal::Logs => {
495            for rl in list(root, "resourceLogs", "resource_logs")? {
496                let rl = object(rl, "resourceLogs entry")?;
497                summary.add_resource(json_attributes(rl.get("resource"))?);
498                for sl in list(rl, "scopeLogs", "scope_logs")? {
499                    for record in list(object(sl, "scopeLogs entry")?, "logRecords", "log_records")?
500                    {
501                        let record = object(record, "log record")?;
502                        summary.item_count += 1;
503                        if json_number(field(record, "severityNumber", "severity_number"))
504                            .is_some_and(|n| n >= 17)
505                        {
506                            summary.error_count += 1;
507                        }
508                        summary.add_name(
509                            &json_any_value_text(record.get("body")),
510                            MAX_LOG_BODIES,
511                            VALUE_BYTES,
512                        );
513                    }
514                }
515            }
516        }
517    }
518    Ok(summary)
519}
520
521/// Decode a request in either encoding.
522pub fn summarize(signal: Signal, encoding: Encoding, body: &[u8]) -> Result<Summary, String> {
523    match encoding {
524        Encoding::Protobuf => summarize_protobuf(signal, body),
525        Encoding::Json => summarize_json(signal, body),
526    }
527}
528
529/// The export response: full success when `partial` is `None`, else a partial success naming
530/// how many items were rejected and why.
531pub fn export_response(
532    signal: Signal,
533    encoding: Encoding,
534    partial: Option<(i64, &str)>,
535) -> Vec<u8> {
536    match encoding {
537        Encoding::Protobuf => match signal {
538            Signal::Traces => ExportTraceServiceResponse {
539                partial_success: partial.map(|(n, m)| ExportTracePartialSuccess {
540                    rejected_spans: n,
541                    error_message: m.to_string(),
542                }),
543            }
544            .encode_to_vec(),
545            Signal::Metrics => ExportMetricsServiceResponse {
546                partial_success: partial.map(|(n, m)| ExportMetricsPartialSuccess {
547                    rejected_data_points: n,
548                    error_message: m.to_string(),
549                }),
550            }
551            .encode_to_vec(),
552            Signal::Logs => ExportLogsServiceResponse {
553                partial_success: partial.map(|(n, m)| ExportLogsPartialSuccess {
554                    rejected_log_records: n,
555                    error_message: m.to_string(),
556                }),
557            }
558            .encode_to_vec(),
559        },
560        Encoding::Json => {
561            let body = match partial {
562                None => json!({}),
563                Some((n, message)) => {
564                    let key = match signal {
565                        Signal::Traces => "rejectedSpans",
566                        Signal::Metrics => "rejectedDataPoints",
567                        Signal::Logs => "rejectedLogRecords",
568                    };
569                    // int64 is a string in protobuf's JSON mapping.
570                    json!({"partialSuccess": {key: n.to_string(), "errorMessage": message}})
571                }
572            };
573            body.to_string().into_bytes()
574        }
575    }
576}
577
578/// `google.rpc.Status`, the body of every OTLP/HTTP failure response.
579#[derive(Clone, PartialEq, prost::Message)]
580pub struct RpcStatus {
581    #[prost(int32, tag = "1")]
582    pub code: i32,
583    #[prost(string, tag = "2")]
584    pub message: String,
585}
586
587/// The gRPC code an HTTP failure status corresponds to, per the OTLP specification's mapping.
588pub fn grpc_code(http_status: u16) -> i32 {
589    match http_status {
590        400 => 3,        // INVALID_ARGUMENT
591        401 => 16,       // UNAUTHENTICATED
592        403 => 7,        // PERMISSION_DENIED
593        404 => 5,        // NOT_FOUND
594        413 | 429 => 8,  // RESOURCE_EXHAUSTED
595        502 | 503 => 14, // UNAVAILABLE
596        504 => 4,        // DEADLINE_EXCEEDED
597        _ => 13,         // INTERNAL
598    }
599}
600
601/// The statuses a model may refuse with. 429, 502, 503 and 504 are the ones OTLP clients retry.
602pub const REJECT_STATUSES: &[u16] = &[400, 401, 403, 413, 429, 500, 502, 503, 504];
603
604/// Whether an OTLP client retries after this status.
605pub fn retryable(http_status: u16) -> bool {
606    matches!(http_status, 429 | 502 | 503 | 504)
607}
608
609/// A failure body in the request's encoding.
610pub fn status_body(encoding: Encoding, http_status: u16, message: &str) -> Vec<u8> {
611    let code = grpc_code(http_status);
612    match encoding {
613        Encoding::Protobuf => RpcStatus {
614            code,
615            message: message.to_string(),
616        }
617        .encode_to_vec(),
618        Encoding::Json => json!({"code": code, "message": message})
619            .to_string()
620            .into_bytes(),
621    }
622}
netget/server/otlp/codec.rs

netget/server/otlp/
codec.rs
