diff --git a/Cargo.lock b/Cargo.lock index d8b2f36221..41cbe41991 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1233,6 +1233,18 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "const-hex" +version = "1.19.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e59eef12462b0f9b0a3620219be5d639afd79fe39dff0a42c3997061f9298b4" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "proptest", + "serde_core", +] + [[package]] name = "const-oid" version = "0.9.6" @@ -1300,6 +1312,12 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "core_detect" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f8f80099a98041a3d1622845c271458a2d73e688351bf3cb999266764b81d48" + [[package]] name = "countme" version = "3.0.1" @@ -5543,9 +5561,12 @@ version = "0.33.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d576f5dddc3299341bcd85a04be17b2564844c0c9563214f02d85f2524d1afae" dependencies = [ + "base64 0.22.1", + "const-hex", "opentelemetry", "opentelemetry_sdk", "prost", + "serde", "tonic", "tonic-prost", ] @@ -6158,6 +6179,22 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "proptest" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8530004ccb15eae51c7e40009fbe317f341f804db54dc033eec1c50be28cfa0" +dependencies = [ + "bitflags 2.13.2", + "chacha20", + "core_detect", + "num-traits", + "rand 0.10.2", + "rand_xorshift", + "regex-syntax", + "unarray", +] + [[package]] name = "prost" version = "0.14.3" @@ -6486,6 +6523,15 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rand_xorshift" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "60aa6af80be32871323012e02e6e65f8a7cc7890931ae421d217ad8fe0df2ccf" +dependencies = [ + "rand_core 0.10.1", +] + [[package]] name = "rand_xoshiro" version = "0.7.0" @@ -8817,6 +8863,12 @@ version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971" +[[package]] +name = "unarray" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94" + [[package]] name = "unicase" version = "2.9.0" diff --git a/crates/openshell-supervisor/Cargo.toml b/crates/openshell-supervisor/Cargo.toml index efbffc5d35..0d87981764 100644 --- a/crates/openshell-supervisor/Cargo.toml +++ b/crates/openshell-supervisor/Cargo.toml @@ -40,7 +40,8 @@ hyper-util = { workspace = true, features = ["tokio", "http1", "server", "server miette = { workspace = true } nix = { workspace = true } opentelemetry = { workspace = true } -opentelemetry-proto = { workspace = true } +# `with-serde` is the JSON half of the OTLP/HTTP relay's transcoding. +opentelemetry-proto = { workspace = true, features = ["with-serde"] } prost = { workspace = true } prost-types = { workspace = true } rustls = { workspace = true } diff --git a/crates/openshell-supervisor/README.md b/crates/openshell-supervisor/README.md index 1fa7f6ab9e..3abcb90ea0 100644 --- a/crates/openshell-supervisor/README.md +++ b/crates/openshell-supervisor/README.md @@ -125,9 +125,9 @@ Console JSON is installed independently of optional file appenders and uses an I ## OTLP agent trace relay -The `otlp_relay` module forwards the workload's OTLP/HTTP protobuf trace exports to the collector named by `OPENSHELL_OTLP_AGENT_ENDPOINT`, over the transport named by `OPENSHELL_OTLP_AGENT_PROTOCOL` (`grpc` when unset, or `http/protobuf`). Each export carries the collector headers from the launch bundle's `agent_otlp_headers` and, for an `https` collector, verifies it and authenticates with `agent_otlp_tls`; the supervisor's own span and OCSF exporters use the `traces_` and `ocsf_` fields the same way. The gateway sets those variables only when `[openshell.gateway.otlp.agent]` is enabled, so trace export alone leaves the relay inactive. Metrics and logs exports to the relay are acknowledged and discarded. The workload exports to the reserved address `192.0.0.8:4318`; the sandbox broker stages that connection to the supervisor, and the network proxy hands the stream to the relay's HTTP/1.1 receiver through the generic reserved-destination hook instead of evaluating policy. The receiver replaces the `openshell.sandbox.id` and `openshell.telemetry.source=agent` resource attributes, pushes the re-encoded batch into a bounded buffer, and a single export task forwards the encoded batches to the collector with a lazy connection. +The `otlp_relay` module forwards the workload's OTLP/HTTP trace exports, as protobuf or JSON, to the collector named by `OPENSHELL_OTLP_AGENT_ENDPOINT`, over the transport named by `OPENSHELL_OTLP_AGENT_PROTOCOL` (`grpc` when unset, or `http/protobuf`). Each export carries the collector headers from the launch bundle's `agent_otlp_headers` and, for an `https` collector, verifies it and authenticates with `agent_otlp_tls`; the supervisor's own span and OCSF exporters use the `traces_` and `ocsf_` fields the same way. The gateway sets those variables only when `[openshell.gateway.otlp.agent]` is enabled, so trace export alone leaves the relay inactive. Metrics and logs exports to the relay are acknowledged and discarded. The workload exports to the reserved address `192.0.0.8:4318`; the sandbox broker stages that connection to the supervisor, and the network proxy hands the stream to the relay's HTTP/1.1 receiver through the generic reserved-destination hook instead of evaluating policy. The receiver transcodes a JSON body to protobuf, replaces the `openshell.sandbox.id` and `openshell.telemetry.source=agent` resource attributes, pushes the re-encoded batch into a bounded buffer, and a single export task forwards the encoded batches to the collector with a lazy connection. -Limits are fixed: 2 MiB per request body, at most 512 `ResourceSpans` entries, 256 KiB of encoded `Resource` fields, and 8192 resource attributes per request (each of these answers 413), a 10 second header-read timeout, 16 concurrent connections, 32 buffered batches, and a 5 second deadline per export attempt. Only the resources are decoded; spans are copied byte for byte, at most two requests at a time. Worst-case memory attributable to the agent is about 67 MiB of buffered batches, plus 32 MiB of bodies awaiting enrichment across the 16 connections, plus about 12 MiB for the two enrichments in progress, plus about 4 MiB for the batch being exported and its encoded copy. A full buffer answers 503 with `Retry-After`. Without an agent endpoint nothing starts and the proxy refuses the relay address. After the agent exits, `run_sandbox_with_backend` flushes buffered batches for up to 5 seconds before it reports the exit to the gateway, then logs a final summary of rejected and failed counts. Flushes skip the wait while the most recent export failed, and batches still pending when the relay is dropped without a flush count as failed. The supervisor tells the sandbox runtime through `AgentSpec::otlp_relay_active` whether the relay runs, and the runtime injects the generic `OTEL_EXPORTER_OTLP_*` defaults only then. +Limits are fixed: 2 MiB per request body, at most 512 `ResourceSpans` entries, 256 KiB of encoded `Resource` fields, and 8192 resource attributes per request (each of these answers 413), a 10 second header-read timeout, 16 concurrent connections, 32 buffered batches, and a 5 second deadline per export attempt. A JSON body is also refused with 413 when it nests more than 32 levels or holds more than 100,000 values (every `{`, `[` and `,` outside strings), checked before it is decoded, and a JSON error message returned to the agent is cut to 1 KiB. Two requests are enriched at a time, on the blocking pool. For a protobuf body only the resources are decoded and spans are copied byte for byte; a JSON body is decoded in full, so `json.rs` bounds it by depth and value count instead of size. The decoder cannot read an `arrayValue` or `kvlistValue` without `values` or a `doubleValue` given as `"NaN"`, which answer 400. Worst-case memory attributable to the agent is about 67 MiB of buffered batches, plus 32 MiB of bodies awaiting enrichment across the 16 connections, plus about 12 MiB for the two enrichments in progress (about 88 MiB when both are JSON, from a worst case of about 44 MiB measured within the limits), plus about 4 MiB for the batch being exported and its encoded copy. A full buffer answers 503 with `Retry-After`. Without an agent endpoint nothing starts and the proxy refuses the relay address. After the agent exits, `run_sandbox_with_backend` flushes buffered batches for up to 5 seconds before it reports the exit to the gateway, then logs a final summary of rejected and failed counts. Flushes skip the wait while the most recent export failed, and batches still pending when the relay is dropped without a flush count as failed. The supervisor tells the sandbox runtime through `AgentSpec::otlp_relay_active` whether the relay runs, and the runtime injects the generic `OTEL_EXPORTER_OTLP_*` defaults only then. ## Safe runtime and parent liveness diff --git a/crates/openshell-supervisor/src/otlp_relay/enrichment.rs b/crates/openshell-supervisor/src/otlp_relay/enrichment.rs index 7fce2483d2..7e70c0003b 100644 --- a/crates/openshell-supervisor/src/otlp_relay/enrichment.rs +++ b/crates/openshell-supervisor/src/otlp_relay/enrichment.rs @@ -100,7 +100,8 @@ const BUFFER_UNDERFLOW: &str = "buffer underflow"; /// The request could not be enriched. #[derive(Debug)] pub enum EnrichError { - /// The bytes are not a well-formed protobuf message for this schema. + /// The bytes are not a well-formed protobuf message for this schema, or a + /// JSON body is not a well-formed OTLP/JSON request. Malformed(String), /// More than [`MAX_RESOURCE_SPANS`] entries. TooManyResourceSpans, @@ -108,6 +109,8 @@ pub enum EnrichError { TooManyResourceAttributes, /// The `resource` fields of the request exceed [`MAX_RESOURCE_BYTES`]. ResourceTooLarge, + /// A JSON body exceeds the shape limits in [`super::json`]. + JsonTooComplex, } impl std::fmt::Display for EnrichError { @@ -126,6 +129,12 @@ impl std::fmt::Display for EnrichError { "resource fields exceed {MAX_RESOURCE_BYTES} bytes in total" ) } + Self::JsonTooComplex => write!( + f, + "JSON body nests more than {} levels or holds more than {} values", + super::json::MAX_DEPTH, + super::json::MAX_VALUES + ), } } } diff --git a/crates/openshell-supervisor/src/otlp_relay/json.rs b/crates/openshell-supervisor/src/otlp_relay/json.rs new file mode 100644 index 0000000000..e4509d6929 --- /dev/null +++ b/crates/openshell-supervisor/src/otlp_relay/json.rs @@ -0,0 +1,282 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! OTLP/HTTP JSON request bodies. +//! +//! Enrichment works on protobuf bytes, so a JSON body is first transcoded to +//! protobuf with the serde impls of `opentelemetry-proto`. Those impls decode +//! the whole request, and what a body allocates grows with the number of +//! values it holds times how deeply they nest, not with its size: serde +//! buffers each `AnyValue` and copies that buffer again at every nested +//! `arrayValue` or `kvlistValue`. [`check_shape`] therefore scans the body +//! without allocating and refuses one that nests deeper than [`MAX_DEPTH`] or +//! holds more than [`MAX_VALUES`] values before anything is decoded. +//! +//! Memory: within both limits, decoding and re-encoding peaked at about 44 MiB +//! in the worst shape tried (a list of about 49 thousand strings inside an +//! `arrayValue` nested seven levels deep), about 17 MiB for 50 thousand empty +//! spans, and under 3 MiB for span batches of a realistic shape with the same +//! value count. +//! +//! The decoder follows the proto3 JSON mapping with two known gaps in +//! `opentelemetry-proto` 0.33: an `arrayValue` or `kvlistValue` must carry +//! its `values` field even when empty, and a `doubleValue` must be a JSON +//! number, so `"NaN"` is refused. Both answer 400. + +use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; +use prost::Message as _; + +use super::enrichment::EnrichError; + +/// Deepest container nesting a JSON body may have, counting the request +/// object itself. A span attribute's value sits ten levels down, so seven +/// levels of nested `AnyValue` fit. +pub const MAX_DEPTH: usize = 32; +/// Most values a JSON body may hold, counted as every `{`, `[` and `,` +/// outside strings: the containers plus the separators. A batch of 512 +/// spans with ten attributes each holds about 25 thousand. +pub const MAX_VALUES: usize = 100_000; +/// Longest decoder message returned to the agent. `serde_json` echoes the +/// offending string, which can be as long as the body. +const MAX_ERROR_BYTES: usize = 1024; + +/// Converts an OTLP/HTTP JSON body to the protobuf wire format that +/// enrichment works on. +pub fn transcode(body: &[u8]) -> Result, EnrichError> { + check_shape(body)?; + serde_json::from_slice::(body) + .map(|request| request.encode_to_vec()) + .map_err(|error| EnrichError::Malformed(bounded(error.to_string()))) +} + +/// Rejects a body that nests deeper than [`MAX_DEPTH`] or holds more than +/// [`MAX_VALUES`] values, in one pass that allocates nothing. Invalid JSON +/// passes; the decoder reports it. +fn check_shape(body: &[u8]) -> Result<(), EnrichError> { + let (mut depth, mut values, mut in_string, mut escaped) = (0usize, 0usize, false, false); + for &byte in body { + if in_string { + match (escaped, byte) { + (true, _) => escaped = false, + (false, b'\\') => escaped = true, + (false, b'"') => in_string = false, + _ => {} + } + continue; + } + match byte { + b'"' => in_string = true, + b'{' | b'[' => { + depth += 1; + values += 1; + if depth > MAX_DEPTH { + return Err(EnrichError::JsonTooComplex); + } + } + b'}' | b']' => depth = depth.saturating_sub(1), + b',' => values += 1, + _ => {} + } + if values > MAX_VALUES { + return Err(EnrichError::JsonTooComplex); + } + } + Ok(()) +} + +/// Cuts `message` to [`MAX_ERROR_BYTES`] on a character boundary. +fn bounded(mut message: String) -> String { + if message.len() > MAX_ERROR_BYTES { + message.truncate(message.floor_char_boundary(MAX_ERROR_BYTES)); + message.push_str("..."); + } + message +} + +#[cfg(test)] +pub mod test_util { + /// A request whose single span has one attribute with the given JSON + /// `AnyValue`. + pub fn with_attribute_value(value: &str) -> String { + format!( + r#"{{"resourceSpans":[{{"scopeSpans":[{{"spans":[{{"attributes":[{{"key":"k","value":{value}}}]}}]}}]}}]}}"# + ) + } + + /// A request whose attribute holds `zeros` scalars inside `levels` + /// nested `arrayValue`s. It has few braces, so counting `{` alone admits + /// it, but decoding it copies the scalars once per level. + pub fn nested_attribute_body(levels: usize, zeros: usize) -> String { + let mut value = format!(r#"{{"x":[{}]}}"#, vec!["0"; zeros].join(",")); + for _ in 0..levels { + value = format!(r#"{{"arrayValue":{{"values":[{value}]}}}}"#); + } + with_attribute_value(&value) + } +} + +#[cfg(test)] +mod tests { + use super::test_util::{nested_attribute_body, with_attribute_value}; + use super::*; + + fn too_complex(body: &str) -> bool { + matches!( + check_shape(body.as_bytes()), + Err(EnrichError::JsonTooComplex) + ) + } + + #[test] + fn depth_limit_is_inclusive() { + let nested = |depth: usize| format!("{}{}", "[".repeat(depth), "]".repeat(depth)); + assert!(!too_complex(&nested(MAX_DEPTH))); + assert!(too_complex(&nested(MAX_DEPTH + 1))); + } + + #[test] + fn value_limit_is_inclusive() { + // `[` plus one comma per further element is one value per element. + let list = |elements: usize| format!("[{}]", vec!["0"; elements].join(",")); + assert!(!too_complex(&list(MAX_VALUES))); + assert!(too_complex(&list(MAX_VALUES + 1))); + } + + #[test] + fn brackets_and_commas_inside_strings_are_not_counted() { + let text = "{[,".repeat(MAX_VALUES); + assert!(!too_complex(&format!(r#"{{"stringValue":"{text}"}}"#))); + // An escaped quote does not end the string. + let escaped = r#"\",{[,"#.repeat(MAX_VALUES); + assert!(!too_complex(&format!(r#"{{"stringValue":"{escaped}"}}"#))); + } + + #[test] + fn nested_values_are_refused_before_decoding() { + assert!(matches!( + transcode(nested_attribute_body(39, 20_000).as_bytes()), + Err(EnrichError::JsonTooComplex) + )); + } + + /// Shaped like the JSON written by the JavaScript OTLP exporter that + /// Claude Code uses: integer and string 64-bit values, hex IDs of either + /// case, every `AnyValue` variant, events, links and a status. + const EXPORTER_SHAPED: &str = r#"{ + "resourceSpans": [{ + "resource": { + "attributes": [ + {"key": "service.name", "value": {"stringValue": "claude-code"}}, + {"key": "process.pid", "value": {"intValue": 4242}} + ], + "droppedAttributesCount": 0 + }, + "scopeSpans": [{ + "scope": {"name": "claude-code.tracing", "version": "1.0.0"}, + "spans": [{ + "traceId": "5B8EFFF798038103D269B633813FC60C", + "spanId": "eee19b7ec3c1b174", + "parentSpanId": "eee19b7ec3c1b173", + "name": "llm_request", + "kind": 3, + "startTimeUnixNano": "1544712660000000000", + "endTimeUnixNano": 1544712661000000000, + "attributes": [ + {"key": "tokens", "value": {"intValue": "123"}}, + {"key": "cost", "value": {"doubleValue": 0.25}}, + {"key": "ok", "value": {"boolValue": true}}, + {"key": "blob", "value": {"bytesValue": "AQID"}}, + {"key": "list", "value": {"arrayValue": {"values": [ + {"stringValue": "a"}, {"intValue": 1} + ]}}}, + {"key": "map", "value": {"kvlistValue": {"values": [ + {"key": "inner", "value": {"stringValue": "b"}} + ]}}} + ], + "droppedAttributesCount": 0, + "events": [{ + "timeUnixNano": "1544712660500000000", + "name": "prompt", + "attributes": [], + "droppedAttributesCount": 0 + }], + "droppedEventsCount": 0, + "links": [{ + "traceId": "5b8efff798038103d269b633813fc60c", + "spanId": "eee19b7ec3c1b172", + "attributes": [], + "droppedAttributesCount": 0 + }], + "droppedLinksCount": 0, + "status": {"code": 1} + }] + }] + }] + }"#; + + #[test] + fn exporter_shaped_payload_is_transcoded() { + use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue, any_value::Value}; + + let protobuf = transcode(EXPORTER_SHAPED.as_bytes()).expect("transcoded"); + let request = ExportTraceServiceRequest::decode(protobuf.as_slice()).expect("protobuf"); + let resource = &request.resource_spans[0]; + let process_pid = &resource.resource.as_ref().expect("resource").attributes[1]; + assert_eq!( + process_pid.value, + Some(AnyValue { + value: Some(Value::IntValue(4242)) + }) + ); + let span = &resource.scope_spans[0].spans[0]; + assert_eq!(span.name, "llm_request"); + assert_eq!(span.kind, 3); + assert_eq!(span.start_time_unix_nano, 1_544_712_660_000_000_000); + assert_eq!(span.end_time_unix_nano, 1_544_712_661_000_000_000); + assert_eq!(span.trace_id[0], 0x5b); + assert_eq!(span.trace_id[15], 0x0c); + assert_eq!(span.parent_span_id[7], 0x73); + let attribute = |index: usize| -> &KeyValue { &span.attributes[index] }; + let value = |index: usize| attribute(index).value.clone().expect("value").value; + assert_eq!(value(0), Some(Value::IntValue(123))); + assert_eq!(value(1), Some(Value::DoubleValue(0.25))); + assert_eq!(value(2), Some(Value::BoolValue(true))); + assert_eq!(value(3), Some(Value::BytesValue(vec![1, 2, 3]))); + assert!(matches!(value(4), Some(Value::ArrayValue(list)) if list.values.len() == 2)); + assert!(matches!(value(5), Some(Value::KvlistValue(map)) if map.values.len() == 1)); + assert_eq!(span.events[0].name, "prompt"); + assert_eq!(span.links[0].span_id[7], 0x72); + assert_eq!(span.status.as_ref().expect("status").code, 1); + } + + #[test] + fn known_decoder_gaps_answer_malformed() { + // Valid OTLP/JSON that `opentelemetry-proto` 0.33 cannot decode. This + // pins the documented behavior so a dependency update that lifts the + // gaps updates the docs with it. + for value in [ + r#"{"arrayValue":{}}"#, + r#"{"kvlistValue":{}}"#, + r#"{"doubleValue":"NaN"}"#, + ] { + assert!( + matches!( + transcode(with_attribute_value(value).as_bytes()), + Err(EnrichError::Malformed(_)) + ), + "{value}" + ); + } + } + + #[test] + fn decoder_message_is_bounded() { + let text = "\u{200b}".repeat(300_000); + let body = format!(r#"{{"resourceSpans":"{text}"}}"#); + let Err(EnrichError::Malformed(message)) = transcode(body.as_bytes()) else { + panic!("a string is not a list of resource spans"); + }; + assert!(message.len() <= MAX_ERROR_BYTES + 3, "{}", message.len()); + assert!(message.ends_with("...")); + } +} diff --git a/crates/openshell-supervisor/src/otlp_relay/mod.rs b/crates/openshell-supervisor/src/otlp_relay/mod.rs index 4b45cda49c..161573c38f 100644 --- a/crates/openshell-supervisor/src/otlp_relay/mod.rs +++ b/crates/openshell-supervisor/src/otlp_relay/mod.rs @@ -3,10 +3,10 @@ //! Supervisor-direct OTLP trace relay. //! -//! A sandboxed agent exports OTLP/HTTP protobuf traces to the reserved -//! address [`openshell_core::sandbox_env::OTLP_RELAY_ADDR`]. The sandbox -//! broker stages that connection to the supervisor, the network proxy hands -//! the stream to [`receiver::OtlpConnectionServer`], and the accepted batch +//! A sandboxed agent exports OTLP/HTTP traces, as protobuf or JSON, to the +//! reserved address [`openshell_core::sandbox_env::OTLP_RELAY_ADDR`]. The +//! sandbox broker stages that connection to the supervisor, the network proxy +//! hands the stream to [`receiver::OtlpConnectionServer`], and the accepted batch //! is enriched with the sandbox id, buffered, and forwarded over OTLP/gRPC or //! OTLP/HTTP to the collector the gateway names in //! `OPENSHELL_OTLP_AGENT_ENDPOINT`, using `OPENSHELL_OTLP_AGENT_PROTOCOL`. The @@ -15,6 +15,7 @@ pub mod buffer; pub mod enrichment; pub mod exporter; +pub mod json; pub mod receiver; use std::net::SocketAddr; @@ -46,11 +47,13 @@ pub const MAX_BODY_BYTES: usize = 2 * 1024 * 1024; /// export buffer, [`BUFFER_CAPACITY`] batches of this size (about 67 MiB); /// bodies read but not yet enriched, [`MAX_CONCURRENT_CONNECTIONS`] times /// [`MAX_BODY_BYTES`] (32 MiB); and the working set of the -/// [`ENRICHMENT_CONCURRENCY`] enrichments in progress, each about twice the -/// body plus at most about 1 MiB of decoded resource elements (about -/// 12 MiB). On top of that, the export task holds the batch it took from -/// the buffer and tonic encodes a copy of it, one batch each (about 4 MiB). -/// Every allocation is sized to its content before it is held. +/// [`ENRICHMENT_CONCURRENCY`] enrichments in progress. A protobuf body +/// takes about twice its size plus at most about 1 MiB of decoded resource +/// elements (about 12 MiB for both). A JSON body is decoded in full first, +/// which peaked at about 44 MiB in the worst shape measured within the +/// limits in [`json`], about 88 MiB for both. On top of that, the export task holds the batch it took +/// from the buffer and tonic encodes a copy of it, one batch each (about +/// 4 MiB). Every allocation is sized to its content before it is held. pub const MAX_ENRICHED_BYTES: usize = MAX_BODY_BYTES + enrichment::MAX_RESOURCE_SPANS * enrichment::ATTRIBUTION_BYTES_PER_ENTRY; /// How long a connection may idle before a complete request header arrives. @@ -62,9 +65,10 @@ pub const BODY_READ_TIMEOUT: Duration = Duration::from_secs(30); pub const MAX_CONCURRENT_CONNECTIONS: usize = 16; /// Requests enriched at the same time. /// -/// Enrichment is a short CPU-bound copy whose working set is about twice the -/// body plus the decoded resources, so two slots bound that transient -/// regardless of how many connections hold a body. +/// Enrichment is a CPU-bound copy, run on the blocking pool, whose working +/// set is about twice the body plus the decoded resources, or the decoded +/// request for a JSON body, so two slots bound that transient regardless of +/// how many connections hold a body. pub const ENRICHMENT_CONCURRENCY: usize = 2; /// Number of enriched batches held between receipt and export. pub const BUFFER_CAPACITY: usize = 32; @@ -101,11 +105,15 @@ pub struct RelayCounters { /// Metrics and logs exports acknowledged and dropped, since the relay /// forwards traces only. discarded_signals: AtomicU64, + /// Trace exports refused with 400, 413 or 415 because the request itself + /// was unacceptable. + invalid_requests: AtomicU64, /// Whether the most recent export attempt failed. Cleared by the next /// success; flushes skip waiting while it is set. last_export_failed: AtomicBool, first_rejection_logged: AtomicBool, first_discard_logged: AtomicBool, + first_invalid_logged: AtomicBool, first_failure_logged: AtomicBool, final_summary_logged: AtomicBool, } @@ -146,6 +154,24 @@ impl RelayCounters { self.discarded_signals.load(Ordering::Relaxed) } + /// Records one export refused with `status` (400, 413 or 415) because + /// the request was malformed, over a limit or in an unsupported encoding. + /// Warns on the first occurrence only; the reason is logged at debug. + pub fn record_invalid_request(&self, status: u16) { + self.invalid_requests.fetch_add(1, Ordering::Relaxed); + if !self.first_invalid_logged.swap(true, Ordering::Relaxed) { + warn!( + status, + "OTLP relay refused an agent export as invalid; the response states the reason" + ); + } + } + + /// Exports refused as invalid so far. + pub fn invalid_requests(&self) -> u64 { + self.invalid_requests.load(Ordering::Relaxed) + } + /// Records one export attempt that failed or timed out. Warns on the /// first occurrence only. pub fn record_export_failure(&self, endpoint: &str, error: &dyn std::fmt::Display) { @@ -200,6 +226,7 @@ impl RelayCounters { rejected, export_failures, discarded_signals = self.discarded_signals(), + invalid_requests = self.invalid_requests(), "OTLP relay summary" ); } @@ -490,14 +517,14 @@ fn assemble( } /// Logs the cumulative counts every [`SUMMARY_INTERVAL`], but only while the -/// rejected or failed counts keep changing. +/// rejected, invalid or failed counts keep changing. async fn summary_loop(counters: Arc) { let mut interval = tokio::time::interval_at( tokio::time::Instant::now() + SUMMARY_INTERVAL, SUMMARY_INTERVAL, ); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - let mut last_reported = (0, 0); + let mut last_reported = (0, 0, 0); loop { interval.tick().await; let CounterSnapshot { @@ -505,8 +532,9 @@ async fn summary_loop(counters: Arc) { export_failures, .. } = counters.snapshot(); - if (rejected, export_failures) != last_reported { - last_reported = (rejected, export_failures); + let current = (rejected, counters.invalid_requests(), export_failures); + if current != last_reported { + last_reported = current; counters.log_summary(); } } @@ -1254,6 +1282,33 @@ mod tests { assert_eq!(counters.snapshot().rejected, 10); } + #[tokio::test(start_paused = true)] + async fn invalid_requests_warn_once_and_reach_the_summary() { + let logs = LogCapture::default(); + let _guard = logs.install(); + let counters = Arc::new(RelayCounters::default()); + let summary = tokio::spawn(summary_loop(Arc::clone(&counters))); + // Let the loop start its interval before time moves. + tokio::task::yield_now().await; + for status in [400, 413, 415] { + counters.record_invalid_request(status); + } + assert_eq!( + logs.count( + tracing::Level::WARN, + "OTLP relay refused an agent export as invalid; the response states the reason" + ), + 1 + ); + + tokio::time::advance(SUMMARY_INTERVAL + Duration::from_millis(1)).await; + tokio::task::yield_now().await; + let summaries = logs.events(tracing::Level::INFO, "OTLP relay summary"); + assert_eq!(summaries.len(), 1, "invalid requests alone trigger it"); + assert_eq!(summaries[0].field("invalid_requests"), Some(3)); + summary.abort(); + } + #[tokio::test] async fn first_export_failure_warns_once() { let logs = LogCapture::default(); diff --git a/crates/openshell-supervisor/src/otlp_relay/receiver.rs b/crates/openshell-supervisor/src/otlp_relay/receiver.rs index ea664698db..5fce6e8665 100644 --- a/crates/openshell-supervisor/src/otlp_relay/receiver.rs +++ b/crates/openshell-supervisor/src/otlp_relay/receiver.rs @@ -7,7 +7,7 @@ //! broker stages that connect for the supervisor, whose proxy hands the //! resulting duplex stream to [`OtlpConnectionServer::serve`] instead of //! dialing upstream. The server speaks HTTP/1.1 on that stream and accepts -//! `POST /v1/traces` with an uncompressed protobuf body. No socket is ever +//! `POST /v1/traces` with an uncompressed protobuf or JSON body. No socket is ever //! bound. use std::convert::Infallible; @@ -34,7 +34,7 @@ use tonic_types::Status as RpcStatus; use super::{ BODY_READ_TIMEOUT, ENRICHMENT_CONCURRENCY, HEADER_READ_TIMEOUT, MAX_BODY_BYTES, MAX_CONCURRENT_CONNECTIONS, MAX_ENRICHED_BYTES, RECEIVER_SHUTDOWN_TIMEOUT, RETRY_AFTER_SECS, - RelayCounters, enrichment, + RelayCounters, enrichment, json, }; /// Reservation for one relay connection. @@ -57,7 +57,7 @@ pub struct OtlpConnectionServer { counters: Arc, connections: Arc, /// Bounds how many bodies are being enriched at once. - enrichment_slots: Semaphore, + enrichment_slots: Arc, accepting: AtomicBool, /// `None` once shutdown has started. Watchers are minted under this lock, /// so none can be created after the shutdown signal is sent. @@ -88,7 +88,7 @@ impl OtlpConnectionServer { sandbox_id, counters, connections: Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTIONS)), - enrichment_slots: Semaphore::new(ENRICHMENT_CONCURRENCY), + enrichment_slots: Arc::new(Semaphore::new(ENRICHMENT_CONCURRENCY)), accepting: AtomicBool::new(true), graceful: Mutex::new(Some(GracefulShutdown::new())), abort_rx, @@ -181,6 +181,27 @@ impl OtlpConnectionServer { &self, request: Request, ) -> Result>, Infallible> { + let response = self.respond(request).await?; + if matches!( + response.status(), + StatusCode::BAD_REQUEST + | StatusCode::PAYLOAD_TOO_LARGE + | StatusCode::UNSUPPORTED_MEDIA_TYPE + ) { + self.counters + .record_invalid_request(response.status().as_u16()); + } + Ok(response) + } + + async fn respond( + &self, + request: Request, + ) -> Result>, Infallible> { + let encoding = Encoding::of(&request); + // Every response, success or error, uses the request's encoding, + // protobuf when it has none. + let response_encoding = encoding.unwrap_or(Encoding::Protobuf); // The relay forwards traces only. Agents whose SDK also exports // metrics or logs to the generic endpoint get an OTLP success, so they // do not retry forever, and the body is drained without being kept. @@ -191,14 +212,11 @@ impl OtlpConnectionServer { // the connection because the request was not consumed. let _ = tokio::time::timeout(BODY_READ_TIMEOUT, drain_body(request.into_body())).await; self.counters.record_discarded_signal(signal); - return Ok(Response::builder() - .status(StatusCode::OK) - .header(CONTENT_TYPE, "application/x-protobuf") - .body(Full::new(Bytes::new())) - .expect("static response builds")); + return Ok(success(response_encoding)); } if request.method() != Method::POST || request.uri().path() != "/v1/traces" { return Ok(rejected( + response_encoding, StatusCode::NOT_FOUND, "not found", "not found; the relay serves POST /v1/traces only", @@ -216,29 +234,20 @@ impl OtlpConnectionServer { }); if compressed { return Ok(rejected( + response_encoding, StatusCode::UNSUPPORTED_MEDIA_TYPE, "compressed body", - "compression is not supported; send an uncompressed application/x-protobuf body", + "compression is not supported; send an uncompressed body", )); } - // Media types are case-insensitive and may carry parameters. - let protobuf = request - .headers() - .get(CONTENT_TYPE) - .and_then(|value| value.to_str().ok()) - .and_then(|value| value.split(';').next()) - .is_some_and(|media_type| { - media_type - .trim() - .eq_ignore_ascii_case("application/x-protobuf") - }); - if !protobuf { + let Some(encoding) = encoding else { return Ok(rejected( + response_encoding, StatusCode::UNSUPPORTED_MEDIA_TYPE, "unsupported content type", - "unsupported content type; send application/x-protobuf", + "unsupported content type; send application/x-protobuf or application/json", )); - } + }; let declared_length = request .headers() .get(CONTENT_LENGTH) @@ -254,6 +263,7 @@ impl OtlpConnectionServer { Ok(BodyRead::Complete(body)) => body, Ok(BodyRead::TooLarge) => { return Ok(rejected( + response_encoding, StatusCode::PAYLOAD_TOO_LARGE, "body exceeds limit", "request body exceeds the 2 MiB limit", @@ -262,6 +272,7 @@ impl OtlpConnectionServer { Ok(BodyRead::Failed(error)) => { debug!(%error, "OTLP relay failed to read a request body"); return Ok(rejected( + response_encoding, StatusCode::BAD_REQUEST, "unreadable body", "failed to read request body", @@ -271,6 +282,7 @@ impl OtlpConnectionServer { // because the request was not consumed, which frees the slot. Err(_) => { return Ok(rejected( + response_encoding, StatusCode::REQUEST_TIMEOUT, "body read timed out", "request body was not received in time", @@ -278,31 +290,43 @@ impl OtlpConnectionServer { } }; - let enriched = { - let _slot = self - .enrichment_slots - .acquire() - .await - .expect("the enrichment semaphore is never closed"); - enrichment::enrich(&body, &self.sandbox_id) - }; - // The raw body is no longer needed once the enriched copy exists. - drop(body); + // Transcoding and enrichment are CPU-bound and take milliseconds on a + // large body, so they run off the async workers. The slot moves into + // the task, which keeps the concurrency bound when this request is + // cancelled while the task runs; the body is dropped there too. + let slot = Arc::clone(&self.enrichment_slots) + .acquire_owned() + .await + .expect("the enrichment semaphore is never closed"); + let sandbox_id = self.sandbox_id.clone(); + let enriched = tokio::task::spawn_blocking(move || { + let _slot = slot; + match encoding { + Encoding::Protobuf => enrichment::enrich(&body, &sandbox_id), + Encoding::Json => json::transcode(&body) + .and_then(|protobuf| enrichment::enrich(&protobuf, &sandbox_id)), + } + }) + .await + .expect("the enrichment task neither panics nor is cancelled"); let enriched = match enriched { Ok(enriched) => enriched, Err(error @ enrichment::EnrichError::Malformed(_)) => { return Ok(rejected( + response_encoding, StatusCode::BAD_REQUEST, - "protobuf decode failed", + "decode failed", &error.to_string(), )); } Err( error @ (enrichment::EnrichError::TooManyResourceSpans | enrichment::EnrichError::TooManyResourceAttributes - | enrichment::EnrichError::ResourceTooLarge), + | enrichment::EnrichError::ResourceTooLarge + | enrichment::EnrichError::JsonTooComplex), ) => { return Ok(rejected( + response_encoding, StatusCode::PAYLOAD_TOO_LARGE, "request exceeds structural limits", &error.to_string(), @@ -315,6 +339,7 @@ impl OtlpConnectionServer { // assumption. if enriched.len() > MAX_ENRICHED_BYTES { return Ok(rejected( + response_encoding, StatusCode::PAYLOAD_TOO_LARGE, "enriched body exceeds limit", "request exceeds the relay's batch limit after attribution", @@ -327,21 +352,18 @@ impl OtlpConnectionServer { match sent { Ok(()) => { self.counters.record_accepted(); - Ok(Response::builder() - .status(StatusCode::OK) - .header(CONTENT_TYPE, "application/x-protobuf") - .body(Full::new(Bytes::new())) - .expect("static response builds")) + Ok(success(response_encoding)) } Err(SendError::BufferFull) => { self.counters.record_rejection(); Ok(unavailable( + response_encoding, "relay buffer is full; retry after the Retry-After interval", )) } Err(SendError::BufferClosed) => { debug!("OTLP relay refused an export: relay is shutting down"); - Ok(unavailable("relay is shutting down")) + Ok(unavailable(response_encoding, "relay is shutting down")) } } } @@ -442,10 +464,57 @@ async fn read_body(mut body: Incoming, declared_length: Option) -> BodyRe } } -/// Builds an error response the way OTLP/HTTP specifies: a protobuf-encoded -/// `google.rpc.Status` in the request's content type, so collectors and SDKs -/// that decode it surface `message` instead of a bare status code. -fn rejected(status: StatusCode, reason: &str, message: &str) -> Response> { +/// The OTLP/HTTP body encoding of a request. +#[derive(Clone, Copy, PartialEq, Eq)] +enum Encoding { + Protobuf, + Json, +} + +impl Encoding { + /// Media types are case-insensitive and may carry parameters. + fn of(request: &Request) -> Option { + let value = request.headers().get(CONTENT_TYPE)?.to_str().ok()?; + let media_type = value.split(';').next()?.trim(); + if media_type.eq_ignore_ascii_case("application/x-protobuf") { + Some(Self::Protobuf) + } else if media_type.eq_ignore_ascii_case("application/json") { + Some(Self::Json) + } else { + None + } + } + + fn content_type(self) -> &'static str { + match self { + Self::Protobuf => "application/x-protobuf", + Self::Json => "application/json", + } + } +} + +/// The empty OTLP success: no bytes for protobuf, an empty object for JSON. +fn success(encoding: Encoding) -> Response> { + let body: &'static [u8] = match encoding { + Encoding::Protobuf => b"", + Encoding::Json => b"{}", + }; + Response::builder() + .status(StatusCode::OK) + .header(CONTENT_TYPE, encoding.content_type()) + .body(Full::new(Bytes::from_static(body))) + .expect("static response builds") +} + +/// Builds an error response the way OTLP/HTTP specifies: a `google.rpc.Status` +/// in the request's content type, so collectors and SDKs that decode it +/// surface `message` instead of a bare status code. +fn rejected( + encoding: Encoding, + status: StatusCode, + reason: &str, + message: &str, +) -> Response> { debug!( status = status.as_u16(), reason, "OTLP relay rejected request" @@ -456,11 +525,12 @@ fn rejected(status: StatusCode, reason: &str, message: &str) -> Response tonic::Code::DeadlineExceeded, _ => tonic::Code::InvalidArgument, }; - status_response(status, code, message) + status_response(encoding, status, code, message) } -fn unavailable(message: &str) -> Response> { +fn unavailable(encoding: Encoding, message: &str) -> Response> { let mut response = status_response( + encoding, StatusCode::SERVICE_UNAVAILABLE, tonic::Code::Unavailable, message, @@ -472,16 +542,27 @@ fn unavailable(message: &str) -> Response> { response } -fn status_response(status: StatusCode, code: tonic::Code, message: &str) -> Response> { - let body = RpcStatus { - code: code as i32, - message: message.to_string(), - details: Vec::new(), - } - .encode_to_vec(); +fn status_response( + encoding: Encoding, + status: StatusCode, + code: tonic::Code, + message: &str, +) -> Response> { + let body = match encoding { + Encoding::Protobuf => RpcStatus { + code: code as i32, + message: message.to_string(), + details: Vec::new(), + } + .encode_to_vec(), + Encoding::Json => { + serde_json::to_vec(&serde_json::json!({ "code": code as i32, "message": message })) + .expect("a status object serializes") + } + }; Response::builder() .status(status) - .header(CONTENT_TYPE, "application/x-protobuf") + .header(CONTENT_TYPE, encoding.content_type()) .body(Full::new(Bytes::from(body))) .expect("static response builds") } @@ -647,6 +728,7 @@ mod tests { use super::super::enrichment::test_util::{ encoded_request, encoded_request_with_large_span, resource_attributes, }; + use super::super::json::test_util::nested_attribute_body; use super::super::{BUFFER_CAPACITY, enrichment}; use super::test_util::{connect, read_head, request, request_with_headers, send}; use super::*; @@ -727,23 +809,209 @@ mod tests { assert_eq!(server.counters.discarded_signals(), 2); } + const JSON: &str = "application/json"; + + fn json_body() -> Vec { + br#"{"resourceSpans":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"json-agent"}}]},"scopeSpans":[{"spans":[{"traceId":"0af7651916cd43dd8448eb211c80319c","spanId":"b7ad6b7169203331","name":"json-span","kind":1,"startTimeUnixNano":"1","endTimeUnixNano":"2","attributes":[{"key":"tokens","value":{"intValue":"7"}}]}]}]}]}"#.to_vec() + } + + /// A valid JSON export of `spans` empty spans, which holds `2 * spans + 5` + /// values. + fn empty_spans_body(spans: usize) -> String { + format!( + "{{\"resourceSpans\":[{{\"scopeSpans\":[{{\"spans\":[{}]}}]}}]}}", + vec!["{}"; spans].join(",") + ) + } + + /// Decodes the JSON `google.rpc.Status` of an error response. + fn json_status(head: &test_util::Head) -> (i64, String) { + assert_eq!(head.header("content-type"), Some(JSON)); + let status: serde_json::Value = serde_json::from_slice(&head.body).expect("json status"); + ( + status["code"].as_i64().expect("code"), + status["message"].as_str().expect("message").to_string(), + ) + } + + #[tokio::test] + async fn json_request_is_transcoded_enriched_and_answered_in_json() { + use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; + + let (server, _handle, mut rx) = start(BUFFER_CAPACITY); + let mut client = connect(&server); + let head = send( + &mut client, + &request("POST", "/v1/traces", JSON, &json_body()), + ) + .await; + assert_eq!(head.status, 200); + assert_eq!(head.header("content-type"), Some(JSON)); + assert_eq!(head.body, b"{}"); + + let buffered = rx.recv().await.expect("one batch buffered"); + let attributes = resource_attributes(&buffered); + assert!(attributes.contains(&("service.name".to_string(), "json-agent".to_string()))); + assert!(attributes.contains(&( + enrichment::SANDBOX_ID_KEY.to_string(), + "sb-test".to_string() + ))); + assert!(attributes.contains(&( + enrichment::SOURCE_KEY.to_string(), + enrichment::SOURCE_VALUE.to_string() + ))); + let decoded = ExportTraceServiceRequest::decode(&buffered[..]).expect("protobuf"); + let span = &decoded.resource_spans[0].scope_spans[0].spans[0]; + assert_eq!(span.name, "json-span"); + // 64-bit integers arrive as strings in OTLP/JSON. + assert_eq!(span.kind, 1); + assert_eq!(span.start_time_unix_nano, 1); + assert_eq!(span.end_time_unix_nano, 2); + assert_eq!( + span.attributes[0] + .value + .as_ref() + .and_then(|v| v.value.clone()), + Some(opentelemetry_proto::tonic::common::v1::any_value::Value::IntValue(7)) + ); + assert_eq!( + span.trace_id, + [ + 0x0a, 0xf7, 0x65, 0x19, 0x16, 0xcd, 0x43, 0xdd, 0x84, 0x48, 0xeb, 0x21, 0x1c, 0x80, + 0x31, 0x9c + ] + ); + assert_eq!( + span.span_id, + [0xb7, 0xad, 0x6b, 0x71, 0x69, 0x20, 0x33, 0x31] + ); + } + + #[tokio::test] + async fn malformed_json_is_a_bad_request_in_json() { + let (server, _handle, mut rx) = start(BUFFER_CAPACITY); + let mut client = connect(&server); + let head = send( + &mut client, + &request("POST", "/v1/traces", JSON, b"{\"resourceSpans\": 1}"), + ) + .await; + assert_eq!(head.status, 400); + let (code, _message) = json_status(&head); + assert_eq!(code, i64::from(tonic::Code::InvalidArgument as i32)); + assert_eq!(server.counters.invalid_requests(), 1); + assert!( + tokio::time::timeout(Duration::from_millis(100), rx.recv()) + .await + .is_err(), + "nothing is buffered" + ); + } + + #[tokio::test] + async fn json_body_is_accepted_up_to_the_value_limit() { + // 49,997 empty spans hold 99,999 values and 49,998 hold 100,001, one + // either side of the limit. + let under = (json::MAX_VALUES - 5) / 2; + let (server, _handle, mut rx) = start(BUFFER_CAPACITY); + let mut client = connect(&server); + let head = send( + &mut client, + &request( + "POST", + "/v1/traces", + JSON, + empty_spans_body(under).as_bytes(), + ), + ) + .await; + assert_eq!(head.status, 200); + assert!(rx.recv().await.is_some(), "the batch is buffered"); + + let head = send( + &mut client, + &request( + "POST", + "/v1/traces", + JSON, + empty_spans_body(under + 1).as_bytes(), + ), + ) + .await; + assert_eq!(head.status, 413); + let (code, message) = json_status(&head); + assert_eq!(code, i64::from(tonic::Code::ResourceExhausted as i32)); + assert!( + message.contains("JSON body") && message.contains("100000 values"), + "{message}" + ); + assert!( + tokio::time::timeout(Duration::from_millis(100), rx.recv()) + .await + .is_err(), + "nothing more is buffered" + ); + assert_eq!(server.counters.invalid_requests(), 1); + } + + #[tokio::test] + async fn deeply_nested_json_is_too_large_and_never_buffered() { + let body = nested_attribute_body(39, 20_000); + let (server, _handle, mut rx) = start(BUFFER_CAPACITY); + let mut client = connect(&server); + let head = send( + &mut client, + &request("POST", "/v1/traces", JSON, body.as_bytes()), + ) + .await; + assert_eq!(head.status, 413); + let (_code, message) = json_status(&head); + assert!(message.contains("nests more than 32 levels"), "{message}"); + assert!( + tokio::time::timeout(Duration::from_millis(100), rx.recv()) + .await + .is_err(), + "nothing is buffered" + ); + } + + #[tokio::test] + async fn metrics_and_logs_json_are_acknowledged_with_an_empty_object() { + let (server, _handle, mut rx) = start(BUFFER_CAPACITY); + let mut client = connect(&server); + for path in ["/v1/metrics", "/v1/logs"] { + let head = send(&mut client, &request("POST", path, JSON, b"{}")).await; + assert_eq!(head.status, 200, "{path}"); + assert_eq!(head.header("content-type"), Some(JSON)); + assert_eq!(head.body, b"{}"); + } + assert_eq!(server.counters.discarded_signals(), 2); + assert!( + tokio::time::timeout(Duration::from_millis(100), rx.recv()) + .await + .is_err(), + "metrics and logs are not buffered" + ); + } + #[tokio::test] - async fn json_body_is_unsupported() { + async fn other_content_types_are_unsupported() { let (server, _handle, _rx) = start(BUFFER_CAPACITY); let mut client = connect(&server); let head = send( &mut client, - &request("POST", "/v1/traces", "application/json", b"{}"), + &request("POST", "/v1/traces", "text/plain", b"{}"), ) .await; assert_eq!(head.status, 415); - // The spec requires the body to state the supported content type, as + // The spec requires the body to state the supported content types, as // a google.rpc.Status in the request's content type. let status = head.rpc_status(); assert_eq!(status.code, tonic::Code::InvalidArgument as i32); assert!( - status.message.contains("application/x-protobuf"), - "415 body must name the supported content type: {}", + status.message.contains("application/x-protobuf") + && status.message.contains("application/json"), + "415 body must name the supported content types: {}", status.message ); } diff --git a/docs/how-it-works/gateways/configuration.mdx b/docs/how-it-works/gateways/configuration.mdx index 408ac0977a..375b76a1ce 100644 --- a/docs/how-it-works/gateways/configuration.mdx +++ b/docs/how-it-works/gateways/configuration.mdx @@ -867,7 +867,7 @@ While the relay is active, the sandbox runtime sets `OTEL_EXPORTER_OTLP_ENDPOINT Every forwarded resource carries `openshell.sandbox.id`, with the same value as the supervisor's own spans, and `openshell.telemetry.source=agent`. Values the agent supplies for those two keys are replaced at the resource level; span and scope attributes pass through unchanged, so filter and join on the resource attributes. Filter on `openshell.telemetry.source` to separate agent traces from infrastructure traces, and join on `openshell.sandbox.id` to correlate them. -Agents must export traces as OTLP/HTTP protobuf with compression disabled. The relay answers 415 to JSON bodies and to any `Content-Encoding` other than `identity`, 413 to bodies larger than 2 MiB, to requests with more than 512 `ResourceSpans` entries, and to requests whose resource attributes exceed 8192 in number or 256 KiB in total, 408 to requests whose body does not arrive within 30 seconds, and 503 with `Retry-After` while its buffer of 32 batches is full, which standard SDKs retry with backoff. Error responses carry a protobuf `google.rpc.Status` whose message states the reason. While the relay is off, connections to the relay address are refused immediately. +Agents must export traces as OTLP/HTTP with compression disabled, as protobuf or JSON. The relay answers 415 to other content types and to any `Content-Encoding` other than `identity`, 413 to bodies larger than 2 MiB, to requests with more than 512 `ResourceSpans` entries, and to requests whose resource attributes exceed 8192 in number or 256 KiB in total, 408 to requests whose body does not arrive within 30 seconds, and 503 with `Retry-After` while its buffer of 32 batches is full, which standard SDKs retry with backoff. A JSON body is converted to protobuf before it is forwarded, and the relay answers 413 to a JSON body that nests more than 32 levels or holds more than 100,000 values, counting every `{`, `[` and `,` outside strings, and 400 to an `arrayValue` or `kvlistValue` without a `values` field or a `doubleValue` given as a string. Responses use the request's content type, and error responses carry a `google.rpc.Status` whose message states the reason. While the relay is off, connections to the relay address are refused immediately. Delivery is best-effort. The relay answers 200 once a batch is buffered and makes one export attempt per batch, so a batch is lost if the collector is unavailable at that moment. The supervisor logs the first failed export and a periodic summary of failures. Before the sandbox reports that its main process exited, and before a suspended sandbox's supervisor exits, the relay flushes buffered batches for up to 5 seconds, and skips that wait when its most recent export failed. It keeps serving `openshell sandbox exec` sessions after the main process exits and flushes once more when the sandbox is torn down.