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

micromegas/servers/
otlp.rs
