micromegas/servers/
webhook.rs

1//! Generic header-described webhook ingestion route for `telemetry-ingestion-srv`.
2//!
3//! Exposes `POST /ingestion/webhook`. Any header-capable webhook producer (GitLab,
4//! GitHub, generic SaaS) can report directly to micromegas: three `X-Micromegas-*`
5//! request headers synthesize an OTLP `Resource` + scope name, and the request body
6//! becomes a single log record's body, stored verbatim when it is valid UTF-8 (the
7//! common case: JSON payloads from GitLab/GitHub/etc.). There is no header to describe
8//! an alternate codec, so a non-UTF8 body is stored via lossy UTF-8 conversion (invalid
9//! byte sequences become U+FFFD) rather than rejected or stored as opaque binary. The
10//! synthetic request is fed into the existing OTLP logs split/write path
11//! (`handler::ingest_webhook`) — no new identity, block, or write logic.
12//!
13//! The body is opaque JSON/text: this endpoint does not negotiate `Content-Type` or
14//! parse the body at all (contrast with `otlp_router`, which must switch proto/JSON
15//! decoding).
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::handler;
28use micromegas_otel_ingestion::proto::{AnyValue, KeyValue, any_value};
29use micromegas_tracing::prelude::*;
30use std::sync::Arc;
31
32const HEADER_SERVICE_NAME: &str = "X-Micromegas-Service-Name";
33const HEADER_SERVICE_NAMESPACE: &str = "X-Micromegas-Service-Namespace";
34const HEADER_TARGET: &str = "X-Micromegas-Target";
35
36/// Reads one header and, if present and decodable as ASCII/UTF-8, pushes a
37/// `KeyValue { key: otel_key, value: StringValue(header value) }` onto `attrs`.
38/// Missing or non-decodable headers are silently skipped — same as an OTLP resource
39/// that omits an attribute.
40fn push_attr_from_header(
41    attrs: &mut Vec<KeyValue>,
42    headers: &HeaderMap,
43    header_name: &str,
44    otel_key: &str,
45) {
46    let Some(value) = headers.get(header_name) else {
47        return;
48    };
49    let Ok(value) = value.to_str() else {
50        return;
51    };
52    attrs.push(KeyValue {
53        key: otel_key.to_string(),
54        key_strindex: 0,
55        value: Some(AnyValue {
56            value: Some(any_value::Value::StringValue(value.to_string())),
57        }),
58    });
59}
60
61/// Reads `X-Micromegas-Target`, returning an empty string when absent or non-decodable.
62fn target_from_header(headers: &HeaderMap) -> String {
63    headers
64        .get(HEADER_TARGET)
65        .and_then(|v| v.to_str().ok())
66        .unwrap_or("")
67        .to_string()
68}
69
70/// Canonicalizes the full incoming header set into stable bytes for `block_id` hashing
71/// (`handler::ingest_webhook`'s `header_hash_input`).
72///
73/// Only the 3 recognized `X-Micromegas-*` headers become OTel resource attrs, so without
74/// this, any other header a producer sends (a delivery-id, a signature, an event-type
75/// header) would have zero influence on `block_id` — two deliveries with the same body but
76/// different unrecognized headers would collide and dedup as if they were retries of the
77/// same delivery. Folding in the raw header set closes that gap, at the cost of also
78/// widening it the other way: a genuine retry of the same delivery that picks up a new
79/// hop-by-hop header along the way (e.g. a proxy stamping a fresh `Date` or request-id) no
80/// longer dedups. That tradeoff is accepted here — see the "Webhook ingestion" docs section.
81///
82/// Headers are lowercased and sorted by (name, value) so the hash doesn't depend on wire
83/// order, which servers/proxies are free to reshuffle.
84fn canonical_header_bytes(headers: &HeaderMap) -> Vec<u8> {
85    let mut pairs: Vec<(String, &[u8])> = headers
86        .iter()
87        .map(|(name, value)| (name.as_str().to_ascii_lowercase(), value.as_bytes()))
88        .collect();
89    pairs.sort();
90
91    let mut buf = Vec::new();
92    for (name, value) in pairs {
93        buf.extend_from_slice(name.as_bytes());
94        buf.push(0);
95        buf.extend_from_slice(value);
96        buf.push(0);
97    }
98    buf
99}
100
101fn build_error_response(status: StatusCode, message: &str, retryable: bool) -> Response {
102    let mut response = Response::builder()
103        .status(status)
104        .header(
105            header::CONTENT_TYPE,
106            HeaderValue::from_static("text/plain; charset=utf-8"),
107        )
108        .body(Body::from(message.to_string()))
109        .expect("building webhook error response");
110    if retryable && let Ok(value) = HeaderValue::from_str(&RETRY_AFTER_SECONDS.to_string()) {
111        response.headers_mut().insert(header::RETRY_AFTER, value);
112    }
113    response
114}
115
116async fn webhook_handler(
117    Extension(service): Extension<Arc<WebIngestionService>>,
118    ctx: Option<Extension<AuthContext>>,
119    headers: HeaderMap,
120    body: bytes::Bytes,
121) -> Response {
122    if body.is_empty() {
123        return build_error_response(StatusCode::BAD_REQUEST, "empty body", false);
124    }
125
126    let audience = resolve_write_audience(ctx.as_ref(), service.default_audience());
127
128    let mut resource_attrs = Vec::new();
129    push_attr_from_header(
130        &mut resource_attrs,
131        &headers,
132        HEADER_SERVICE_NAME,
133        "service.name",
134    );
135    push_attr_from_header(
136        &mut resource_attrs,
137        &headers,
138        HEADER_SERVICE_NAMESPACE,
139        "service.namespace",
140    );
141    let target = target_from_header(&headers);
142    let header_hash_input = canonical_header_bytes(&headers);
143
144    match handler::ingest_webhook(
145        service,
146        resource_attrs,
147        target,
148        body,
149        &header_hash_input,
150        &audience,
151    )
152    .await
153    {
154        Ok(()) => Response::builder()
155            .status(StatusCode::OK)
156            .body(Body::empty())
157            .expect("building webhook success response"),
158        Err(err) => {
159            let retryable = err.is_retryable();
160            let status = StatusCode::from_u16(err.http_status())
161                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
162            // Detailed error (includes raw sqlx / object-store messages) is logged
163            // server-side; only the sanitized public form goes to the client.
164            error!("webhook error: {err}");
165            build_error_response(status, &err.public_message(), retryable)
166        }
167    }
168}
169
170/// Builds a sub-Router carrying the webhook route plus the shared body-limit and
171/// gzip-decompression layers scoped to it (see `ingestion_limits`).
172pub fn webhook_router() -> Router {
173    apply_ingestion_body_limits(Router::new().route("/ingestion/webhook", post(webhook_handler)))
174}
micromegas/servers/webhook.rs

micromegas/servers/
webhook.rs
