remove x-p-log-source for otel ingestion, derive from endpoint (#1773)

Author: nikhilsinhaparseable

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)) {

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:

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<DatasetTag>,
@@ -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(_)

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<String, Strin
     p_custom_fields
 }
 
+pub fn get_otel_custom_fields_from_header(
+    req: &HttpRequest,
+    log_source: &LogSource,
+) -> HashMap<String, String> {
+    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<String>,
@@ -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()

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<T: Serialize + ?Sized>(
     endpoint: &str,
     signal: &str,
-    log_source: &str,
     stream: &str,
     auth: &str,
     tenant_id: Option<&str>,
@@ -381,7 +377,6 @@ async fn send_signal<T: Serialize + ?Sized>(
         .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 {

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 <token>,x-p-stream=my-stream,x-p-log-source=otel-traces\`
+///   \`authorization=Basic <token>,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.

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]")]