From 5243e2241ff85a710c3dbebdbc31fff12bc5a943 Mon Sep 17 00:00:00 2001 From: Nikhil Sinha Date: Thu, 29 Oct 2026 08:37:29 +0530 Subject: [PATCH] feat: use default stream name if header missing in ingestion x-p-stream no longer requires to be mandatory header if found in header, use that for ingestion else, use default stream name arbitrary logs - default_logs otel logs - default_otel_logs otel traces - default_otel_traces otel metrics - default_otel_metrics with this change, otel collector config simplifies as one exporter config works instead of config for each telemetry type --- src/handlers/http/ingest.rs | 20 +++++----- src/handlers/http/middleware.rs | 65 +++++++++++++++++++++++++++------ src/handlers/mod.rs | 38 +++++++++++++++++++ 3 files changed, 101 insertions(+), 22 deletions(-) diff --git a/src/handlers/http/ingest.rs b/src/handlers/http/ingest.rs index 2f79ba105..dfc1141af 100644 --- a/src/handlers/http/ingest.rs +++ b/src/handlers/http/ingest.rs @@ -37,8 +37,9 @@ use crate::handlers::http::modal::utils::ingest_utils::{ ingest_helper, process_otel_content, validate_stream_for_ingestion, }; use crate::handlers::{ - DatasetTag, EXTRACT_LOG_KEY, LOG_SOURCE_KEY, STREAM_NAME_HEADER_KEY, TELEMETRY_TYPE_KEY, - TelemetryType, + DEFAULT_OTEL_LOGS_STREAM_NAME, DEFAULT_OTEL_METRICS_STREAM_NAME, + DEFAULT_OTEL_TRACES_STREAM_NAME, DEFAULT_STREAM_NAME, DatasetTag, EXTRACT_LOG_KEY, + LOG_SOURCE_KEY, TELEMETRY_TYPE_KEY, TelemetryType, stream_name_from_headers, }; use crate::metadata::SchemaVersion; use crate::metastore::MetastoreError; @@ -70,11 +71,8 @@ pub async fn ingest( req: HttpRequest, Json(json): Json, ) -> Result { - let Some(stream_name) = req.headers().get(STREAM_NAME_HEADER_KEY) else { - return Err(PostError::Header(ParseHeaderError::MissingStreamName)); - }; let tenant_id = get_tenant_id_from_request(&req); - let stream_name = stream_name.to_str().unwrap().to_owned(); + let stream_name = stream_name_from_headers(req.headers(), DEFAULT_STREAM_NAME)?; let internal_stream_names = PARSEABLE.streams.list_internal_streams(&tenant_id); if internal_stream_names.contains(&stream_name) { return Err(PostError::InternalStream(stream_name)); @@ -229,11 +227,13 @@ pub async fn setup_otel_stream( dataset_tags: Vec, dataset_labels: Vec, ) -> Result<(String, LogSource, LogSourceEntry, Option), PostError> { - let Some(stream_name) = req.headers().get(STREAM_NAME_HEADER_KEY) else { - return Err(PostError::Header(ParseHeaderError::MissingStreamName)); + let default_stream_name = match log_source { + LogSource::OtelLogs => DEFAULT_OTEL_LOGS_STREAM_NAME, + LogSource::OtelMetrics => DEFAULT_OTEL_METRICS_STREAM_NAME, + LogSource::OtelTraces => DEFAULT_OTEL_TRACES_STREAM_NAME, + _ => DEFAULT_STREAM_NAME, }; - - let stream_name = stream_name.to_str().unwrap().to_owned(); + let stream_name = stream_name_from_headers(req.headers(), default_stream_name)?; let log_source_entry = LogSourceEntry::new( log_source.clone(), diff --git a/src/handlers/http/middleware.rs b/src/handlers/http/middleware.rs index 13332968d..579090eb4 100644 --- a/src/handlers/http/middleware.rs +++ b/src/handlers/http/middleware.rs @@ -34,7 +34,7 @@ use ulid::Ulid; use crate::{ handlers::{ AUTHORIZATION_KEY, KINESIS_COMMON_ATTRIBUTES_KEY, LOG_SOURCE_KEY, LOG_SOURCE_KINESIS, - STREAM_NAME_HEADER_KEY, TENANT_ID, + STREAM_NAME_HEADER_KEY, TENANT_ID, default_stream_name_for_path, http::{ingest::PostError, modal::OIDC_CLIENT, rbac::RBACError}, }, option::Mode, @@ -70,7 +70,7 @@ pub struct CommonAttributes { #[serde(rename = "Authorization")] authorization: String, #[serde(rename = "X-P-Stream")] - pub x_p_stream: String, + pub x_p_stream: Option, } pub trait RouteExt { @@ -300,7 +300,7 @@ where } } -/// Extract Kinesis Firehose headers (Authorization, X-P-Stream) from +/// Extract Kinesis Firehose headers (Authorization and optional X-P-Stream) from /// the x-amz-firehose-common-attributes custom header when present. fn extract_kinesis_headers(req: &mut ServiceRequest) { if let Some(kinesis_common_attributes) = @@ -309,15 +309,22 @@ fn extract_kinesis_headers(req: &mut ServiceRequest) { && let Ok(message) = serde_json::from_str::(attribute_value) && let Ok(auth_value) = header::HeaderValue::from_str(&message.common_attributes.authorization) - && let Ok(stream_name_key) = - header::HeaderValue::from_str(&message.common_attributes.x_p_stream) { + let stream_name_value = match message.common_attributes.x_p_stream { + Some(stream_name) => match header::HeaderValue::from_str(&stream_name) { + Ok(value) => Some(value), + Err(_) => return, + }, + None => None, + }; req.headers_mut() .insert(HeaderName::from_static(AUTHORIZATION_KEY), auth_value); - req.headers_mut().insert( - HeaderName::from_static(STREAM_NAME_HEADER_KEY), - stream_name_key, - ); + if let Some(stream_name_value) = stream_name_value { + req.headers_mut().insert( + HeaderName::from_static(STREAM_NAME_HEADER_KEY), + stream_name_value, + ); + } req.headers_mut().insert( HeaderName::from_static(LOG_SOURCE_KEY), header::HeaderValue::from_static(LOG_SOURCE_KINESIS), @@ -604,9 +611,13 @@ pub fn auth_resource_context( } else if let Some(stream) = stream { creds.map(|key| Users.authorize(key, action, Some(stream), None)) } else { - if let Some(stream_name) = req.headers().get(STREAM_NAME_HEADER_KEY) { - stream = Some(stream_name.to_str().unwrap()); - } + stream = req + .headers() + .get(STREAM_NAME_HEADER_KEY) + .and_then(|stream_name| stream_name.to_str().ok()) + .or_else(|| { + (action == Action::Ingest).then(|| default_stream_name_for_path(req.path())) + }); creds.map(|key| Users.authorize(key, action, stream, None)) } } @@ -991,4 +1002,34 @@ mod api_key_header_tests { assert_eq!(req.headers().get(TENANT_ID).unwrap(), "acme"); } + + #[test] + fn kinesis_common_attributes_allow_missing_stream_name() { + let common_attributes = r#"{"commonAttributes":{"Authorization":"Basic token"}}"#; + let mut req = actix_test::TestRequest::default() + .insert_header((KINESIS_COMMON_ATTRIBUTES_KEY, common_attributes)) + .to_srv_request(); + + extract_kinesis_headers(&mut req); + + assert_eq!(req.headers().get(AUTHORIZATION_KEY).unwrap(), "Basic token"); + assert_eq!( + req.headers().get(LOG_SOURCE_KEY).unwrap(), + LOG_SOURCE_KINESIS + ); + assert!(req.headers().get(STREAM_NAME_HEADER_KEY).is_none()); + } + + #[test] + fn kinesis_common_attributes_preserve_stream_name() { + let common_attributes = + r#"{"commonAttributes":{"Authorization":"Basic token","X-P-Stream":"custom"}}"#; + let mut req = actix_test::TestRequest::default() + .insert_header((KINESIS_COMMON_ATTRIBUTES_KEY, common_attributes)) + .to_srv_request(); + + extract_kinesis_headers(&mut req); + + assert_eq!(req.headers().get(STREAM_NAME_HEADER_KEY).unwrap(), "custom"); + } } diff --git a/src/handlers/mod.rs b/src/handlers/mod.rs index 857d4b660..c4b65a228 100644 --- a/src/handlers/mod.rs +++ b/src/handlers/mod.rs @@ -19,14 +19,21 @@ use std::collections::HashSet; use std::fmt::Display; +use actix_web::http::header::HeaderMap; use serde::{Deserialize, Serialize}; use tracing::warn; +use crate::utils::header_parsing::ParseHeaderError; + pub mod airplane; pub mod http; pub mod livetail; pub const STREAM_NAME_HEADER_KEY: &str = "x-p-stream"; +pub const DEFAULT_STREAM_NAME: &str = "default_logs"; +pub const DEFAULT_OTEL_LOGS_STREAM_NAME: &str = "default_otel_logs"; +pub const DEFAULT_OTEL_METRICS_STREAM_NAME: &str = "default_otel_metrics"; +pub const DEFAULT_OTEL_TRACES_STREAM_NAME: &str = "default_otel_traces"; pub const LOG_SOURCE_KEY: &str = "x-p-log-source"; pub const EXTRACT_LOG_KEY: &str = "x-p-extract-log"; pub const TIME_PARTITION_KEY: &str = "x-p-time-partition"; @@ -56,6 +63,37 @@ pub const KINESIS_COMMON_ATTRIBUTES_KEY: &str = "x-amz-firehose-common-attribute pub const CONTENT_TYPE_JSON: &str = "application/json"; pub const CONTENT_TYPE_PROTOBUF: &str = "application/x-protobuf"; +pub fn stream_name_from_headers( + headers: &HeaderMap, + default_stream_name: &str, +) -> Result { + headers + .get(STREAM_NAME_HEADER_KEY) + .map(|stream_name| { + stream_name + .to_str() + .map(str::to_owned) + .map_err(|_| ParseHeaderError::InvalidValue) + }) + .unwrap_or_else(|| Ok(default_stream_name.to_owned())) +} + +pub fn default_stream_name_for_path(path: &str) -> &'static str { + default_stream_name_for_ingest_path(path).unwrap_or(DEFAULT_STREAM_NAME) +} + +pub fn default_stream_name_for_ingest_path(path: &str) -> Option<&'static str> { + match path.trim_end_matches('/') { + path if path.ends_with("/api/v1/ingest") => Some(DEFAULT_STREAM_NAME), + path if path.ends_with("/v1/logs") => Some(DEFAULT_OTEL_LOGS_STREAM_NAME), + path if path.ends_with("/v1/metrics") || path.ends_with("/v1/prometheus/write") => { + Some(DEFAULT_OTEL_METRICS_STREAM_NAME) + } + path if path.ends_with("/v1/traces") => Some(DEFAULT_OTEL_TRACES_STREAM_NAME), + _ => None, + } +} + #[derive(Debug, Clone, Copy, Serialize, Deserialize, Default, PartialEq, Eq)] #[serde(rename_all = "lowercase")] pub enum TelemetryType {