diff --git a/src/common/meta/ingestion.rs b/src/common/meta/ingestion.rs index 8fdf1f6a98..cca9eb36ba 100644 --- a/src/common/meta/ingestion.rs +++ b/src/common/meta/ingestion.rs @@ -323,11 +323,12 @@ pub struct GCPIngestionResponse { pub enum IngestionRequest<'a> { JSON(&'a web::Bytes), Multi(&'a web::Bytes), + Hec(&'a Vec), + Loki(&'a Vec), GCP(&'a GCPIngestionRequest), KinesisFH(&'a KinesisFHRequest), RUM(&'a web::Bytes), Usage(&'a web::Bytes), - Hec(&'a Vec), } pub enum IngestionData<'a> { diff --git a/src/config/src/meta/self_reporting/usage.rs b/src/config/src/meta/self_reporting/usage.rs index 18e64a5e5b..27aa25eb3a 100644 --- a/src/config/src/meta/self_reporting/usage.rs +++ b/src/config/src/meta/self_reporting/usage.rs @@ -186,6 +186,8 @@ pub enum UsageType { Multi, #[serde(rename = "/logs/_hec")] Hec, + #[serde(rename = "/logs/_loki")] + Loki, #[serde(rename = "/_kinesis_firehose")] KinesisFirehose, #[serde(rename = "/gcp/_sub")] @@ -239,8 +241,9 @@ impl UsageType { self, UsageType::Bulk | UsageType::Json - | UsageType::Hec | UsageType::Multi + | UsageType::Hec + | UsageType::Loki | UsageType::KinesisFirehose | UsageType::GCPSubscription | UsageType::Logs @@ -266,6 +269,7 @@ impl std::fmt::Display for UsageType { UsageType::Json => write!(f, "/logs/_json"), UsageType::Multi => write!(f, "/logs/_multi"), UsageType::Hec => write!(f, "/logs/_hec"), + UsageType::Loki => write!(f, "/logs/_loki"), UsageType::KinesisFirehose => write!(f, "/_kinesis_firehose"), UsageType::GCPSubscription => write!(f, "/gcp/_sub"), UsageType::Logs => write!(f, "/otlp/v1/logs"), diff --git a/src/config/src/router.rs b/src/config/src/router.rs index d3b540cce8..b8fe485f3b 100644 --- a/src/config/src/router.rs +++ b/src/config/src/router.rs @@ -57,7 +57,7 @@ pub const INGESTER_ROUTES: [&str; 12] = [ "/_kinesis_firehose", "/_sub", "/v1/logs", - "/loki/v1/push", + "/loki/api/v1/push", "/ingest/metrics/_json", "/v1/metrics", "/traces", diff --git a/src/handler/http/request/logs/loki.rs b/src/handler/http/request/logs/loki.rs index e705d9b49c..7f3ad5396b 100644 --- a/src/handler/http/request/logs/loki.rs +++ b/src/handler/http/request/logs/loki.rs @@ -51,7 +51,7 @@ use crate::{ (status = 500, description = "Internal Server Error - Server error during log processing", content_type = "text/plain", body = String), ) )] -#[post("/{org_id}/loki/v1/push")] +#[post("/{org_id}/loki/api/v1/push")] pub async fn loki_push( thread_id: web::Data, org_id: web::Path, @@ -183,7 +183,7 @@ mod tests { // Test JSON routing let req = test::TestRequest::post() - .uri("/test_org/loki/v1/push") + .uri("/test_org/loki/api/v1/push") .insert_header(("content-type", "application/json")) .set_payload(create_valid_loki_json()) .to_request(); @@ -192,7 +192,7 @@ mod tests { // Test invalid JSON let req = test::TestRequest::post() - .uri("/test_org/loki/v1/push") + .uri("/test_org/loki/api/v1/push") .insert_header(("content-type", "application/json")) .set_payload("invalid") .to_request(); @@ -201,7 +201,7 @@ mod tests { // Test unsupported content type let req = test::TestRequest::post() - .uri("/test_org/loki/v1/push") + .uri("/test_org/loki/api/v1/push") .insert_header(("content-type", "text/plain")) .set_payload("data") .to_request(); diff --git a/src/service/logs/ingest.rs b/src/service/logs/ingest.rs index 59726b0a99..8866cdaac1 100644 --- a/src/service/logs/ingest.rs +++ b/src/service/logs/ingest.rs @@ -140,6 +140,16 @@ pub async fn ingest( UsageType::Multi, IngestionData::Multi(req), ), + IngestionRequest::Hec(logs) => ( + "/api/org/ingest/logs/_hec", + UsageType::Hec, + IngestionData::JSON(logs), + ), + IngestionRequest::Loki(logs) => ( + "/api/org/ingest/logs/_loki", + UsageType::Loki, + IngestionData::JSON(logs), + ), IngestionRequest::GCP(req) => ( "/api/org/ingest/logs/_gcs", UsageType::GCPSubscription, @@ -168,11 +178,6 @@ pub async fn ingest( IngestionData::JSON(&json_req), ) } - IngestionRequest::Hec(logs) => ( - "/api/org/ingest/logs/_hec", - UsageType::Hec, - IngestionData::JSON(logs), - ), }; let mut stream_status = StreamStatus::new(&stream_name); diff --git a/src/service/logs/loki.rs b/src/service/logs/loki.rs index 0cbb568eee..ac6617a06d 100644 --- a/src/service/logs/loki.rs +++ b/src/service/logs/loki.rs @@ -53,35 +53,24 @@ pub async fn handle_request( }; for (stream_name, records) in streams_data { - let json_bytes = serde_json::to_vec(&records).map_err(|e| { - log::error!( - "[Loki] JSON serialization error for stream '{}' in org '{}': {}", - stream_name, - org_id, - e - ); - LokiError::JsonParse { source: e } - })?; - let json_bytes = actix_web::web::Bytes::from(json_bytes); - logs::ingest::ingest( thread_id, org_id, &stream_name, - IngestionRequest::JSON(&json_bytes), + IngestionRequest::Loki(&records), user_email, None, ) .await .map_err(|e| { log::error!( - "[Loki] Stream '{}' ingestion failed for org '{}': {}", + "[Loki] Stream {} ingestion failed for org {}: {}", stream_name, org_id, e ); LokiError::from(anyhow::anyhow!( - "Stream '{}' ingestion failed: {:?}", + "Stream {} ingestion failed: {:?}", stream_name, e ))