diff --git a/scripts/ingest.ps1 b/scripts/ingest.ps1 index e3fd9d59c..b1a706578 100644 --- a/scripts/ingest.ps1 +++ b/scripts/ingest.ps1 @@ -434,8 +434,7 @@ function Setup-Collector { " compression: none", " headers:", " X-API-Key: $apiKeyValue", - " X-P-Stream: $streamNameValue", - " X-P-Log-Source: otel-metrics" + " X-P-Stream: $streamNameValue" ) if (-not [string]::IsNullOrWhiteSpace($TenantId)) { diff --git a/scripts/ingest.sh b/scripts/ingest.sh index 862500118..7eeff6ca9 100755 --- a/scripts/ingest.sh +++ b/scripts/ingest.sh @@ -409,7 +409,6 @@ exporters: headers: X-API-Key: '$api_key_yaml' X-P-Stream: '$stream_name_yaml' - X-P-Log-Source: otel-metrics ${tenant_header} service: diff --git a/src/handlers/http/ingest.rs b/src/handlers/http/ingest.rs index a98e8ab3a..f10c8adf9 100644 --- a/src/handlers/http/ingest.rs +++ b/src/handlers/http/ingest.rs @@ -219,10 +219,11 @@ pub async fn ingest_internal_stream( Ok(()) } -// Common validation and setup for OTEL ingestion +// Common validation and setup for OTEL ingestion. The endpoint determines the log source, so +// clients do not need to provide x-p-log-source. pub async fn setup_otel_stream( req: &HttpRequest, - expected_log_source: LogSource, + log_source: LogSource, known_fields: &[&str], telemetry_type: TelemetryType, dataset_tags: Vec, @@ -232,18 +233,6 @@ pub async fn setup_otel_stream( return Err(PostError::Header(ParseHeaderError::MissingStreamName)); }; - let Some(log_source) = req.headers().get(LOG_SOURCE_KEY) else { - return Err(PostError::Header(ParseHeaderError::MissingLogSource)); - }; - - let log_source = LogSource::from(log_source.to_str().unwrap()); - if log_source != expected_log_source { - return Err(PostError::IncorrectLogSource( - expected_log_source, - telemetry_type.to_string(), - )); - } - let stream_name = stream_name.to_str().unwrap().to_owned(); let log_source_entry = LogSourceEntry::new( @@ -531,8 +520,6 @@ pub enum PostError { OtelNotSupported(String), #[error("The stream {0} is reserved for internal use and cannot be ingested into")] InternalStream(String), - #[error(r#"Please use "x-p-log-source: {0}" for ingesting otel {1} data"#)] - IncorrectLogSource(LogSource, String), #[error("Ingestion is not allowed in Query mode")] IngestionNotAllowed, #[error("Missing field for time partition in json: {0}")] @@ -563,7 +550,6 @@ impl actix_web::ResponseError for PostError { | Header(_) | Invalid(_) | InternalStream(_) - | IncorrectLogSource(_, _) | IngestionNotAllowed | MissingTimePartition(_) | KnownFormat(_) diff --git a/src/handlers/http/modal/utils/ingest_utils.rs b/src/handlers/http/modal/utils/ingest_utils.rs index d3fcba7ac..f84b1e30a 100644 --- a/src/handlers/http/modal/utils/ingest_utils.rs +++ b/src/handlers/http/modal/utils/ingest_utils.rs @@ -87,7 +87,7 @@ pub async fn process_otel_content( log_source: &LogSource, telemetry_type: TelemetryType, ) -> Result<(), PostError> { - let p_custom_fields = get_custom_fields_from_header(req); + let p_custom_fields = get_otel_custom_fields_from_header(req, log_source); let content_type = req .headers() .get("Content-Type") @@ -467,6 +467,15 @@ pub fn get_custom_fields_from_header(req: &HttpRequest) -> HashMap HashMap { + let mut custom_fields = get_custom_fields_from_header(req); + custom_fields.insert(FORMAT_KEY.to_string(), log_source.to_string()); + custom_fields +} + fn verify_dataset_fields_count( stream_name: &str, tenant_id: &Option, @@ -583,6 +592,19 @@ mod tests { assert_eq!(custom_fields.get(FORMAT_KEY).unwrap(), "otel-logs"); } + #[test] + fn test_otel_custom_fields_use_endpoint_log_source() { + let req = TestRequest::default().to_http_request(); + let custom_fields = get_otel_custom_fields_from_header(&req, &LogSource::OtelMetrics); + assert_eq!(custom_fields.get(FORMAT_KEY).unwrap(), "otel-metrics"); + + let req = TestRequest::default() + .insert_header((LOG_SOURCE_KEY, "otel-traces")) + .to_http_request(); + let custom_fields = get_otel_custom_fields_from_header(&req, &LogSource::OtelMetrics); + assert_eq!(custom_fields.get(FORMAT_KEY).unwrap(), "otel-metrics"); + } + #[test] fn test_get_custom_fields_empty_header_after_prefix() { let req = TestRequest::default() diff --git a/src/otel_generator.rs b/src/otel_generator.rs index 332609450..048d20a1b 100644 --- a/src/otel_generator.rs +++ b/src/otel_generator.rs @@ -334,7 +334,6 @@ async fn send_batch(endpoint: &str, auth: &str, tenant_id: Option<&str>, batch: send_signal( endpoint, "traces", - "otel-traces", "otel-demo-traces", auth, tenant_id, @@ -343,7 +342,6 @@ async fn send_batch(endpoint: &str, auth: &str, tenant_id: Option<&str>, batch: send_signal( endpoint, "metrics", - "otel-metrics", "otel-demo-metrics", auth, tenant_id, @@ -352,7 +350,6 @@ async fn send_batch(endpoint: &str, auth: &str, tenant_id: Option<&str>, batch: send_signal( endpoint, "logs", - "otel-logs", "otel-demo-logs", auth, tenant_id, @@ -370,7 +367,6 @@ async fn send_batch(endpoint: &str, auth: &str, tenant_id: Option<&str>, batch: async fn send_signal( endpoint: &str, signal: &str, - log_source: &str, stream: &str, auth: &str, tenant_id: Option<&str>, @@ -381,7 +377,6 @@ async fn send_signal( .post(format!("{endpoint}/v1/{signal}")) .header(AUTHORIZATION, auth) .header(CONTENT_TYPE, "application/json") - .header("X-P-Log-Source", log_source) .header("X-P-Stream", stream) .body(body); if let Some(tenant_id) = tenant_id { diff --git a/src/telemetry.rs b/src/telemetry.rs index 431d4938a..888dfa484 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -48,7 +48,7 @@ const OTEL_EXPORTER_OTLP_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_PROTOCOL"; /// Ignored when endpoint is \`stdout\`. /// - \`OTEL_EXPORTER_OTLP_HEADERS\` — comma-separated \`key=value\` pairs forwarded /// as gRPC metadata or HTTP headers, e.g. -/// \`authorization=Basic ,x-p-stream=my-stream,x-p-log-source=otel-traces\` +/// \`authorization=Basic ,x-p-stream=my-stream\` /// /// Returns \`None\` when \`OTEL_EXPORTER_OTLP_ENDPOINT\` or `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` is not set (OTEL disabled). /// The caller must call \`provider.shutdown()\` before process exit. diff --git a/src/utils/header_parsing.rs b/src/utils/header_parsing.rs index 787f3d138..83b9c340a 100644 --- a/src/utils/header_parsing.rs +++ b/src/utils/header_parsing.rs @@ -33,8 +33,6 @@ pub enum ParseHeaderError { SeperatorInValue(char), #[error("Stream name not found in header [x-p-stream]")] MissingStreamName, - #[error("Log source not found in header [x-p-log-source]")] - MissingLogSource, #[error("Tenant id not found in header [tenant]")] MissingTenantId, #[error("Invalid tenant id found in header [tenant]")]