mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-06 00:09:22 +00:00
fix: improve Loki performance (#7301)
### **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** 📝
<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><table>
<tr>
<td>
<details>
<summary><strong>ingestion.rs</strong><dd><code>Add Loki request enum
variant</code>
</dd></summary>
<hr>
src/common/meta/ingestion.rs
<li>Added <code>Hec</code> and <code>Loki</code> variants to
<code>IngestionRequest</code><br> <li> Removed duplicate
<code>Hec</code> entry and reordered variants
</details>
</td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7301/files#diff-5effcbeb41f448c6f75b1f17baa683a02d668865eb09f898633797cd1327e3a2">+2/-1</a>
</td>
</tr>
<tr>
<td>
<details>
<summary><strong>usage.rs</strong><dd><code>Support Loki in UsageType
enum</code>
</dd></summary>
<hr>
src/config/src/meta/self_reporting/usage.rs
<li>Inserted <code>Loki</code> variant with serde rename<br> <li>
Included <code>Loki</code> in <code>is_ingestion</code> checks<br> <li>
Added Display case for <code>Loki</code>
</details>
</td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7301/files#diff-b15a6adcb163dc27993c83528105a6726c44c86d8a84c5cf5d1d382c6ec8cc29">+5/-1</a>
</td>
</tr>
<tr>
<td>
<details>
<summary><strong>loki.rs</strong><dd><code>Change Loki HTTP API path and
tests</code>
</dd></summary>
<hr>
src/handler/http/request/logs/loki.rs
- Changed HTTP route attribute path
- Updated test URIs to new route
</details>
</td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7301/files#diff-34691872632b619be8aadf255104f0e004fcf701dba98f88e231de5966eed099">+4/-4</a>
</td>
</tr>
<tr>
<td>
<details>
<summary><strong>ingest.rs</strong><dd><code>Extend ingest service for
Loki and HEC</code>
</dd></summary>
<hr>
src/service/logs/ingest.rs
<li>Added match arms for <code>Hec</code> and <code>Loki</code>
ingestion<br> <li> Removed duplicate <code>Hec</code> handling block
</details>
</td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7301/files#diff-e62118e45be183266d39ffa08491f255017954235ff0f3d6937683e314da406f">+10/-5</a>
</td>
</tr>
<tr>
<td>
<details>
<summary><strong>loki.rs</strong><dd><code>Use Loki variant in ingest
call</code>
</dd></summary>
<hr>
src/service/logs/loki.rs
<li>Switched to <code>IngestionRequest::Loki</code> usage<br> <li>
Removed intermediate JSON-to-Bytes conversion<br> <li> Simplified log
error messages formatting
</details>
</td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7301/files#diff-a970c8c4785d4e4d878218615b17eec0cf4b4976121766b002c2b2f76371bfe6">+3/-14</a>
</td>
</tr>
</table></td></tr><tr><td><strong>Configuration
changes</strong></td><td><table>
<tr>
<td>
<details>
<summary><strong>router.rs</strong><dd><code>Update Loki ingester route
path</code>
</dd></summary>
<hr>
src/config/src/router.rs
<li>Updated Loki route path in <code>INGESTER_ROUTES</code><br> <li>
Changed "/loki/v1/push" to "/loki/api/v1/push"
</details>
</td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7301/files#diff-5227b263a6c37a040ee1e7482b6903ead2186dd71090ee70bdaf1c3b894487ad">+1/-1</a>
</td>
</tr>
</table></td></tr></tr></tbody></table>
___
> <details> <summary> Need help?</summary><li>Type <code>/help how to
...</code> in the comments thread for any questions about PR-Agent
usage.</li><li>Check out the <a
href="https://qodo-merge-docs.qodo.ai/usage-guide/">documentation</a>
for more information.</li></details>
This commit is contained in:
Hengfei Yang
@@ -323,11 +323,12 @@ pub struct GCPIngestionResponse {
|
||||
pub enum IngestionRequest<'a> {
|
||||
JSON(&'a web::Bytes),
|
||||
Multi(&'a web::Bytes),
|
||||
Hec(&'a Vec<json::Value>),
|
||||
Loki(&'a Vec<json::Value>),
|
||||
GCP(&'a GCPIngestionRequest),
|
||||
KinesisFH(&'a KinesisFHRequest),
|
||||
RUM(&'a web::Bytes),
|
||||
Usage(&'a web::Bytes),
|
||||
Hec(&'a Vec<json::Value>),
|
||||
}
|
||||
|
||||
pub enum IngestionData<'a> {
|
||||
|
||||
@@ -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"),
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<usize>,
|
||||
org_id: web::Path<String>,
|
||||
@@ -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();
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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
|
||||
))
|
||||
|
||||
Reference in New Issue
Block a user