From 06dfea980908ba3b255731dfa88c9dc0fc8ea17c Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Mon, 23 Jun 2025 09:41:06 +0800 Subject: [PATCH] fix: improve Loki performance (#7301) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### **User description** - [x] Change Loki API from `/api/{org}/loki/vi/push` to `/api/{org}/loki/api/v1/push` - [x] Improve Loki API performance - [x] Correct usage report and metrics path for Loki API ___ ### **PR Type** Enhancement ___ ### **Description** - Add `Loki` ingestion variant to enums - Update Loki API route to `/loki/api/v1/push` - Extend ingest service for Loki and HEC - Adjust HTTP handler and tests for new route ___ ### **Changes walkthrough** 📝
Relevant files
Enhancement
ingestion.rs
Add Loki request enum variant                                                       
src/common/meta/ingestion.rs
  • Added Hec and Loki variants to IngestionRequest
  • Removed duplicate Hec entry and reordered variants
    		• +2/-1     
    usage.rs
    Support Loki in UsageType enum                                                     
    src/config/src/meta/self_reporting/usage.rs
  • Inserted Loki variant with serde rename
  • Included Loki in is_ingestion checks
  • Added Display case for Loki
    		• +5/-1     
    loki.rs
    Change Loki HTTP API path and tests                                           
    src/handler/http/request/logs/loki.rs - Changed HTTP route attribute path - Updated test URIs to new route
    		+4/-4     
    ingest.rs
    Extend ingest service for Loki and HEC                                     
    src/service/logs/ingest.rs
  • Added match arms for Hec and Loki ingestion
  • Removed duplicate Hec handling block
    		• +10/-5   
    loki.rs
    Use Loki variant in ingest call                                                   
    src/service/logs/loki.rs
  • Switched to IngestionRequest::Loki usage
  • Removed intermediate JSON-to-Bytes conversion
  • Simplified log error messages formatting
    		• +3/-14   
    Configuration changes
    router.rs
    Update Loki ingester route path                                                   
    src/config/src/router.rs
  • Updated Loki route path in INGESTER_ROUTES
  • Changed "/loki/v1/push" to "/loki/api/v1/push"
    		• +1/-1     
    ___ >
    Need help?
  • Type /help how to ... in the comments thread for any questions about PR-Agent usage.
  • Check out the documentation for more information.
    • --- src/common/meta/ingestion.rs | 3 ++- src/config/src/meta/self_reporting/usage.rs | 6 +++++- src/config/src/router.rs | 2 +- src/handler/http/request/logs/loki.rs | 8 ++++---- src/service/logs/ingest.rs | 15 ++++++++++----- src/service/logs/loki.rs | 17 +++-------------- 6 files changed, 25 insertions(+), 26 deletions(-) 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 ))