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

micromegas/servers/
webhook.rs
