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

micromegas/servers/
firehose_cloudwatch_logs.rs
