jianmu
/
openobserve
mirror of https://github.com/openobserve/openobserve
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor: optimize log write procedure (#2552)

This commit is contained in:
xiaoziv 2024-01-24 21:04:52 +08:00 committed by GitHub
parent f1fbd2e7f4
commit cc404ecb7e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 208 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/common/utils/json.rs
View File

@ -56,3 +56,16 @@ pub fn estimate_json_bytes(val: &Value) -> usize {
}
size
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_estimate_json_bytes() {
let json = r#"{"a":null,"b":true,"c":false,"d":{"a":"b","c":true,"d":false,"e":123456},"e":[""],"f":["a"],"g":["a","b"],"h":"bcdef","i":{},"j":{"ok":"yes"}}"#;
let val: Value = from_str(json).unwrap();
assert_eq!(estimate_json_bytes(&val), json.len());
}
}

14
src/service/ingestion/mod.rs
View File

@ -27,6 +27,8 @@ use vrl::{
prelude::state,
};
use anyhow::{anyhow, Result};
use crate::{
common::{
infra::{
@ -271,7 +273,7 @@ pub fn apply_stream_transform(
stream_vrl_map: &HashMap<String, VRLResultResolver>,
stream_name: &str,
runtime: &mut Runtime,
) -> Result<Value, anyhow::Error> {
) -> Result<Value> {
for trans in local_trans {
let func_key = format!("{stream_name}/{}", trans.transform.name);
if stream_vrl_map.contains_key(&func_key) && !value.is_null() {
@ -319,22 +321,22 @@ pub async fn write_file(
req_stats
}
pub fn is_ingestion_allowed(org_id: &str, stream_name: Option<&str>) -> Option<anyhow::Error> {
pub fn check_ingestion_allowed(org_id: &str, stream_name: Option<&str>) -> Result<()> {
if !cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) {
return Some(anyhow::anyhow!("not an ingester"));
return Err(anyhow!("not an ingester"));
}
if !db::file_list::BLOCKED_ORGS.is_empty() && db::file_list::BLOCKED_ORGS.contains(&org_id) {
return Some(anyhow::anyhow!("Quota exceeded for this organization"));
return Err(anyhow!("Quota exceeded for this organization"));
}
// check if we are allowed to ingest
if let Some(stream_name) = stream_name {
if db::compact::retention::is_deleting_stream(org_id, stream_name, StreamType::Logs, None) {
return Some(anyhow::anyhow!("stream [{stream_name}] is being deleted"));
return Err(anyhow!("stream [{stream_name}] is being deleted"));
}
};
None
Ok(())
}
pub fn get_float_value(val: &Value) -> f64 {

37
src/service/logs/bulk.rs
View File

@ -202,7 +202,10 @@ pub async fn ingest(
// End row based transform
// get json object
let local_val = value.as_object_mut().unwrap();
let mut local_val = match value.take() {
json::Value::Object(v) => v,
_ => unreachable!(),
};
// set _id
if !doc_id.is_empty() {
local_val.insert("_id".to_string(), json::Value::String(doc_id.clone()));
@ -260,6 +263,23 @@ pub async fn ingest(
let mut status = RecordStatus::default();
let need_trigger = !stream_trigger_map.contains_key(&stream_name);
let mut to_add_distinct_values = vec![];
// get distinct_value items
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
to_add_distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.clone(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
let local_trigger = match super::add_valid_record(
&StreamMeta {
org_id: org_id.to_string(),
@ -296,20 +316,7 @@ pub async fn ingest(
}
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.clone(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
distinct_values.extend(to_add_distinct_values);
if status.failed > 0 {
bulk_res.errors = true;

55
src/service/logs/ingest.rs
View File

@ -19,6 +19,7 @@ use std::{
};
use actix_web::http;
use anyhow::{anyhow, Result};
use chrono::{Duration, Utc};
use config::{meta::stream::StreamType, metrics, CONFIG, DISTINCT_FIELDS};
use datafusion::arrow::datatypes::Schema;
@ -42,7 +43,7 @@ use crate::{
},
service::{
distinct_values, get_formatted_stream_name,
ingestion::{evaluate_trigger, is_ingestion_allowed, write_file, TriggerAlertData},
ingestion::{check_ingestion_allowed, evaluate_trigger, write_file, TriggerAlertData},
logs::StreamMeta,
schema::get_upto_discard_error,
usage::report_request_usage_stats,
@ -54,15 +55,13 @@ pub async fn ingest(
in_stream_name: &str,
in_req: IngestionRequest<'_>,
thread_id: usize,
) -> Result<IngestionResponse, anyhow::Error> {
) -> Result<IngestionResponse> {
let start = std::time::Instant::now();
// check stream
let mut stream_schema_map: HashMap<String, Schema> = HashMap::new();
let mut stream_params = StreamParams::new(org_id, in_stream_name, StreamType::Logs);
let stream_name = &get_formatted_stream_name(&mut stream_params, &mut stream_schema_map).await;
if let Some(value) = is_ingestion_allowed(org_id, Some(stream_name)) {
return Err(value);
}
check_ingestion_allowed(org_id, Some(stream_name))?;
// check memtable
if let Err(e) = ingester::check_memtable_size() {
@ -129,7 +128,8 @@ pub async fn ingest(
Ok(item) => item,
Err(e) => {
log::error!("IngestionError: {:?}", e);
return Err(anyhow::anyhow!("Failed processing: {:?}", e));
continue;
// return Err(anyhow::anyhow!("Failed processing: {:?}", e));
}
};
@ -148,13 +148,33 @@ pub async fn ingest(
}
};
let local_val = res.as_object_mut().unwrap();
if let Err(e) = handle_timestamp(local_val, min_ts) {
let mut local_val = match res.take() {
json::Value::Object(val) => val,
_ => unreachable!(),
};
if let Err(e) = handle_timestamp(&mut local_val, min_ts) {
stream_status.status.failed += 1;
stream_status.status.error = e.to_string();
continue;
}
let mut to_add_distinct_values = vec![];
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
to_add_distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
let local_trigger = match super::add_valid_record(
&StreamMeta {
org_id: org_id.to_string(),
@ -181,22 +201,7 @@ pub async fn ingest(
if local_trigger.is_some() {
trigger = local_trigger;
}
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
distinct_values.extend(to_add_distinct_values);
}
// write data to wal
@ -268,7 +273,7 @@ pub fn apply_functions<'a>(
stream_vrl_map: &'a HashMap<String, VRLResultResolver>,
stream_name: &'a str,
runtime: &mut Runtime,
) -> Result<json::Value, anyhow::Error> {
) -> Result<json::Value> {
let mut value = flatten::flatten(item)?;
if !local_trans.is_empty() {

17
src/service/logs/mod.rs
View File

@ -15,6 +15,7 @@
use std::{collections::HashMap, sync::Arc};
use anyhow::Result;
use arrow_schema::{DataType, Field};
use config::{meta::stream::StreamType, utils::schema_ext::SchemaExt, CONFIG};
use datafusion::arrow::datatypes::Schema;
@ -149,9 +150,9 @@ async fn add_valid_record(
stream_schema_map: &mut HashMap<String, Schema>,
status: &mut RecordStatus,
write_buf: &mut HashMap<String, SchemaRecords>,
record_val: &mut Map<String, Value>,
mut record_val: Map<String, Value>,
need_trigger: bool,
) -> Result<Option<TriggerAlertData>, anyhow::Error> {
) -> Result<Option<TriggerAlertData>> {
let mut trigger: TriggerAlertData = Vec::new();
let timestamp: i64 = record_val
.get(&CONFIG.common.column_timestamp)
@ -165,7 +166,7 @@ async fn add_valid_record(
&stream_meta.stream_name,
StreamType::Logs,
stream_schema_map,
record_val,
&record_val,
timestamp,
)
.await?;
@ -177,7 +178,7 @@ async fn add_valid_record(
timestamp,
stream_meta.partition_keys,
unwrap_partition_time_level(*stream_meta.partition_time_level, StreamType::Logs),
record_val,
&record_val,
Some(&schema_key),
);
@ -186,14 +187,14 @@ async fn add_valid_record(
let ret_val = if !CONFIG.common.widening_schema_evolution
|| !schema_evolution.is_schema_changed
{
cast_to_type(record_val, delta)
cast_to_type(&mut record_val, delta)
} else {
let local_delta = delta
.into_iter()
.filter(|x| x.metadata().contains_key("zo_cast"))
.collect::<Vec<_>>();
if !local_delta.is_empty() {
cast_to_type(record_val, local_delta)
cast_to_type(&mut record_val, local_delta)
} else {
Ok(())
}
@ -221,7 +222,7 @@ async fn add_valid_record(
);
if let Some(alerts) = stream_meta.stream_alerts_map.get(&key) {
for alert in alerts {
if let Ok(Some(v)) = alert.evaluate(Some(record_val)).await {
if let Ok(Some(v)) = alert.evaluate(Some(&record_val)).await {
trigger.push((alert.clone(), v));
}
}
@ -238,7 +239,7 @@ async fn add_valid_record(
records_size: 0,
}
});
let record_val = Value::Object(record_val.clone());
let record_val = Value::Object(record_val);
let record_size = estimate_json_bytes(&record_val);
hour_buf.records.push(Arc::new(record_val));
hour_buf.records_size += record_size;

54
src/service/logs/multi.rs
View File

@ -19,6 +19,7 @@ use std::{
};
use actix_web::{http, web};
use anyhow::Result;
use chrono::{Duration, Utc};
use config::{meta::stream::StreamType, metrics, CONFIG, DISTINCT_FIELDS};
use datafusion::arrow::datatypes::Schema;
@ -35,7 +36,7 @@ use crate::{
},
service::{
distinct_values, get_formatted_stream_name,
ingestion::{evaluate_trigger, is_ingestion_allowed, write_file, TriggerAlertData},
ingestion::{check_ingestion_allowed, evaluate_trigger, write_file, TriggerAlertData},
logs::StreamMeta,
schema::get_upto_discard_error,
usage::report_request_usage_stats,
@ -66,7 +67,7 @@ async fn ingest_inner(
body: web::Bytes,
extend_json: &HashMap<String, serde_json::Value>,
thread_id: usize,
) -> Result<IngestionResponse, anyhow::Error> {
) -> Result<IngestionResponse> {
let start = std::time::Instant::now();
let mut stream_schema_map: HashMap<String, Schema> = HashMap::new();
@ -74,9 +75,7 @@ async fn ingest_inner(
let mut stream_params = StreamParams::new(org_id, in_stream_name, StreamType::Logs);
let stream_name = &get_formatted_stream_name(&mut stream_params, &mut stream_schema_map).await;
if let Some(value) = is_ingestion_allowed(org_id, Some(stream_name)) {
return Err(value);
}
check_ingestion_allowed(org_id, Some(stream_name))?;
let mut runtime = crate::service::ingestion::init_functions_runtime();
let min_ts =
@ -137,15 +136,12 @@ async fn ingest_inner(
)?;
}
if value.is_null() || !value.is_object() {
stream_status.status.failed += 1; // transform failed or dropped
continue;
}
let mut local_val = match value.take() {
json::Value::Object(v) => v,
_ => unreachable!(),
};
// End row based transform
// get json object
let local_val = value.as_object_mut().unwrap();
// handle timestamp
let timestamp = match local_val.get(&CONFIG.common.column_timestamp) {
Some(v) => match parse_timestamp_micro_from_value(v) {
@ -169,6 +165,23 @@ async fn ingest_inner(
json::Value::Number(timestamp.into()),
);
let mut to_add_distinct_values = vec![];
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
to_add_distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
// write data
let local_trigger = match super::add_valid_record(
&StreamMeta {
@ -197,21 +210,8 @@ async fn ingest_inner(
trigger = local_trigger;
}
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
// add distinct values
distinct_values.extend(to_add_distinct_values);
}
// write data to wal

88
src/service/logs/otlp_grpc.rs
View File

@ -16,6 +16,7 @@
use std::collections::HashMap;
use actix_web::{http, web, HttpResponse};
use anyhow::Result;
use bytes::BytesMut;
use chrono::{Duration, Utc};
use config::{meta::stream::StreamType, metrics, CONFIG, DISTINCT_FIELDS};
@ -25,6 +26,7 @@ use opentelemetry_proto::tonic::collector::logs::v1::{
ExportLogsServiceRequest, ExportLogsServiceResponse,
};
use prost::Message;
use anyhow::Result;
use super::StreamMeta;
use crate::{
@ -56,7 +58,7 @@ pub async fn usage_ingest(
in_stream_name: &str,
body: web::Bytes,
thread_id: usize,
) -> Result<IngestionResponse, anyhow::Error> {
) -> Result<IngestionResponse> {
let start = std::time::Instant::now();
let mut stream_schema_map: HashMap<String, Schema> = HashMap::new();
let mut distinct_values = Vec::with_capacity(16);
@ -108,7 +110,13 @@ pub async fn usage_ingest(
let mut value = flatten::flatten(item)?;
// get json object
let local_val = value.as_object_mut().unwrap();
let mut local_val = match value.take() {
json::Value::Object(v) => v,
_ => {
stream_status.status.failed += 1; // transform failed or dropped
continue;
}
};
// handle timestamp
let timestamp = match local_val.get(&CONFIG.common.column_timestamp) {
@ -133,6 +141,23 @@ pub async fn usage_ingest(
json::Value::Number(timestamp.into()),
);
let mut to_add_distinct_values = vec![];
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
to_add_distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
let local_trigger = match super::add_valid_record(
&StreamMeta {
org_id: org_id.to_string(),
@ -159,22 +184,7 @@ pub async fn usage_ingest(
if local_trigger.is_some() {
trigger = local_trigger;
}
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
distinct_values.extend(to_add_distinct_values);
}
// write data to wal
@ -226,7 +236,7 @@ pub async fn handle_grpc_request(
request: ExportLogsServiceRequest,
is_grpc: bool,
in_stream_name: Option<&str>,
) -> Result<HttpResponse, anyhow::Error> {
) -> Result<HttpResponse> {
if !cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) {
return Ok(
HttpResponse::InternalServerError().json(MetaHttpResponse::error(
@ -398,8 +408,29 @@ pub async fn handle_grpc_request(
&mut runtime,
)?;
}
// get json object
let local_val = rec.as_object_mut().unwrap();
let local_val = match rec.take() {
json::Value::Object(v) => v,
_ => unreachable!(),
};
let mut to_add_distinct_values = vec![];
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
to_add_distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
let local_trigger = match super::add_valid_record(
&StreamMeta {
@ -427,22 +458,7 @@ pub async fn handle_grpc_request(
if local_trigger.is_some() {
trigger = local_trigger;
}
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
distinct_values.extend(to_add_distinct_values);
}
}
}

40
src/service/logs/otlp_http.rs
View File

@ -265,7 +265,7 @@ pub async fn logs_json_handler(
let mut value: json::Value = json::to_value(log).unwrap();
// get json object
let mut local_val = value.as_object_mut().unwrap();
let local_val = value.as_object_mut().unwrap();
if log.get("attributes").is_some() {
let attributes = log.get("attributes").unwrap().as_array().unwrap();
@ -351,7 +351,27 @@ pub async fn logs_json_handler(
.unwrap();
}
local_val = value.as_object_mut().unwrap();
let local_val = match value.take() {
json::Value::Object(v) => v,
_ => unreachable!(),
};
let mut to_add_distinct_values = vec![];
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
to_add_distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
let local_trigger = match super::add_valid_record(
&StreamMeta {
@ -380,21 +400,7 @@ pub async fn logs_json_handler(
trigger = local_trigger;
}
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
distinct_values.extend(to_add_distinct_values);
}
}
}

45
src/service/logs/syslog.rs
View File

@ -16,6 +16,7 @@
use std::{collections::HashMap, net::SocketAddr};
use actix_web::{http, HttpResponse};
use anyhow::Result;
use chrono::{Duration, Utc};
use config::{meta::stream::StreamType, metrics, CONFIG, DISTINCT_FIELDS};
use datafusion::arrow::datatypes::Schema;
@ -41,7 +42,7 @@ use crate::{
},
};
pub async fn ingest(msg: &str, addr: SocketAddr) -> Result<HttpResponse, anyhow::Error> {
pub async fn ingest(msg: &str, addr: SocketAddr) -> Result<HttpResponse> {
let start = std::time::Instant::now();
let ip = addr.ip();
let matching_route = get_org_for_ip(ip).await;
@ -129,13 +130,11 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result<HttpResponse, anyhow:
&mut runtime,
)?;
}
if value.is_null() || !value.is_object() {
stream_status.status.failed += 1; // transform failed or dropped
}
// End row based transform
// get json object
let local_val = value.as_object_mut().unwrap();
let mut local_val = match value.take() {
json::Value::Object(v) => v,
_ => unreachable!(),
};
// handle timestamp
let timestamp = match local_val.get(&CONFIG.common.column_timestamp) {
@ -157,6 +156,23 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result<HttpResponse, anyhow:
json::Value::Number(timestamp.into()),
);
let mut to_add_distinct_values = vec![];
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
to_add_distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
let local_trigger = match super::add_valid_record(
&StreamMeta {
org_id: org_id.to_string(),
@ -185,20 +201,7 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result<HttpResponse, anyhow:
}
// get distinct_value item
for field in DISTINCT_FIELDS.iter() {
if let Some(val) = local_val.get(field) {
if !val.is_null() {
distinct_values.push(distinct_values::DvItem {
stream_type: StreamType::Logs,
stream_name: stream_name.to_string(),
field_name: field.to_string(),
field_value: val.as_str().unwrap().to_string(),
filter_name: "".to_string(),
filter_value: "".to_string(),
});
}
}
}
distinct_values.extend(to_add_distinct_values);
// write data to wal
let writer = ingester::get_writer(thread_id, org_id, &StreamType::Logs.to_string()).await;