micromegas/servers/
firehose.rs

1//! Kinesis Data Firehose HTTP Endpoint Delivery route for `telemetry-ingestion-srv`.
2//!
3//! Exposes `POST /ingestion/otlp/v1/metrics/firehose` so a CloudWatch Metric Stream can
4//! push metrics into micromegas as **Metric Stream → Firehose → micromegas**, with no
5//! Lambda, no Kinesis Data Stream, and no collector process in between. Firehose is a
6//! dumb managed pipe: it wraps each delivered record (in OpenTelemetry 1.0.0 output mode,
7//! one-or-more length-delimited OTLP `ExportMetricsServiceRequest` protobuf messages) in
8//! a small JSON envelope and expects a fixed ack shape back.
9//!
10//! Shared Firehose transport plumbing (auth, ack shape, request-id parsing) lives in
11//! `firehose_common` — this module only knows about the metrics-specific decode/ingest
12//! calls.
13//!
14//! Once a record's bytes are extracted from the envelope, `handler::ingest_firehose_metrics`
15//! decodes each length-delimited message in turn, runs it through
16//! `micromegas_otel_ingestion::cloudwatch_metrics::rewrite_cloudwatch_metric_streams` (a
17//! CloudWatch-specific resource rewrite that partitions a matching degenerate resource into
18//! one process per CloudWatch namespace — see that module's docs), and reuses the existing
19//! split/write logic per message.
20
21use super::firehose_common::{firehose_auth_middleware, firehose_response, request_id_from};
22use super::ingestion_limits::apply_ingestion_body_limits;
23use axum::Extension;
24use axum::Router;
25use axum::http::{HeaderMap, StatusCode};
26use axum::middleware;
27use axum::response::Response;
28use axum::routing::post;
29use micromegas_auth::types::AuthProvider;
30use micromegas_ingestion::web_ingestion_service::WebIngestionService;
31use micromegas_otel_ingestion::{Signal, handler};
32use micromegas_tracing::prelude::*;
33use std::sync::Arc;
34
35async fn firehose_handler(
36    Extension(service): Extension<Arc<WebIngestionService>>,
37    headers: HeaderMap,
38    body: bytes::Bytes,
39) -> Response {
40    let mut request_id = request_id_from(&headers);
41    let envelope = match handler::decode_firehose_envelope(&body, Signal::Metrics) {
42        Ok(e) => e,
43        Err(err) => {
44            error!("firehose decode error (request_id={request_id}): {err}");
45            let status = StatusCode::from_u16(err.http_status())
46                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
47            return firehose_response(status, &request_id, Some(&err.public_message()));
48        }
49    };
50    if request_id.is_empty() {
51        request_id = envelope.request_id.clone(); // header preferred; body requestId is fallback
52    }
53    match handler::ingest_firehose_metrics(service, envelope.records).await {
54        Ok(()) => firehose_response(StatusCode::OK, &request_id, None),
55        Err(err) => {
56            error!("firehose ingest error (request_id={request_id}): {err}");
57            let status = StatusCode::from_u16(err.http_status())
58                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
59            firehose_response(status, &request_id, Some(&err.public_message()))
60        }
61    }
62}
63
64/// Builds the Firehose sub-router: route + service extension + optional Firehose-auth
65/// layer + shared ingestion body limits (gzip + 20 MiB wire / 300 MiB decompressed).
66///
67/// Deliberately not merged into `protected_app` — it must not sit under the global Bearer
68/// `auth_middleware`, since Firehose can only send its credential via
69/// `X-Amz-Firehose-Access-Key`. Auth is applied only when `auth_provider` is `Some`,
70/// matching every other ingestion route's dev-mode-open behavior.
71pub fn firehose_router(
72    service: Arc<WebIngestionService>,
73    auth_provider: Option<Arc<dyn AuthProvider>>,
74) -> Router {
75    let mut router = Router::new()
76        .route(
77            "/ingestion/otlp/v1/metrics/firehose",
78            post(firehose_handler),
79        )
80        .layer(Extension(service));
81    if let Some(provider) = auth_provider {
82        router = router.layer(middleware::from_fn(move |req, next| {
83            firehose_auth_middleware(provider.clone(), req, next)
84        }));
85    }
86    apply_ingestion_body_limits(router)
87}
micromegas/servers/firehose.rs

micromegas/servers/
firehose.rs
