lodestone_client/
observability_ingest.rs

1//! Byte-exact client adapter for the generic observability Gateway leaf.
2//!
3//! The adapter deliberately transports the caller's encoded message without
4//! decoding OTLP or Elasticsearch bulk content. Collection credentials and
5//! trust remain owned by the validated Lodestone client session.
6
7use std::fmt::{self, Display, Formatter};
8use std::time::Duration;
9
10use http::{HeaderMap, HeaderValue, StatusCode, Version, header};
11use reqwest::{Certificate, Url, redirect::Policy};
12
13use crate::transport::{Client, authorization_value};
14use crate::{ClientError, CredentialProviderError};
15
16pub const MAX_OBSERVABILITY_INGEST_MESSAGE_BYTES: usize = 16 * 1024 * 1024;
17const MAX_OBSERVABILITY_RESPONSE_BYTES: usize = 16 * 1024 * 1024;
18const OPERATION_ID_HEADER: &str = "x-lodestone-operation-id";
19const REQUEST_DIGEST_HEADER: &str = "x-lodestone-request-digest";
20const MODE_HEADER: &str = "x-lodestone-ingest-mode";
21const PROFILE_HEADER: &str = "x-lodestone-observability-profile";
22const DISPOSITION_HEADER: &str = "x-lodestone-submission-disposition";
23const ACCEPTED_HEADER: &str = "x-lodestone-accepted-records";
24const REJECTED_HEADER: &str = "x-lodestone-rejected-records";
25const RECEIPT_DIGEST_HEADER: &str = "x-lodestone-receipt-digest";
26const INDEX_ID_HEADER: &str = "x-lodestone-index-id";
27const PROTOCOL_HEADER: &str = "x-lodestone-observability-protocol";
28const COLLECTION_ID_HEADER: &str = "x-lodestone-collection-id";
29const INDEX_NAME_HEADER: &str = "x-lodestone-index-name";
30
31#[derive(Clone, Debug, Eq, PartialEq)]
32pub struct ObservabilityIngestEndpointV1 {
33    origin: Url,
34}
35
36impl ObservabilityIngestEndpointV1 {
37    pub fn parse(value: &str) -> Result<Self, ObservabilityIngestErrorV1> {
38        let origin = Url::parse(value).map_err(|_| ObservabilityIngestErrorV1::InvalidRequest)?;
39        if origin.scheme() != "https"
40            || origin.cannot_be_a_base()
41            || origin.username() != ""
42            || origin.password().is_some()
43            || origin.query().is_some()
44            || origin.fragment().is_some()
45            || origin.host_str().is_none()
46            || !matches!(origin.path(), "" | "/")
47        {
48            return Err(ObservabilityIngestErrorV1::InvalidRequest);
49        }
50        Ok(Self { origin })
51    }
52
53    fn request_url(&self, path: &str) -> Result<Url, ObservabilityIngestErrorV1> {
54        self.origin
55            .join(path.trim_start_matches('/'))
56            .map_err(|_| ObservabilityIngestErrorV1::InvalidRequest)
57    }
58}
59
60#[derive(Clone, Copy, Debug, Eq, PartialEq)]
61pub enum ObservabilityIngestProtocolV1 {
62    OtlpLogsHttpProtobuf,
63    OtlpLogsHttpJson,
64    OtlpTracesHttpProtobuf,
65    OtlpTracesHttpJson,
66    OtlpLogsGrpc,
67    OtlpTracesGrpc,
68    ElasticsearchBulkVector,
69    ElasticsearchBulkFluentBit,
70}
71
72impl ObservabilityIngestProtocolV1 {
73    #[must_use]
74    pub const fn as_str(self) -> &'static str {
75        match self {
76            Self::OtlpLogsHttpProtobuf => "otlp_logs_http_protobuf",
77            Self::OtlpLogsHttpJson => "otlp_logs_http_json",
78            Self::OtlpTracesHttpProtobuf => "otlp_traces_http_protobuf",
79            Self::OtlpTracesHttpJson => "otlp_traces_http_json",
80            Self::OtlpLogsGrpc => "otlp_logs_grpc",
81            Self::OtlpTracesGrpc => "otlp_traces_grpc",
82            Self::ElasticsearchBulkVector => "elasticsearch_bulk_vector",
83            Self::ElasticsearchBulkFluentBit => "elasticsearch_bulk_fluent_bit",
84        }
85    }
86
87    const fn path(self) -> &'static str {
88        match self {
89            Self::OtlpLogsHttpProtobuf | Self::OtlpLogsHttpJson => "/v1/logs",
90            Self::OtlpTracesHttpProtobuf | Self::OtlpTracesHttpJson => "/v1/traces",
91            Self::OtlpLogsGrpc => "/opentelemetry.proto.collector.logs.v1.LogsService/Export",
92            Self::OtlpTracesGrpc => "/opentelemetry.proto.collector.trace.v1.TraceService/Export",
93            Self::ElasticsearchBulkVector | Self::ElasticsearchBulkFluentBit => "/v1/elasticsearch/_bulk",
94        }
95    }
96
97    const fn content_type(self) -> &'static str {
98        match self {
99            Self::OtlpLogsHttpProtobuf | Self::OtlpTracesHttpProtobuf => "application/x-protobuf",
100            Self::OtlpLogsHttpJson | Self::OtlpTracesHttpJson => "application/json",
101            Self::OtlpLogsGrpc | Self::OtlpTracesGrpc => "application/grpc",
102            Self::ElasticsearchBulkVector | Self::ElasticsearchBulkFluentBit => "application/x-ndjson",
103        }
104    }
105
106    const fn profile(self) -> Option<&'static str> {
107        match self {
108            Self::ElasticsearchBulkVector => Some("vector"),
109            Self::ElasticsearchBulkFluentBit => Some("fluent_bit"),
110            _ => None,
111        }
112    }
113
114    const fn is_grpc(self) -> bool {
115        matches!(self, Self::OtlpLogsGrpc | Self::OtlpTracesGrpc)
116    }
117}
118
119#[derive(Clone, Copy, Debug, Eq, PartialEq)]
120pub enum ObservabilityContentEncodingV1 {
121    Identity,
122    Gzip,
123}
124
125impl ObservabilityContentEncodingV1 {
126    const fn as_str(self) -> &'static str {
127        match self {
128            Self::Identity => "identity",
129            Self::Gzip => "gzip",
130        }
131    }
132}
133
134#[derive(Clone, Copy, Debug, Eq, PartialEq)]
135pub enum ObservabilityIngestModeV1 {
136    Submit,
137    Resolve,
138}
139
140impl ObservabilityIngestModeV1 {
141    const fn as_str(self) -> &'static str {
142        match self {
143            Self::Submit => "submit",
144            Self::Resolve => "resolve",
145        }
146    }
147}
148
149#[derive(Clone, Copy, Debug, Eq, PartialEq)]
150pub enum ObservabilityIngestDispositionV1 {
151    Committed,
152    Replayed,
153    SubmittedIndeterminate,
154    Conflict,
155    NotSubmitted,
156    Retired,
157}
158
159impl ObservabilityIngestDispositionV1 {
160    fn parse(value: &str) -> Option<Self> {
161        match value {
162            "committed" => Some(Self::Committed),
163            "replayed" => Some(Self::Replayed),
164            "submitted_indeterminate" => Some(Self::SubmittedIndeterminate),
165            "conflict" => Some(Self::Conflict),
166            "not_submitted" => Some(Self::NotSubmitted),
167            "retired" => Some(Self::Retired),
168            _ => None,
169        }
170    }
171}
172
173pub struct ObservabilityIngestRequestV1 {
174    pub endpoint: ObservabilityIngestEndpointV1,
175    pub operation_id: String,
176    pub request_digest: [u8; 32],
177    pub protocol: ObservabilityIngestProtocolV1,
178    pub mode: ObservabilityIngestModeV1,
179    pub expected_index_id: String,
180    pub index_name: String,
181    pub content_encoding: ObservabilityContentEncodingV1,
182    pub exact_encoded_message: Vec<u8>,
183    pub timeout: Duration,
184}
185
186#[derive(Clone, Debug, Eq, PartialEq)]
187pub struct ObservabilityIngestReceiptV1 {
188    pub operation_id: String,
189    pub request_digest: [u8; 32],
190    pub protocol: ObservabilityIngestProtocolV1,
191    pub disposition: ObservabilityIngestDispositionV1,
192    pub accepted_records: u64,
193    pub rejected_records: u64,
194    pub source_bytes: u64,
195    pub index_id: String,
196    pub receipt_digest: [u8; 32],
197}
198
199pub struct ObservabilityIngestResponseV1 {
200    pub receipt: ObservabilityIngestReceiptV1,
201    pub status: StatusCode,
202    pub content_type: String,
203    pub transport_body: Vec<u8>,
204}
205
206#[derive(Clone, Copy, Debug, Eq, PartialEq)]
207pub enum ObservabilityIngestErrorV1 {
208    InvalidRequest,
209    CredentialUnavailable,
210    DependencyUnavailable,
211    SubmittedIndeterminate,
212    Refused { status: u16, retry_safe: bool },
213    InvalidResponse,
214}
215
216impl Display for ObservabilityIngestErrorV1 {
217    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
218        formatter.write_str(match self {
219            Self::InvalidRequest => "observability ingest request is invalid",
220            Self::CredentialUnavailable => "observability ingest credential is unavailable",
221            Self::DependencyUnavailable => "observability ingest dependency is unavailable",
222            Self::SubmittedIndeterminate => "observability ingest submission is indeterminate",
223            Self::Refused { .. } => "observability ingest was refused",
224            Self::InvalidResponse => "observability ingest response is invalid",
225        })
226    }
227}
228
229impl std::error::Error for ObservabilityIngestErrorV1 {}
230
231impl Client {
232    pub async fn observability_ingest(
233        &self,
234        request: ObservabilityIngestRequestV1,
235    ) -> Result<ObservabilityIngestResponseV1, ObservabilityIngestErrorV1> {
236        validate_request(self, &request)?;
237        let material = self.credential().await.map_err(map_client_error)?;
238        let authorization = authorization_value(&material).map_err(map_client_error)?;
239        let trust = self
240            .transport_config()
241            .trust
242            .certificates_der()
243            .iter()
244            .map(|certificate| {
245                Certificate::from_der(certificate).map_err(|_| ObservabilityIngestErrorV1::InvalidRequest)
246            })
247            .collect::<Result<Vec<_>, _>>()?;
248        let mut builder = reqwest::Client::builder()
249            .https_only(true)
250            .no_proxy()
251            .redirect(Policy::none())
252            .connect_timeout(self.pool_config().connect_timeout)
253            .timeout(request.timeout)
254            .tls_certs_only(trust);
255        if request.protocol.is_grpc() {
256            builder = builder.http2_prior_knowledge();
257        }
258        let client = builder
259            .build()
260            .map_err(|_| ObservabilityIngestErrorV1::DependencyUnavailable)?;
261        let body = if request.protocol.is_grpc() {
262            grpc_frame(&request.exact_encoded_message, request.content_encoding)?
263        } else {
264            request.exact_encoded_message.clone()
265        };
266        let mut outbound = client
267            .post(request.endpoint.request_url(request.protocol.path())?)
268            .header(header::AUTHORIZATION, sensitive_header(authorization.as_slice())?)
269            .header(COLLECTION_ID_HEADER, self.expected_collection_id())
270            .header(OPERATION_ID_HEADER, &request.operation_id)
271            .header(REQUEST_DIGEST_HEADER, lower_hex(&request.request_digest))
272            .header(MODE_HEADER, request.mode.as_str())
273            .header(INDEX_ID_HEADER, &request.expected_index_id)
274            .header(header::CONTENT_TYPE, request.protocol.content_type())
275            .body(body);
276        if request.protocol.is_grpc() {
277            outbound = outbound
278                .version(Version::HTTP_2)
279                .header("te", "trailers")
280                .header(INDEX_NAME_HEADER, &request.index_name)
281                .header("grpc-encoding", request.content_encoding.as_str());
282        } else {
283            outbound = outbound.header(header::CONTENT_ENCODING, request.content_encoding.as_str());
284        }
285        if let Some(profile) = request.protocol.profile() {
286            outbound = outbound.header(PROFILE_HEADER, profile);
287        }
288        let response = outbound.send().await.map_err(map_reqwest_error)?;
289        let status = response.status();
290        let headers = response.headers().clone();
291        let content_type = exact_header(&headers, header::CONTENT_TYPE.as_str())
292            .unwrap_or("application/octet-stream")
293            .to_owned();
294        let body = response
295            .bytes()
296            .await
297            .map_err(|_| ObservabilityIngestErrorV1::SubmittedIndeterminate)?;
298        if body.len() > MAX_OBSERVABILITY_RESPONSE_BYTES {
299            return Err(ObservabilityIngestErrorV1::InvalidResponse);
300        }
301        let receipt_headers_present = headers.contains_key(DISPOSITION_HEADER);
302        if !receipt_headers_present {
303            return Err(map_refusal(status));
304        }
305        let transport_body = if request.protocol.is_grpc() {
306            unframe_grpc_response(&body)?
307        } else {
308            body.to_vec()
309        };
310        let receipt = parse_receipt(&headers, &request)?;
311        Ok(ObservabilityIngestResponseV1 {
312            receipt,
313            status,
314            content_type,
315            transport_body,
316        })
317    }
318}
319
320fn validate_request(client: &Client, request: &ObservabilityIngestRequestV1) -> Result<(), ObservabilityIngestErrorV1> {
321    let server_name = client.transport_config().identity.tls_server_name();
322    if request.endpoint.origin.host_str() != Some(server_name)
323        || !valid_uuid_v7(&request.operation_id)
324        || request.request_digest == [0; 32]
325        || blake3::hash(&request.exact_encoded_message).as_bytes() != &request.request_digest
326        || !valid_lower_hex(&request.expected_index_id, 16)
327        || request.index_name.is_empty()
328        || request.index_name.len() > 255
329        || request.index_name.bytes().any(|byte| byte.is_ascii_control())
330        || request.exact_encoded_message.is_empty()
331        || request.exact_encoded_message.len() > MAX_OBSERVABILITY_INGEST_MESSAGE_BYTES
332        || request.timeout.is_zero()
333        || request.timeout > Duration::from_secs(60)
334    {
335        return Err(ObservabilityIngestErrorV1::InvalidRequest);
336    }
337    Ok(())
338}
339
340fn parse_receipt(
341    headers: &HeaderMap,
342    request: &ObservabilityIngestRequestV1,
343) -> Result<ObservabilityIngestReceiptV1, ObservabilityIngestErrorV1> {
344    let disposition = ObservabilityIngestDispositionV1::parse(exact_header(headers, DISPOSITION_HEADER)?)
345        .ok_or(ObservabilityIngestErrorV1::InvalidResponse)?;
346    let accepted_records = canonical_u64(exact_header(headers, ACCEPTED_HEADER)?)?;
347    let rejected_records = canonical_u64(exact_header(headers, REJECTED_HEADER)?)?;
348    let receipt_digest = decode_digest(exact_header(headers, RECEIPT_DIGEST_HEADER)?)?;
349    let index_id = exact_header(headers, INDEX_ID_HEADER)?;
350    let protocol = exact_header(headers, PROTOCOL_HEADER)?;
351    if index_id != request.expected_index_id
352        || protocol != request.protocol.as_str()
353        || receipt_digest == [0; 32]
354        || matches!(
355            disposition,
356            ObservabilityIngestDispositionV1::Committed | ObservabilityIngestDispositionV1::Replayed
357        ) && accepted_records == 0
358    {
359        return Err(ObservabilityIngestErrorV1::InvalidResponse);
360    }
361    Ok(ObservabilityIngestReceiptV1 {
362        operation_id: request.operation_id.clone(),
363        request_digest: request.request_digest,
364        protocol: request.protocol,
365        disposition,
366        accepted_records,
367        rejected_records,
368        source_bytes: u64::try_from(request.exact_encoded_message.len())
369            .map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)?,
370        index_id: index_id.to_owned(),
371        receipt_digest,
372    })
373}
374
375fn exact_header<'a>(headers: &'a HeaderMap, name: &'static str) -> Result<&'a str, ObservabilityIngestErrorV1> {
376    let mut values = headers.get_all(name).iter();
377    let value = values
378        .next()
379        .and_then(|value| value.to_str().ok())
380        .filter(|value| !value.is_empty())
381        .ok_or(ObservabilityIngestErrorV1::InvalidResponse)?;
382    if values.next().is_some() {
383        return Err(ObservabilityIngestErrorV1::InvalidResponse);
384    }
385    Ok(value)
386}
387
388fn canonical_u64(value: &str) -> Result<u64, ObservabilityIngestErrorV1> {
389    if value.is_empty() || !value.bytes().all(|byte| byte.is_ascii_digit()) || value.starts_with('0') && value != "0" {
390        return Err(ObservabilityIngestErrorV1::InvalidResponse);
391    }
392    value
393        .parse::<u64>()
394        .map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)
395}
396
397fn decode_digest(value: &str) -> Result<[u8; 32], ObservabilityIngestErrorV1> {
398    if !valid_lower_hex(value, 32) {
399        return Err(ObservabilityIngestErrorV1::InvalidResponse);
400    }
401    let mut digest = [0_u8; 32];
402    for (output, pair) in digest.iter_mut().zip(value.as_bytes().as_chunks::<2>().0) {
403        let pair = std::str::from_utf8(pair).map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)?;
404        *output = u8::from_str_radix(pair, 16).map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)?;
405    }
406    Ok(digest)
407}
408
409fn valid_lower_hex(value: &str, bytes: usize) -> bool {
410    value.len() == bytes.saturating_mul(2)
411        && value
412            .bytes()
413            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
414        && value.bytes().any(|byte| byte != b'0')
415}
416
417fn valid_uuid_v7(value: &str) -> bool {
418    value.len() == 36
419        && value.as_bytes().get(8) == Some(&b'-')
420        && value.as_bytes().get(13) == Some(&b'-')
421        && value.as_bytes().get(18) == Some(&b'-')
422        && value.as_bytes().get(23) == Some(&b'-')
423        && value.as_bytes().get(14) == Some(&b'7')
424        && value
425            .as_bytes()
426            .get(19)
427            .is_some_and(|value| matches!(value, b'8' | b'9' | b'a' | b'b'))
428        && value.bytes().enumerate().all(|(index, byte)| {
429            matches!(index, 8 | 13 | 18 | 23) || byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)
430        })
431}
432
433fn grpc_frame(message: &[u8], encoding: ObservabilityContentEncodingV1) -> Result<Vec<u8>, ObservabilityIngestErrorV1> {
434    let length = u32::try_from(message.len()).map_err(|_| ObservabilityIngestErrorV1::InvalidRequest)?;
435    let mut framed = Vec::with_capacity(message.len().saturating_add(5));
436    framed.push(u8::from(encoding == ObservabilityContentEncodingV1::Gzip));
437    framed.extend_from_slice(&length.to_be_bytes());
438    framed.extend_from_slice(message);
439    Ok(framed)
440}
441
442fn unframe_grpc_response(body: &[u8]) -> Result<Vec<u8>, ObservabilityIngestErrorV1> {
443    if body.len() < 5 || body[0] != 0 {
444        return Err(ObservabilityIngestErrorV1::InvalidResponse);
445    }
446    let declared = usize::try_from(u32::from_be_bytes([body[1], body[2], body[3], body[4]]))
447        .map_err(|_| ObservabilityIngestErrorV1::InvalidResponse)?;
448    if declared != body.len().saturating_sub(5) {
449        return Err(ObservabilityIngestErrorV1::InvalidResponse);
450    }
451    Ok(body[5..].to_vec())
452}
453
454fn sensitive_header(value: &[u8]) -> Result<HeaderValue, ObservabilityIngestErrorV1> {
455    let mut header = HeaderValue::from_bytes(value).map_err(|_| ObservabilityIngestErrorV1::CredentialUnavailable)?;
456    header.set_sensitive(true);
457    Ok(header)
458}
459
460fn lower_hex(bytes: &[u8]) -> String {
461    let mut output = String::with_capacity(bytes.len().saturating_mul(2));
462    for byte in bytes {
463        use std::fmt::Write as _;
464        let _ = write!(output, "{byte:02x}");
465    }
466    output
467}
468
469fn map_client_error(error: ClientError) -> ObservabilityIngestErrorV1 {
470    match error {
471        ClientError::Credential(
472            CredentialProviderError::Unavailable
473            | CredentialProviderError::NotFound
474            | CredentialProviderError::InvalidMaterial
475            | CredentialProviderError::RandomUnavailable,
476        ) => ObservabilityIngestErrorV1::CredentialUnavailable,
477        _ => ObservabilityIngestErrorV1::DependencyUnavailable,
478    }
479}
480
481fn map_reqwest_error(error: reqwest::Error) -> ObservabilityIngestErrorV1 {
482    if error.is_builder() || error.is_connect() {
483        ObservabilityIngestErrorV1::DependencyUnavailable
484    } else {
485        ObservabilityIngestErrorV1::SubmittedIndeterminate
486    }
487}
488
489fn map_refusal(status: StatusCode) -> ObservabilityIngestErrorV1 {
490    ObservabilityIngestErrorV1::Refused {
491        status: status.as_u16(),
492        retry_safe: matches!(status, StatusCode::TOO_MANY_REQUESTS | StatusCode::SERVICE_UNAVAILABLE),
493    }
494}
495
496#[cfg(test)]
497mod tests {
498    use super::*;
499
500    #[test]
501    fn endpoint_is_an_exact_https_origin() {
502        assert!(ObservabilityIngestEndpointV1::parse("https://core.example:8444").is_ok());
503        assert!(ObservabilityIngestEndpointV1::parse("http://core.example:8444").is_err());
504        assert!(ObservabilityIngestEndpointV1::parse("https://core.example:8444/v1/logs").is_err());
505        assert!(ObservabilityIngestEndpointV1::parse("https://user@core.example:8444").is_err());
506    }
507
508    #[test]
509    fn grpc_framing_preserves_the_exact_message() -> Result<(), Box<dyn std::error::Error>> {
510        let message = b"exact opaque bytes";
511        let framed = grpc_frame(message, ObservabilityContentEncodingV1::Identity)?;
512        assert_eq!(unframe_grpc_response(&framed)?, message);
513        let compressed = grpc_frame(message, ObservabilityContentEncodingV1::Gzip)?;
514        assert_eq!(compressed[0], 1);
515        Ok(())
516    }
517
518    #[test]
519    fn receipt_headers_are_closed_and_request_bound() -> Result<(), Box<dyn std::error::Error>> {
520        let message = b"opaque".to_vec();
521        let request = ObservabilityIngestRequestV1 {
522            endpoint: ObservabilityIngestEndpointV1::parse("https://core.example:8444")?,
523            operation_id: "019a3bc2-0e66-7f38-a446-ef98386bfcd1".to_owned(),
524            request_digest: *blake3::hash(&message).as_bytes(),
525            protocol: ObservabilityIngestProtocolV1::OtlpLogsHttpProtobuf,
526            mode: ObservabilityIngestModeV1::Submit,
527            expected_index_id: "01".repeat(16),
528            index_name: "logs".to_owned(),
529            content_encoding: ObservabilityContentEncodingV1::Identity,
530            exact_encoded_message: message,
531            timeout: Duration::from_secs(1),
532        };
533        let mut headers = HeaderMap::new();
534        for (name, value) in [
535            (DISPOSITION_HEADER, "committed".to_owned()),
536            (ACCEPTED_HEADER, "1".to_owned()),
537            (REJECTED_HEADER, "0".to_owned()),
538            (RECEIPT_DIGEST_HEADER, "02".repeat(32)),
539            (INDEX_ID_HEADER, request.expected_index_id.clone()),
540            (PROTOCOL_HEADER, request.protocol.as_str().to_owned()),
541        ] {
542            headers.insert(
543                http::header::HeaderName::from_static(name),
544                HeaderValue::from_str(&value)?,
545            );
546        }
547        let receipt = parse_receipt(&headers, &request)?;
548        assert_eq!(receipt.accepted_records, 1);
549        headers.insert(INDEX_ID_HEADER, HeaderValue::from_static("01"));
550        assert_eq!(
551            parse_receipt(&headers, &request),
552            Err(ObservabilityIngestErrorV1::InvalidResponse)
553        );
554        Ok(())
555    }
556}
lodestone_client/observability_ingest.rs

lodestone_client/
observability_ingest.rs
