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 super::write_audience::resolve_write_audience;
24use axum::Extension;
25use axum::Router;
26use axum::http::{HeaderMap, StatusCode};
27use axum::middleware;
28use axum::response::Response;
29use axum::routing::post;
30use micromegas_auth::types::{AuthContext, AuthProvider};
31use micromegas_ingestion::web_ingestion_service::WebIngestionService;
32use micromegas_otel_ingestion::{Signal, handler};
33use micromegas_tracing::prelude::*;
34use std::sync::Arc;
35
36async fn firehose_handler(
37    Extension(service): Extension<Arc<WebIngestionService>>,
38    ctx: Option<Extension<AuthContext>>,
39    headers: HeaderMap,
40    body: bytes::Bytes,
41) -> Response {
42    let request_id_header = request_id_from(&headers);
43
44    // Resolved from the `AuthContext` `firehose_auth_middleware` inserts.
45    let audience = resolve_write_audience(ctx.as_ref(), service.default_audience());
46
47    let mut request_id = request_id_header;
48    let envelope = match handler::decode_firehose_envelope(&body, Signal::Metrics) {
49        Ok(e) => e,
50        Err(err) => {
51            error!("firehose decode error (request_id={request_id}): {err}");
52            let status = StatusCode::from_u16(err.http_status())
53                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
54            return firehose_response(status, &request_id, Some(&err.public_message()));
55        }
56    };
57    if request_id.is_empty() {
58        request_id = envelope.request_id.clone(); // header preferred; body requestId is fallback
59    }
60    match handler::ingest_firehose_metrics(service, envelope.records, &audience).await {
61        Ok(()) => firehose_response(StatusCode::OK, &request_id, None),
62        Err(err) => {
63            error!("firehose ingest error (request_id={request_id}): {err}");
64            let status = StatusCode::from_u16(err.http_status())
65                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
66            firehose_response(status, &request_id, Some(&err.public_message()))
67        }
68    }
69}
70
71/// Builds the Firehose sub-router: route + service extension + optional Firehose-auth layer +
72/// shared ingestion body limits (gzip + 20 MiB wire / 300 MiB decompressed).
73///
74/// Deliberately not merged into `protected_app` — it must not sit under the global Bearer
75/// `auth_middleware`, since Firehose can only send its credential via
76/// `X-Amz-Firehose-Access-Key`. Auth is applied only when `auth_provider` is `Some`,
77/// matching every other ingestion route's dev-mode-open behavior.
78pub fn firehose_router(
79    service: Arc<WebIngestionService>,
80    auth_provider: Option<Arc<dyn AuthProvider>>,
81) -> Router {
82    let mut router = Router::new()
83        .route(
84            "/ingestion/otlp/v1/metrics/firehose",
85            post(firehose_handler),
86        )
87        .layer(Extension(service));
88    if let Some(provider) = auth_provider {
89        router = router.layer(middleware::from_fn(move |req, next| {
90            firehose_auth_middleware(provider.clone(), req, next)
91        }));
92    }
93    apply_ingestion_body_limits(router)
94}
micromegas/servers/firehose.rs

micromegas/servers/
firehose.rs
