diff --git a/src/common/utils/json.rs b/src/common/utils/json.rs index 6c03520568..7c8fe62f4b 100644 --- a/src/common/utils/json.rs +++ b/src/common/utils/json.rs @@ -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()); + } + +} diff --git a/src/service/ingestion/mod.rs b/src/service/ingestion/mod.rs index dfbbe41c5b..f5229e0f03 100644 --- a/src/service/ingestion/mod.rs +++ b/src/service/ingestion/mod.rs @@ -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, stream_name: &str, runtime: &mut Runtime, -) -> Result { +) -> Result { 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 { +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 { diff --git a/src/service/logs/bulk.rs b/src/service/logs/bulk.rs index bf60c8e387..2974262078 100644 --- a/src/service/logs/bulk.rs +++ b/src/service/logs/bulk.rs @@ -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; diff --git a/src/service/logs/ingest.rs b/src/service/logs/ingest.rs index d653932493..bb00747db2 100644 --- a/src/service/logs/ingest.rs +++ b/src/service/logs/ingest.rs @@ -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 { +) -> Result { let start = std::time::Instant::now(); // check stream let mut stream_schema_map: HashMap = 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, stream_name: &'a str, runtime: &mut Runtime, -) -> Result { +) -> Result { let mut value = flatten::flatten(item)?; if !local_trans.is_empty() { diff --git a/src/service/logs/mod.rs b/src/service/logs/mod.rs index c7b2f91f7b..c9ff9f97cc 100644 --- a/src/service/logs/mod.rs +++ b/src/service/logs/mod.rs @@ -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, status: &mut RecordStatus, write_buf: &mut HashMap, - record_val: &mut Map, + mut record_val: Map, need_trigger: bool, -) -> Result, anyhow::Error> { +) -> Result> { 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::>(); 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; diff --git a/src/service/logs/multi.rs b/src/service/logs/multi.rs index e9b685d4ab..124e7e8dd3 100644 --- a/src/service/logs/multi.rs +++ b/src/service/logs/multi.rs @@ -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, thread_id: usize, -) -> Result { +) -> Result { let start = std::time::Instant::now(); let mut stream_schema_map: HashMap = 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 diff --git a/src/service/logs/otlp_grpc.rs b/src/service/logs/otlp_grpc.rs index 9b7c3533a5..3ba84f54be 100644 --- a/src/service/logs/otlp_grpc.rs +++ b/src/service/logs/otlp_grpc.rs @@ -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 { +) -> Result { let start = std::time::Instant::now(); let mut stream_schema_map: HashMap = 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 { +) -> Result { 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); } } } diff --git a/src/service/logs/otlp_http.rs b/src/service/logs/otlp_http.rs index 6ebfcb47b8..0a129dc201 100644 --- a/src/service/logs/otlp_http.rs +++ b/src/service/logs/otlp_http.rs @@ -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); } } } diff --git a/src/service/logs/syslog.rs b/src/service/logs/syslog.rs index 96270ba7dc..ced2623139 100644 --- a/src/service/logs/syslog.rs +++ b/src/service/logs/syslog.rs @@ -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 { +pub async fn ingest(msg: &str, addr: SocketAddr) -> Result { 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 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 Result