micromegas/servers/
otlp.rs

1//! OTLP/HTTP route registration for `telemetry-ingestion-srv`.
2//!
3//! Exposes three routes that match the OTLP/HTTP spec:
4//!  - `POST /ingestion/otlp/v1/logs`
5//!  - `POST /ingestion/otlp/v1/metrics`
6//!  - `POST /ingestion/otlp/v1/traces`
7//!
8//! The OTLP sub-router applies its own 20 MiB body limit (matching the OTel Collector
9//! `confighttp.max_request_body_size` default) plus gzip request decompression,
10//! independent of the parent router's 100 MiB limit on `/ingestion/insert_block`.
11//!
12//! Per OTLP/HTTP spec, success responses mirror the request encoding (JSON in → JSON out,
13//! proto in → proto out). Error responses (4xx/5xx) carry a `google.rpc.Status` body
14//! encoded in the same way, except 415 responses which always use protobuf because the
15//! request encoding is unknown at that point.
16
17use super::ingestion_limits::{RETRY_AFTER_SECONDS, apply_ingestion_body_limits};
18use axum::Extension;
19use axum::Router;
20use axum::body::Body;
21use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
22use axum::response::Response;
23use axum::routing::post;
24use micromegas_ingestion::web_ingestion_service::WebIngestionService;
25use micromegas_otel_ingestion::Encoding;
26use micromegas_otel_ingestion::error::OtelError;
27use micromegas_otel_ingestion::handler;
28use micromegas_tracing::prelude::*;
29use prost::Message;
30use std::sync::Arc;
31
32const CONTENT_TYPE_PROTOBUF: &str = "application/x-protobuf";
33const CONTENT_TYPE_JSON: &str = "application/json";
34
35/// Examines the `Content-Type` header and maps it to an `Encoding`. The spec allows
36/// parameters (e.g. `application/json; charset=utf-8`), so we parse rather than
37/// string-compare. Returns `Err(OtlpHttpError::WrongContentType)` for unknown types.
38fn content_type_encoding(headers: &HeaderMap) -> Result<Encoding, OtlpHttpError> {
39    let Some(ct) = headers.get(header::CONTENT_TYPE) else {
40        return Err(OtlpHttpError::WrongContentType);
41    };
42    let Ok(ct) = ct.to_str() else {
43        return Err(OtlpHttpError::WrongContentType);
44    };
45    let media = ct
46        .split(';')
47        .next()
48        .unwrap_or("")
49        .trim()
50        .to_ascii_lowercase();
51    match media.as_str() {
52        CONTENT_TYPE_PROTOBUF => Ok(Encoding::Protobuf),
53        CONTENT_TYPE_JSON => Ok(Encoding::Json),
54        _ => Err(OtlpHttpError::WrongContentType),
55    }
56}
57
58/// Internal error type covering both pre-handler validation failures (415) and
59/// post-handler `OtelError`s. Each variant maps to a single HTTP response shape
60/// (status code, optional `Retry-After`, `google.rpc.Status` body).
61enum OtlpHttpError {
62    WrongContentType,
63    Otel(OtelError),
64}
65
66impl OtlpHttpError {
67    fn into_otlp_response(self, encoding: Encoding) -> Response {
68        match self {
69            OtlpHttpError::WrongContentType => build_error_response(
70                StatusCode::UNSUPPORTED_MEDIA_TYPE,
71                3, // INVALID_ARGUMENT
72                "Content-Type must be application/x-protobuf or application/json",
73                false,
74                // encoding is unknown for 415; always emit proto Status (OTLP/HTTP default)
75                Encoding::Protobuf,
76            ),
77            OtlpHttpError::Otel(err) => {
78                let retryable = err.is_retryable();
79                let status = match err.http_status() {
80                    400 => StatusCode::BAD_REQUEST,
81                    503 => StatusCode::SERVICE_UNAVAILABLE,
82                    other => {
83                        StatusCode::from_u16(other).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR)
84                    }
85                };
86                let code = err.grpc_code();
87                // Detailed error (includes raw sqlx / object-store messages) is
88                // logged server-side; only the sanitized public form goes to the
89                // client to avoid leaking backend internals.
90                error!("OTLP error: {}", err);
91                build_error_response(status, code, &err.public_message(), retryable, encoding)
92            }
93        }
94    }
95}
96
97fn build_error_response(
98    status: StatusCode,
99    code: i32,
100    message: &str,
101    retryable: bool,
102    encoding: Encoding,
103) -> Response {
104    let proto_status = micromegas_otel_ingestion::proto::Status {
105        code,
106        message: message.to_string(),
107    };
108    let (body, content_type) = match encoding {
109        Encoding::Protobuf => (proto_status.encode_to_vec(), CONTENT_TYPE_PROTOBUF),
110        Encoding::Json => (
111            serde_json::to_vec(&proto_status).expect("serializing Status to JSON"),
112            CONTENT_TYPE_JSON,
113        ),
114    };
115    let mut response = Response::builder()
116        .status(status)
117        .header(header::CONTENT_TYPE, HeaderValue::from_static(content_type))
118        .body(Body::from(body))
119        .expect("building OTLP error response");
120    if retryable && let Ok(value) = HeaderValue::from_str(&RETRY_AFTER_SECONDS.to_string()) {
121        response.headers_mut().insert(header::RETRY_AFTER, value);
122    }
123    response
124}
125
126fn success_response<M: Message + serde::Serialize>(msg: M, encoding: Encoding) -> Response {
127    let (body, content_type) = match encoding {
128        Encoding::Protobuf => (msg.encode_to_vec(), CONTENT_TYPE_PROTOBUF),
129        Encoding::Json => (
130            serde_json::to_vec(&msg).expect("serializing OTLP response to JSON"),
131            CONTENT_TYPE_JSON,
132        ),
133    };
134    Response::builder()
135        .status(StatusCode::OK)
136        .header(header::CONTENT_TYPE, HeaderValue::from_static(content_type))
137        .body(Body::from(body))
138        .expect("building OTLP success response")
139}
140
141async fn logs_handler(
142    Extension(service): Extension<Arc<WebIngestionService>>,
143    headers: HeaderMap,
144    body: bytes::Bytes,
145) -> Response {
146    let encoding = match content_type_encoding(&headers) {
147        Ok(enc) => enc,
148        Err(e) => return e.into_otlp_response(Encoding::Protobuf),
149    };
150    match handler::ingest_logs(service, body, encoding).await {
151        Ok(resp) => success_response(resp, encoding),
152        Err(e) => OtlpHttpError::Otel(e).into_otlp_response(encoding),
153    }
154}
155
156async fn metrics_handler(
157    Extension(service): Extension<Arc<WebIngestionService>>,
158    headers: HeaderMap,
159    body: bytes::Bytes,
160) -> Response {
161    let encoding = match content_type_encoding(&headers) {
162        Ok(enc) => enc,
163        Err(e) => return e.into_otlp_response(Encoding::Protobuf),
164    };
165    match handler::ingest_metrics(service, body, encoding).await {
166        Ok(resp) => success_response(resp, encoding),
167        Err(e) => OtlpHttpError::Otel(e).into_otlp_response(encoding),
168    }
169}
170
171async fn traces_handler(
172    Extension(service): Extension<Arc<WebIngestionService>>,
173    headers: HeaderMap,
174    body: bytes::Bytes,
175) -> Response {
176    let encoding = match content_type_encoding(&headers) {
177        Ok(enc) => enc,
178        Err(e) => return e.into_otlp_response(Encoding::Protobuf),
179    };
180    match handler::ingest_traces(service, body, encoding).await {
181        Ok(resp) => success_response(resp, encoding),
182        Err(e) => OtlpHttpError::Otel(e).into_otlp_response(encoding),
183    }
184}
185
186/// Builds a sub-Router carrying the three OTLP routes plus the shared body-limit and
187/// gzip-decompression layers scoped to those routes (see `ingestion_limits`).
188pub fn otlp_router() -> Router {
189    apply_ingestion_body_limits(
190        Router::new()
191            .route("/ingestion/otlp/v1/logs", post(logs_handler))
192            .route("/ingestion/otlp/v1/metrics", post(metrics_handler))
193            .route("/ingestion/otlp/v1/traces", post(traces_handler)),
194    )
195}
micromegas/servers/otlp.rs

micromegas/servers/
otlp.rs
