micromegas/servers/
firehose_cloudwatch_logs.rs

1//! Kinesis Data Firehose HTTP Endpoint Delivery route for CloudWatch Logs subscription
2//! filters.
3//!
4//! Exposes `POST /ingestion/cloudwatch/v1/logs/firehose` so CloudWatch Logs can push logs
5//! into micromegas as **CloudWatch Logs → subscription filter → Firehose → micromegas**,
6//! with no intermediate consumer. Unlike the metrics Firehose route, CloudWatch Logs
7//! subscription-filter delivery has exactly one proprietary record format — there is no
8//! OTLP framing on the wire, only in how micromegas happens to store the result. The route
9//! is therefore named `cloudwatch/...` rather than `otlp/...`, to avoid misleadingly
10//! implying the client sends OTLP.
11//!
12//! Shared Firehose transport plumbing (auth, ack shape, request-id parsing) lives in
13//! `firehose_common`.
14
15use super::firehose_common::{firehose_auth_middleware, firehose_response, request_id_from};
16use super::ingestion_limits::apply_ingestion_body_limits;
17use axum::Extension;
18use axum::Router;
19use axum::http::{HeaderMap, StatusCode};
20use axum::middleware;
21use axum::response::Response;
22use axum::routing::post;
23use micromegas_auth::types::AuthProvider;
24use micromegas_ingestion::web_ingestion_service::WebIngestionService;
25use micromegas_otel_ingestion::{Signal, cloudwatch_logs, handler};
26use micromegas_tracing::prelude::*;
27use std::sync::Arc;
28
29async fn cloudwatch_logs_firehose_handler(
30    Extension(service): Extension<Arc<WebIngestionService>>,
31    headers: HeaderMap,
32    body: bytes::Bytes,
33) -> Response {
34    let mut request_id = request_id_from(&headers);
35    let envelope = match handler::decode_firehose_envelope(&body, Signal::Logs) {
36        Ok(e) => e,
37        Err(err) => {
38            error!("cloudwatch logs firehose decode error: {err}");
39            let status = StatusCode::from_u16(err.http_status())
40                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
41            return firehose_response(status, &request_id, Some(&err.public_message()));
42        }
43    };
44    if request_id.is_empty() {
45        request_id = envelope.request_id.clone();
46    }
47    match cloudwatch_logs::ingest_cloudwatch_logs_firehose(service, envelope.records).await {
48        Ok(()) => firehose_response(StatusCode::OK, &request_id, None),
49        Err(err) => {
50            error!("cloudwatch logs firehose ingest error: {err}");
51            let status = StatusCode::from_u16(err.http_status())
52                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
53            firehose_response(status, &request_id, Some(&err.public_message()))
54        }
55    }
56}
57
58/// Builds the CloudWatch Logs Firehose sub-router: route + service extension + optional
59/// Firehose-auth layer + shared ingestion body limits (gzip + 20 MiB wire / 300 MiB
60/// decompressed).
61///
62/// Deliberately not merged into `protected_app` — same reasoning as the metrics Firehose
63/// route: Firehose can only send its credential via `X-Amz-Firehose-Access-Key`, not
64/// `Authorization: Bearer`. Auth is applied only when `auth_provider` is `Some`, matching
65/// every other ingestion route's dev-mode-open behavior.
66pub fn firehose_router(
67    service: Arc<WebIngestionService>,
68    auth_provider: Option<Arc<dyn AuthProvider>>,
69) -> Router {
70    let mut router = Router::new()
71        .route(
72            "/ingestion/cloudwatch/v1/logs/firehose",
73            post(cloudwatch_logs_firehose_handler),
74        )
75        .layer(Extension(service));
76    if let Some(provider) = auth_provider {
77        router = router.layer(middleware::from_fn(move |req, next| {
78            firehose_auth_middleware(provider.clone(), req, next)
79        }));
80    }
81    apply_ingestion_body_limits(router)
82}
micromegas/servers/firehose_cloudwatch_logs.rs

micromegas/servers/
firehose_cloudwatch_logs.rs
