From 8a4c5752b9fbb527472d9740b4151be949104f76 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Wed, 21 Oct 2026 12:49:28 +0000 Subject: [PATCH] perf(logs): route _bulk data lines through the direct-to-Arrow columnar path (#15592) --- src/core/src/logs/bulk.rs | 240 ++++++++++++++++-------- src/core/src/logs/columnar.rs | 255 +++++++++++++++++++++++++- src/core/src/logs/hec.rs | 25 +-- src/core/src/logs/ingest.rs | 334 +++++++++++++++++++++++++++------- src/core/src/logs/mod.rs | 26 +-- 5 files changed, 697 insertions(+), 183 deletions(-) diff --git a/src/core/src/logs/bulk.rs b/src/core/src/logs/bulk.rs index f4642d2c0c..445d971360 100644 --- a/src/core/src/logs/bulk.rs +++ b/src/core/src/logs/bulk.rs @@ -30,15 +30,13 @@ use config::{ }, }; use infra::errors::Result; -use ingestion_common::{ - BulkResponse, BulkResponseError, BulkResponseItem, IngestionRequest, IngestionValueType, -}; +use ingestion_common::{BulkResponse, BulkResponseError, BulkResponseItem}; use schema::{get_future_discard_error, get_upto_discard_error}; use transform::TRANSFORM_FAILED; use crate::{ ingestion::check_ingestion_allowed, - logs::{ingestion_log_enabled, log_failed_record}, + logs::{columnar::JsonColumnar, ingest::BulkPlan, ingestion_log_enabled, log_failed_record}, }; pub const TS_PARSE_FAILED: &str = "timestamp_parsing_failed"; pub const SCHEMA_CONFORMANCE_FAILED: &str = "schema_conformance_failed"; @@ -69,12 +67,14 @@ pub async fn ingest( let log_ingestion_errors = ingestion_log_enabled().await; let mut action = String::new(); + let mut raw_stream_name = String::new(); let mut stream_name = String::new(); let mut doc_id: Option = None; let stream_type = StreamType::Logs; let mut stream_key_cache: HashMap = HashMap::new(); let mut streams_data: HashMap> = HashMap::new(); + let mut plans: HashMap> = HashMap::new(); let mut next_line_is_data = false; // Read lines as bytes to handle potential invalid UTF-8 characters let mut line_buffer = Vec::new(); @@ -94,12 +94,22 @@ pub async fn ingest( if line_buffer.is_empty() { continue; } + + let is_data = std::mem::take(&mut next_line_is_data); + if is_data + && stream_columnar(&mut plans, org_id, &stream_name, (min_ts, max_ts)) + .await + .is_some_and(|c| c.parse_bulk_line(&line_buffer, doc_id.as_deref())) + { + tokio::task::coop::consume_budget().await; + continue; + } // Use from_utf8_lossy to handle potential invalid UTF-8 characters // Invalid UTF-8 sequences will be replaced with the replacement character (�) let line_str = String::from_utf8_lossy(&line_buffer); let mut value: json::Value = json::from_slice(line_str.as_bytes())?; - if !next_line_is_data { + if !is_data { // check bulk operate let Some((line_action, line_stream_name, line_doc_id)) = super::parse_bulk_index(&value) @@ -109,18 +119,24 @@ pub async fn ingest( if line_action != action { action = line_action.to_string(); } - if line_stream_name != stream_name { - stream_name = line_stream_name.to_string(); + // formatting reads the shared config, too costly to repeat for every action line + if line_stream_name != raw_stream_name { + raw_stream_name = line_stream_name.to_string(); + stream_name = if cfg.common.skip_formatting_stream_name { + raw_stream_name.clone() + } else { + format_stream_name(raw_stream_name.clone()) + }; } doc_id = line_doc_id.map(|id| id.to_string()); - if stream_name.is_empty() || stream_name == "_" || stream_name == "/" { + if raw_stream_name.is_empty() || raw_stream_name == "_" || raw_stream_name == "/" { let err_msg = "Invalid stream name: ".to_string() + &line_str; log::warn!("[LOGS:BULK] {err_msg}, org_id: {org_id}"); bulk_res.errors = true; let err = BulkResponseError::new( err_msg.to_string(), - stream_name.to_string(), + raw_stream_name.to_string(), err_msg, "0".to_string(), ); @@ -128,21 +144,17 @@ pub async fn ingest( item.insert( action.to_string(), BulkResponseItem::new_failed( - stream_name.to_string(), + raw_stream_name.to_string(), doc_id.clone().unwrap_or_default(), err, Some(value), - stream_name.to_string(), + raw_stream_name.to_string(), ), ); bulk_res.items.push(item); continue; // skip } - if !cfg.common.skip_formatting_stream_name { - stream_name = format_stream_name(stream_name); - } - // Reject internal rollup streams (_o2_*, _agent_signals) in ALL // editions — written only by internal aggregation jobs (service // graph, agent signals, database monitoring) through the internal @@ -187,57 +199,16 @@ pub async fn ingest( } next_line_is_data = true; } else { - next_line_is_data = false; - // get json object let mut local_val = match value.take() { json::Value::Object(v) => v, _ => unreachable!(), }; - // set _id - if let Some(doc_id) = &doc_id { - local_val.insert("_id".to_string(), json::Value::String(doc_id.to_string())); - } - - // check _timestamp - let (timestamp, has_valid_timestamp) = match local_val.get(TIMESTAMP_COL_NAME) { - Some(v) => match parse_timestamp_micro_from_value(v) { - Ok(t) => (t.0, t.1), - Err(_e) => { - bulk_res.errors = true; - metrics::INGEST_ERRORS - .with_label_values(&[ - org_id, - StreamType::Logs.as_str(), - &stream_name, - TS_PARSE_FAILED, - ]) - .inc(); - log_failed_record(log_ingestion_errors, &value, TS_PARSE_FAILED); - add_record_status( - stream_name.to_string(), - doc_id.clone(), - action.to_string(), - Some(value), - &mut bulk_res, - Some(TS_PARSE_FAILED.to_string()), - Some(TS_PARSE_FAILED.to_string()), - ); - continue; - } - }, - None => (now_micros(), false), - }; - - // check ingestion time - if timestamp < min_ts || timestamp > max_ts { + if let Err(failure_reason) = + prepare_bulk_record(&mut local_val, doc_id.as_deref(), min_ts, max_ts) + { bulk_res.errors = true; - let failure_reason = if timestamp < min_ts { - Some(get_upto_discard_error().to_string()) - } else { - Some(get_future_discard_error().to_string()) - }; metrics::INGEST_ERRORS .with_label_values(&[ org_id, @@ -254,16 +225,10 @@ pub async fn ingest( Some(value), &mut bulk_res, Some(TS_PARSE_FAILED.to_string()), - failure_reason, + Some(failure_reason), ); continue; } - if !has_valid_timestamp { - local_val.insert( - TIMESTAMP_COL_NAME.to_string(), - json::Value::Number(timestamp.into()), - ); - } let val = json::Value::Object(local_val); match streams_data.get_mut(&stream_name) { @@ -276,6 +241,11 @@ pub async fn ingest( tokio::task::coop::consume_budget().await; } + plans.retain(|_, plan| plan.as_ref().is_some_and(|p| p.columnar.rows() > 0)); + for stream in plans.keys() { + streams_data.entry(stream.clone()).or_default(); + } + // Checked up front, pipeline destinations included: a whole-request 503 is retried safely. #[cfg(feature = "vectorscan")] { @@ -300,14 +270,14 @@ pub async fn ingest( for (stream_name, records) in streams_data { #[cfg(feature = "vectorscan")] let record_count = records.len(); - match super::ingest::ingest( + let plan = plans.remove(&stream_name).flatten(); + match super::ingest::ingest_bulk( thread_id, org_id, &stream_name, - IngestionRequest::JsonValues(IngestionValueType::Bulk, records), + records, + plan, user.clone(), - None, - false, ) .await { @@ -441,17 +411,104 @@ pub fn add_record_status( bulk_res.items.push(item); } None => { - item.insert( - action, - BulkResponseItem::new(stream_name.clone(), doc_id, value, stream_name), - ); if !get_config().common.bulk_api_response_errors_only { - bulk_res.items.push(item); + bulk_res + .items + .push(success_item(action, &stream_name, doc_id)); } } } } +/// The response item of a written record; callers skip it under `bulk_api_response_errors_only`. +pub fn success_item( + action: String, + stream_name: &str, + doc_id: String, +) -> HashMap { + HashMap::from([( + action, + BulkResponseItem::new( + stream_name.to_string(), + doc_id, + None, + stream_name.to_string(), + ), + )]) +} + +/// Success items for records written together, reading the config once rather than per record. +pub fn add_success_items( + stream_name: &str, + doc_ids: Vec>, + bulk_res: &mut BulkResponse, +) { + if get_config().common.bulk_api_response_errors_only { + return; + } + bulk_res.items.extend( + doc_ids.into_iter().map(|doc_id| { + success_item("index".to_string(), stream_name, doc_id.unwrap_or_default()) + }), + ); +} + +/// `_bulk`'s own step for a data record: set the action's `_id`, then check and set `_timestamp`. +pub(super) fn prepare_bulk_record( + record: &mut json::Map, + doc_id: Option<&str>, + min_ts: i64, + max_ts: i64, +) -> std::result::Result<(), String> { + if let Some(doc_id) = doc_id { + record.insert("_id".to_string(), json::Value::String(doc_id.to_string())); + } + let parsed = record + .get(TIMESTAMP_COL_NAME) + .map(parse_timestamp_micro_from_value); + let (timestamp, has_valid_timestamp) = resolve_bulk_timestamp(parsed, min_ts, max_ts)?; + if !has_valid_timestamp { + record.insert( + TIMESTAMP_COL_NAME.to_string(), + json::Value::Number(timestamp.into()), + ); + } + Ok(()) +} + +/// `_bulk`'s verdict on a parsed `_timestamp`, stamping now when absent; the error is the reason. +pub(super) fn resolve_bulk_timestamp( + parsed: Option>, + min_ts: i64, + max_ts: i64, +) -> std::result::Result<(i64, bool), String> { + let (timestamp, has_valid_timestamp) = match parsed { + Some(parsed) => parsed.map_err(|_| TS_PARSE_FAILED.to_string())?, + None => (now_micros(), false), + }; + if timestamp < min_ts { + return Err(get_upto_discard_error().to_string()); + } + if timestamp > max_ts { + return Err(get_future_discard_error().to_string()); + } + Ok((timestamp, has_valid_timestamp)) +} + +/// The stream's columnar rows, planned on its first data line of the request. +async fn stream_columnar<'a>( + plans: &'a mut HashMap>, + org_id: &str, + stream_name: &str, + time_range: (i64, i64), +) -> Option<&'a mut JsonColumnar> { + if !plans.contains_key(stream_name) { + let plan = super::ingest::plan_bulk_columnar(org_id, stream_name, time_range).await; + plans.insert(stream_name.to_string(), plan); + } + Some(&mut plans.get_mut(stream_name)?.as_mut()?.columnar) +} + #[cfg(test)] mod tests { use ingestion_common::IngestUser; @@ -477,6 +534,39 @@ mod tests { assert!(bulk_res.items.len() == 1); } + #[test] + fn test_prepare_bulk_record() { + let now = now_micros(); + let (min_ts, max_ts) = (now - 3_600_000_000, now + 3_600_000_000); + let prepare = |record: json::Value, doc_id: Option<&str>| { + let mut record = record.as_object().unwrap().clone(); + prepare_bulk_record(&mut record, doc_id, min_ts, max_ts).map(|()| record) + }; + let record = prepare(json::json!({"_id": "own", "_timestamp": now}), Some("a")).unwrap(); + assert_eq!(record["_id"], "a"); + assert_eq!(record["_timestamp"], json::json!(now)); + let record = prepare(json::json!({"_id": "own", "level": "x"}), None).unwrap(); + assert_eq!(record["_id"], "own"); + assert!(record["_timestamp"].as_i64().is_some_and(|ts| ts >= now)); + let record = prepare(json::json!({"_timestamp": now / 1_000_000}), None).unwrap(); + assert_eq!( + record["_timestamp"], + json::json!(now / 1_000_000 * 1_000_000) + ); + assert_eq!( + prepare(json::json!({"_timestamp": "garbage"}), None), + Err(TS_PARSE_FAILED.to_string()) + ); + assert_eq!( + prepare(json::json!({"_timestamp": min_ts - 1}), None), + Err(get_upto_discard_error().to_string()) + ); + assert_eq!( + prepare(json::json!({"_timestamp": max_ts + 1}), None), + Err(get_future_discard_error().to_string()) + ); + } + #[tokio::test] async fn test_ingest_basic_functionality() { // Create a simple bulk request with one document diff --git a/src/core/src/logs/columnar.rs b/src/core/src/logs/columnar.rs index 301fa42eb9..b90952ab68 100644 --- a/src/core/src/logs/columnar.rs +++ b/src/core/src/logs/columnar.rs @@ -47,7 +47,10 @@ use crate::{ }, }; -/// Builds a `_json` request's records straight into Arrow columns of the stream's current schema. +/// The field `_bulk` sets from an action line's `_id`. +const BULK_ID_COL: &str = "_id"; + +/// Builds `_json` or `_bulk` records straight into Arrow columns of the stream's current schema. pub struct JsonColumnar { stream_name: String, buckets: ColumnarBuckets, @@ -58,6 +61,9 @@ pub struct JsonColumnar { time_range: (i64, i64), dbm_enabled: bool, input_bytes: usize, + /// The action `_id` of each accepted `_bulk` record, for its response item. + bulk_ids: Vec>, + bulk_id_col: Option, } impl JsonColumnar { @@ -110,6 +116,10 @@ impl JsonColumnar { self.buckets.rows() } + pub fn flatten_level(&self) -> u32 { + self.flatten_level + } + /// What `estimate_json_bytes` counts over the accepted records as the client sent them. pub fn input_bytes(&self) -> usize { self.input_bytes @@ -152,6 +162,27 @@ impl JsonColumnar { .collect() } + /// Takes one `_bulk` data line under its action's `_id`; `false` leaves it to the JSON path. + pub fn parse_bulk_line(&mut self, line: &[u8], doc_id: Option<&str>) -> bool { + if doc_id.is_some() && self.bulk_id_col.is_none() { + return false; + } + let mut de = serde_json::Deserializer::from_slice(line); + let mut state = RecordState::new(self); + let accepted = match Record(&mut state).deserialize(&mut de) { + Ok(input_bytes) => de.end().is_ok() && state.finish_bulk(input_bytes, doc_id), + Err(_) => false, + }; + if accepted { + self.bulk_ids.push(doc_id.map(str::to_string)); + } + accepted + } + + pub fn take_bulk_ids(&mut self) -> Vec> { + std::mem::take(&mut self.bulk_ids) + } + pub fn into_entries(self, org_id: &str) -> Result> { let schema = self.buckets.schema(); let mut columns = self.buckets.filled_columns(); @@ -160,7 +191,7 @@ impl JsonColumnar { .into_entries(org_id, &self.stream_name, Some(columns)) } - fn for_schema( + pub(super) fn for_schema( stream_name: &str, schema: &Schema, flatten_level: u32, @@ -175,11 +206,12 @@ impl JsonColumnar { let ts_col = fields .iter() .position(|f| f.name() == TIMESTAMP_COL_NAME && f.data_type() == &DataType::Int64)?; - let col_index = fields + let col_index: hashbrown::HashMap = fields .iter() .enumerate() .map(|(idx, f)| (f.name().clone(), idx)) .collect(); + let bulk_id_col = col_index.get(BULK_ID_COL).copied(); Some(Self { stream_name: stream_name.to_string(), buckets: ColumnarBuckets::new(StreamType::Logs, schema), @@ -190,6 +222,8 @@ impl JsonColumnar { time_range, dbm_enabled, input_bytes: 0, + bulk_ids: Vec::new(), + bulk_id_col, }) } @@ -364,6 +398,8 @@ struct RecordState<'c, 'de> { null_marks: Vec, serial: u32, nested_top: bool, + /// Whether the record has a top-level key spelled exactly `_timestamp`. + raw_timestamp: bool, fallback: bool, } @@ -385,6 +421,7 @@ impl<'c, 'de> RecordState<'c, 'de> { null_marks: vec![0; columns], serial: 0, nested_top: false, + raw_timestamp: false, fallback: false, } } @@ -405,6 +442,7 @@ impl<'c, 'de> RecordState<'c, 'de> { self.null_bytes = 0; self.containers.clear(); self.nested_top = false; + self.raw_timestamp = false; self.fallback = false; } @@ -539,6 +577,40 @@ impl<'c, 'de> RecordState<'c, 'de> { true } + /// `finish` after `_bulk`'s own step: check `_timestamp` as sent, set the action's `_id`. + fn finish_bulk(&mut self, input_bytes: usize, doc_id: Option<&str>) -> bool { + // the JSON path reads a record's own `_id` back into its response item + if let Some(col) = self.columnar.bulk_id_col + && (self.marks[col] == self.serial || self.null_marks[col] == self.serial) + { + return false; + } + // `_bulk` reads the literal key, so a key that only flattens into `_timestamp` differs + if self.fallback || self.raw_timestamp != self.timestamp.is_some() { + return false; + } + // `{}` counts 2 bytes, yet once `_bulk` adds entries the object counts 1 plus theirs + let mut input_bytes = if input_bytes == 2 { 1 } else { input_bytes }; + let (min_ts, max_ts) = self.columnar.time_range; + let leaf = self.timestamp.take(); + let parsed = leaf.as_ref().map(Leaf::timestamp); + let Ok((timestamp, _)) = super::bulk::resolve_bulk_timestamp(parsed, min_ts, max_ts) else { + return false; + }; + input_bytes = match leaf { + Some(leaf) => input_bytes - leaf.json_bytes(), + None => input_bytes + estimate_json_entry_bytes(TIMESTAMP_COL_NAME, 0), + } + timestamp.json_bytes(); + self.timestamp = Some(Leaf::Num(Num::I64(timestamp))); + if let Some(id) = doc_id { + self.path.clear(); + self.path.push_str(BULK_ID_COL); + let bytes = self.push_leaf(Leaf::Str(Cow::Owned(id.to_string()))); + input_bytes += estimate_json_entry_bytes(BULK_ID_COL, bytes); + } + self.finish(input_bytes) + } + /// Flattens an object at `depth` into leaves, returning its `estimate_json_bytes` size. fn read_object>( &mut self, @@ -584,6 +656,8 @@ impl<'c, 'de> RecordState<'c, 'de> { let parent = self.path.len(); if depth > 0 { self.path.push('_'); + } else if name == TIMESTAMP_COL_NAME { + self.raw_timestamp = true; } push_formatted_key(&mut self.path, &name); let value_bytes = map.next_value_seed(FieldSeed { @@ -838,7 +912,7 @@ mod tests { use arrow::record_batch::RecordBatch; use arrow_schema::Field; - use config::utils::record_batch_ext::convert_json_to_record_batch; + use config::utils::{record_batch_ext::convert_json_to_record_batch, time::now_micros}; use super::*; @@ -858,13 +932,41 @@ mod tests { fn json_path(schema: &Schema, body: &str, flatten_level: u32) -> (RecordBatch, usize, usize) { let records: Vec = json::from_str(body).unwrap_or_else(|_| vec![json::from_str(body).unwrap()]); + json_records_path(schema, records, flatten_level, RANGE) + } + + /// What the JSON path writes for `_bulk`'s data `line` once `_bulk` has prepared it. + fn bulk_json_path( + schema: &Schema, + line: &str, + doc_id: Option<&str>, + range: (i64, i64), + ) -> (RecordBatch, usize, usize) { + let json::Value::Object(mut record) = json::from_str(line).unwrap() else { + panic!("{line} is not an object"); + }; + crate::logs::bulk::prepare_bulk_record(&mut record, doc_id, range.0, range.1) + .unwrap_or_else(|e| panic!("_bulk rejects {line}: {e}")); + json_records_path(schema, vec![json::Value::Object(record)], 3, range) + } + + fn json_records_path( + schema: &Schema, + records: Vec, + flatten_level: u32, + range: (i64, i64), + ) -> (RecordBatch, usize, usize) { let (mut rows, mut json_size, mut input_bytes) = (Vec::new(), 0, 0); let mut seen = std::collections::HashSet::new(); for record in records { input_bytes += estimate_json_bytes(&record); - let (value, _) = - crate::logs::ingest::prepare_record(record, flatten_level, RANGE.0, RANGE.1) - .unwrap_or_else(|_| panic!("the JSON path rejects a record of {body}")); + let (value, _) = crate::logs::ingest::prepare_record( + record.clone(), + flatten_level, + range.0, + range.1, + ) + .unwrap_or_else(|_| panic!("the JSON path rejects {record}")); json_size += estimate_json_bytes(&value); for (key, v) in value.as_object().unwrap() { if crate::ingestion::inferred_column_type(key, v).is_some() { @@ -1036,6 +1138,145 @@ mod tests { } } + fn bulk_schema() -> Schema { + let mut fields = logs_schema().fields().to_vec(); + fields.push(Arc::new(Field::new(BULK_ID_COL, DataType::Utf8, true))); + Schema::new(fields) + } + + /// A window around now, for records `_bulk` stamps with the current time. + fn now_range() -> (i64, i64) { + let now = now_micros(); + (now - 3_600_000_000, now + 3_600_000_000) + } + + fn without_timestamp(batch: &RecordBatch) -> RecordBatch { + let mut batch = batch.clone(); + batch.remove_column(batch.schema().index_of(TIMESTAMP_COL_NAME).unwrap()); + batch + } + + fn assert_bulk_same_as_json_path( + schema: &Schema, + line: &str, + doc_id: Option<&str>, + range: (i64, i64), + ) { + let (expected, json_size, input_bytes) = bulk_json_path(schema, line, doc_id, range); + let mut columnar = JsonColumnar::for_schema("logs", schema, 3, range, true).unwrap(); + assert!( + columnar.parse_bulk_line(line.as_bytes(), doc_id), + "{line} fell back" + ); + assert_eq!(columnar.input_bytes(), input_bytes, "input bytes of {line}"); + assert_eq!(columnar.take_bulk_ids(), vec![doc_id.map(str::to_string)]); + let entries = columnar.into_entries("org").unwrap(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].data_size, json_size, "json size of {line}"); + let batch = entries[0].batch.as_ref().unwrap(); + if line.contains(TIMESTAMP_COL_NAME) { + assert_eq!(batch, &expected, "{line}"); + } else { + // both paths stamp the current time, a few microseconds apart + assert_eq!( + without_timestamp(batch), + without_timestamp(&expected), + "{line}" + ); + } + } + + #[test] + fn test_bulk_line_matches_the_bulk_json_path() { + let schema = bulk_schema(); + let lines = [ + format!(r#"{{"_timestamp":{TS},"level":"info","k8s":{{"pod.name":"p-1"}}}}"#), + format!(r#"{{"_timestamp":"{TS}","Level":"warn","status":-3,"message":null}}"#), + format!( + r#"{{"_timestamp":{},"ratio":0.5,"tags":["a",1]}}"#, + TS / 1_000_000 + ), + format!(r#"{{"_timestamp":{TS},"_original":"skip me","ok":true}}"#), + ]; + for line in &lines { + for doc_id in [None, Some("doc-1"), Some(r#"a "quoted" \ id"#)] { + assert_bulk_same_as_json_path(&schema, line, doc_id, RANGE); + } + } + // `_bulk` stamps a record that has no `_timestamp` + for line in [ + r#"{"level":"info","bytes":7}"#, + "{}", + r#"{"_original":"x"}"#, + ] { + for doc_id in [None, Some("doc-2")] { + assert_bulk_same_as_json_path(&schema, line, doc_id, now_range()); + } + } + } + + #[test] + fn test_bulk_line_hands_back_what_bulk_must_decide() { + let schema = bulk_schema(); + let lines = [ + // the record's own `_id` would name its response item + format!(r#"{{"_timestamp":{TS},"_id":"own"}}"#), + format!(r#"{{"_timestamp":{TS},"_id":null,"level":"a"}}"#), + // `_bulk` reads only the literal `_timestamp` key, and rejects what it cannot parse + format!(r#"{{"_Timestamp":{TS},"level":"a"}}"#), + r#"{"_timestamp":null,"level":"a"}"#.to_string(), + r#"{"_timestamp":{},"level":"a"}"#.to_string(), + r#"{"_timestamp":true,"level":"a"}"#.to_string(), + r#"{"_timestamp":0,"level":"a"}"#.to_string(), + format!(r#"{{"_timestamp":{},"level":"a"}}"#, TS - 7_200_000_000), + format!(r#"{{"_timestamp":{},"level":"a"}}"#, TS + 7_200_000_000), + // what `_json` hands back too + format!(r#"{{"_timestamp":{TS},"new_field":"x"}}"#), + format!(r#"{{"_timestamp":{TS},"status":"200"}}"#), + "[1]".to_string(), + r#""text""#.to_string(), + format!(r#"{{"_timestamp":{TS}}} x"#), + format!(r#"{{"_timestamp":{TS},"level":"#), + ]; + for line in &lines { + for doc_id in [None, Some("doc-1")] { + let mut columnar = + JsonColumnar::for_schema("logs", &schema, 3, RANGE, true).unwrap(); + assert!(!columnar.parse_bulk_line(line.as_bytes(), doc_id), "{line}"); + assert_eq!(columnar.rows(), 0, "{line}"); + assert!(columnar.take_bulk_ids().is_empty(), "{line}"); + assert_eq!(columnar.input_bytes(), 0, "{line}"); + } + } + // invalid UTF-8 is left to the JSON path, which repairs it + let mut columnar = JsonColumnar::for_schema("logs", &schema, 3, RANGE, true).unwrap(); + let line = [br#"{"_timestamp":1,"level":""#.as_slice(), b"\xff\"}"].concat(); + assert!(!columnar.parse_bulk_line(&line, None)); + // an action `_id` needs a string `_id` column to land in + let mut columnar = + JsonColumnar::for_schema("logs", &logs_schema(), 3, RANGE, true).unwrap(); + let line = format!(r#"{{"_timestamp":{TS},"level":"a"}}"#); + assert!(!columnar.parse_bulk_line(line.as_bytes(), Some("doc-1"))); + assert!(columnar.parse_bulk_line(line.as_bytes(), None)); + } + + #[test] + fn test_bulk_lines_keep_their_action_ids_in_order() { + let schema = bulk_schema(); + let mut columnar = JsonColumnar::for_schema("logs", &schema, 3, RANGE, true).unwrap(); + let ok = format!(r#"{{"_timestamp":{TS},"level":"a"}}"#); + let new_field = format!(r#"{{"_timestamp":{TS},"nope":1}}"#); + assert!(columnar.parse_bulk_line(ok.as_bytes(), Some("1"))); + assert!(!columnar.parse_bulk_line(new_field.as_bytes(), Some("2"))); + assert!(columnar.parse_bulk_line(ok.as_bytes(), None)); + assert!(columnar.parse_bulk_line(ok.as_bytes(), Some("3"))); + assert_eq!(columnar.rows(), 3); + assert_eq!( + columnar.take_bulk_ids(), + vec![Some("1".to_string()), None, Some("3".to_string())] + ); + } + #[test] fn test_num_answers_as_serde_json_number() { for text in [ diff --git a/src/core/src/logs/hec.rs b/src/core/src/logs/hec.rs index d9f4a41fe0..68bb92216c 100644 --- a/src/core/src/logs/hec.rs +++ b/src/core/src/logs/hec.rs @@ -15,11 +15,8 @@ use axum::body::Bytes; use config::{ get_config, - meta::{ - self_reporting::usage::is_internal_rollup_stream, - stream::{StreamParams, StreamType}, - }, - utils::{json, schema::format_stream_name}, + meta::{self_reporting::usage::is_internal_rollup_stream, stream::StreamType}, + utils::json, }; use hashbrown::HashMap; use infra::{ @@ -34,7 +31,6 @@ use serde::{Deserialize, Deserializer}; use crate::{ ingestion::check_ingestion_allowed, logs::ingest::{PrepareRecordError, prepare_record}, - service::get_formatted_stream_name, }; /// Cap on events in one request: each is held decoded and flattened in memory, @@ -481,23 +477,14 @@ fn parse_hec_time(value: &json::Value) -> std::result::Result std::result::Result { - let cfg = get_config(); - let name = if cfg.common.skip_formatting_stream_name { - get_formatted_stream_name(StreamParams::new(org_id, index, StreamType::Logs)) - .await - .map_err(|e| HecParseError::InvalidFormat(e.to_string()))? - } else { - format_stream_name(index.to_string()) - }; + let name = super::ingest::resolve_stream_name(org_id, index) + .await + .map_err(|e| HecParseError::InvalidFormat(e.to_string()))?; if name.is_empty() { return Err(HecParseError::InvalidIndex); } diff --git a/src/core/src/logs/ingest.rs b/src/core/src/logs/ingest.rs index 0845d6b927..34da2a1f92 100644 --- a/src/core/src/logs/ingest.rs +++ b/src/core/src/logs/ingest.rs @@ -63,7 +63,7 @@ use super::{ }; use crate::{ ingestion::check_ingestion_allowed, logs::handle_timestamp_for_value, - service::get_formatted_stream_name, + pipeline::batch_execution::ExecutablePipeline, service::get_formatted_stream_name, }; /// A misbehaving client can wholly fail thousands of batches a second per stream. @@ -100,6 +100,68 @@ pub enum PrepareRecordError { Timestamp(json::Value, anyhow::Error), } +/// A stream's pipelines and the settings they and the stream impose on each record. +struct StreamRecordNeeds { + executable_pipelines: Vec, + user_defined_schema_map: HashMap>>, + streams_need_original_map: HashMap, + streams_need_all_values_map: HashMap, +} + +impl StreamRecordNeeds { + async fn load(stream_param: StreamParams) -> Self { + // Chat history is replayed verbatim and hash-checked; no user pipeline may + // read (copy elsewhere) or rewrite it. + let executable_pipelines = + if config::meta::self_reporting::ai_chat::is_protected_ai_chat_stream( + &stream_param.stream_name, + ) { + Vec::new() + } else { + crate::ingestion::get_stream_executable_pipelines(&stream_param).await + }; + let mut stream_params = vec![stream_param]; + for exec_pl in &executable_pipelines { + stream_params.extend(exec_pl.get_all_destination_streams()); + } + let mut needs = Self { + executable_pipelines, + user_defined_schema_map: HashMap::new(), + streams_need_original_map: HashMap::new(), + streams_need_all_values_map: HashMap::new(), + }; + crate::ingestion::get_uds_and_original_data_streams( + &stream_params, + &mut needs.user_defined_schema_map, + &mut needs.streams_need_original_map, + &mut needs.streams_need_all_values_map, + ) + .await; + needs + } + + /// Whether the stream's records must keep their JSON form, ruling out the columnar path. + fn json_records(&self, stream_name: &str) -> bool { + !self.executable_pipelines.is_empty() + || matches!(self.user_defined_schema_map.get(stream_name), Some(Some(_))) + || self + .streams_need_original_map + .get(stream_name) + .is_some_and(|v| *v) + || self + .streams_need_all_values_map + .get(stream_name) + .is_some_and(|v| *v) + } +} + +/// A `_bulk` stream's columnar rows with the pipeline and settings snapshot they were planned +/// under. +pub struct BulkPlan { + pub columnar: JsonColumnar, + needs: StreamRecordNeeds, +} + /// The `_o2_` write-guard predicate (design §5.3): true when a logs ingest /// request targeting `stream_name` must be rejected because the stream is an /// internal rollup stream and the request is user-initiated. Internal writers @@ -131,6 +193,91 @@ pub async fn ingest( user: IngestUser, extend_json: Option<&HashMap>, is_derived: bool, +) -> Result { + ingest_inner( + thread_id, + org_id, + in_stream_name, + in_req, + user, + extend_json, + is_derived, + None, + ) + .await +} + +/// Ingests one stream's `_bulk` records beside the rows its data lines already built. +pub async fn ingest_bulk( + thread_id: usize, + org_id: &str, + stream_name: &str, + records: Vec, + plan: Option, + user: IngestUser, +) -> Result { + ingest_inner( + thread_id, + org_id, + stream_name, + IngestionRequest::JsonValues(IngestionValueType::Bulk, records), + user, + None, + false, + plan, + ) + .await +} + +/// The columnar plan for a `_bulk` stream, under the preconditions `ingest` applies to `_json`. +pub async fn plan_bulk_columnar( + org_id: &str, + stream_name: &str, + time_range: (i64, i64), +) -> Option { + // rows are matched to their stream by the name `ingest` resolves + if resolve_stream_name(org_id, stream_name).await.ok()? != stream_name { + return None; + } + let needs = + StreamRecordNeeds::load(StreamParams::new(org_id, stream_name, StreamType::Logs)).await; + if needs.json_records(stream_name) { + return None; + } + let flatten_level = get_flatten_level(org_id, stream_name, StreamType::Logs).await; + let columnar = JsonColumnar::plan(org_id, stream_name, flatten_level, time_range).await?; + Some(BulkPlan { columnar, needs }) +} + +/// Flatten a record and resolve its timestamp — the only two steps of record +/// preparation that can reject an event. +/// +/// Shared with the HEC collector's pre-write validation pass, which must reach +/// the same verdict as the write does; a second implementation would drift. +pub fn prepare_record( + item: json::Value, + flatten_level: u32, + min_ts: i64, + max_ts: i64, +) -> std::result::Result<(json::Value, i64), PrepareRecordError> { + let mut res = + flatten::flatten_with_level(item, flatten_level).map_err(PrepareRecordError::Flatten)?; + match handle_timestamp_for_value(&mut res, min_ts, max_ts) { + Ok(ts) => Ok((res, ts)), + Err(e) => Err(PrepareRecordError::Timestamp(res, e)), + } +} + +#[allow(clippy::too_many_arguments)] +async fn ingest_inner( + thread_id: usize, + org_id: &str, + in_stream_name: &str, + in_req: IngestionRequest, + user: IngestUser, + extend_json: Option<&HashMap>, + is_derived: bool, + bulk_plan: Option, ) -> Result { let start = std::time::Instant::now(); let started_at: i64 = Utc::now().timestamp_micros(); @@ -157,11 +304,7 @@ pub async fn ingest( let stream_type = StreamType::Logs; // check stream - let stream_name = if cfg.common.skip_formatting_stream_name { - get_formatted_stream_name(StreamParams::new(org_id, in_stream_name, stream_type)).await? - } else { - format_stream_name(in_stream_name.to_string()) - }; + let stream_name = resolve_stream_name(org_id, in_stream_name).await?; if stream_name.is_empty() { return Err(Error::IngestionError("Stream name is empty".to_string())); } @@ -221,57 +364,34 @@ pub async fn ingest( derived_streams.insert(stream_name.to_string()); } - // Start retrieve associated pipeline and construct pipeline components - let stream_param = StreamParams::new(org_id, &stream_name, stream_type); - // Chat history is replayed verbatim and hash-checked; no user pipeline may - // read (copy elsewhere) or rewrite it. - let executable_pipelines = - if config::meta::self_reporting::ai_chat::is_protected_ai_chat_stream(&stream_name) { - Vec::new() - } else { - crate::ingestion::get_stream_executable_pipelines(&stream_param).await - }; - let mut stream_params = vec![stream_param]; - let mut pipeline_inputs = Vec::with_capacity(stream_params.len()); - let mut original_options = Vec::with_capacity(stream_params.len()); - // End pipeline params construction - - if !executable_pipelines.is_empty() { - for exec_pl in &executable_pipelines { - let pl_destinations = exec_pl.get_all_destination_streams(); - stream_params.extend(pl_destinations); - } - } - - // Start get user defined schema - let mut user_defined_schema_map: HashMap>> = HashMap::new(); - let mut streams_need_original_map: HashMap = HashMap::new(); - let mut streams_need_all_values_map: HashMap = HashMap::new(); - crate::ingestion::get_uds_and_original_data_streams( - &stream_params, - &mut user_defined_schema_map, - &mut streams_need_original_map, - &mut streams_need_all_values_map, + let (needs, bulk_columnar) = request_needs( + StreamParams::new(org_id, &stream_name, stream_type), + bulk_plan, ) .await; + // JsonColumnar::plan already refuses the fast path under vectorscan, before this is consulted. + let needs_json_records = extend_json.is_some() || needs.json_records(&stream_name); + let StreamRecordNeeds { + executable_pipelines, + mut user_defined_schema_map, + mut streams_need_original_map, + mut streams_need_all_values_map, + } = needs; + let mut pipeline_inputs = Vec::new(); + let mut original_options = Vec::new(); // with pipeline, we need to store original if any of the destinations requires original let store_original_when_pipeline_exists = !executable_pipelines.is_empty() && streams_need_original_map.values().any(|val| *val); // End get user defined schema - let flatten_level = get_flatten_level(org_id, &stream_name, stream_type).await; + let flatten_level = match &bulk_columnar { + Some(columnar) => columnar.flatten_level(), + None => get_flatten_level(org_id, &stream_name, stream_type).await, + }; - // JsonColumnar::plan already refuses the fast path under vectorscan, before this is consulted. - let needs_json_records = !executable_pipelines.is_empty() - || extend_json.is_some() - || matches!(user_defined_schema_map.get(&stream_name), Some(Some(_))) - || streams_need_original_map - .get(&stream_name) - .is_some_and(|v| *v) - || streams_need_all_values_map - .get(&stream_name) - .is_some_and(|v| *v); - let mut columnar = if matches!(in_req, IngestionRequest::JSON(_)) && !needs_json_records { + let mut columnar = if bulk_columnar.is_some() { + bulk_columnar + } else if matches!(in_req, IngestionRequest::JSON(_)) && !needs_json_records { JsonColumnar::plan(org_id, &stream_name, flatten_level, (min_ts, max_ts)).await } else { None @@ -842,25 +962,27 @@ pub async fn ingest( ) } -/// Flatten a record and resolve its timestamp — the only two steps of record -/// preparation that can reject an event. -/// -/// Shared with the HEC collector's pre-write validation pass, which must reach -/// the same verdict as the write does; a second implementation would drift. -pub fn prepare_record( - item: json::Value, - flatten_level: u32, - min_ts: i64, - max_ts: i64, -) -> std::result::Result<(json::Value, i64), PrepareRecordError> { - let mut res = - flatten::flatten_with_level(item, flatten_level).map_err(PrepareRecordError::Flatten)?; - match handle_timestamp_for_value(&mut res, min_ts, max_ts) { - Ok(ts) => Ok((res, ts)), - Err(e) => Err(PrepareRecordError::Timestamp(res, e)), +/// The snapshot a request runs under; a `_bulk` stream's whole batch keeps its plan's. +async fn request_needs( + stream_param: StreamParams, + bulk_plan: Option, +) -> (StreamRecordNeeds, Option) { + match bulk_plan { + Some(BulkPlan { columnar, needs }) => (needs, Some(columnar)), + None => (StreamRecordNeeds::load(stream_param).await, None), } } +/// The name the writer uses for `in_stream_name`; `skip_formatting_stream_name` looks it up. +pub(super) async fn resolve_stream_name(org_id: &str, in_stream_name: &str) -> Result { + Ok(if config::get_config().common.skip_formatting_stream_name { + get_formatted_stream_name(StreamParams::new(org_id, in_stream_name, StreamType::Logs)) + .await? + } else { + format_stream_name(in_stream_name.to_string()) + }) +} + /// A body is an array of records or a single record. fn parse_json_body(body: &[u8]) -> Result> { match json::from_slice(body) { @@ -1491,16 +1613,23 @@ fn construct_values_from_open_telemetry_v1_metric( #[cfg(test)] mod tests { - use config::meta::self_reporting::{ - redaction::REDACTION_EVIDENCE_STREAM, - usage::{ - AUDIT_STREAM, DATA_RETENTION_USAGE_STREAM, ERROR_STREAM, STATS_STREAM, TRIGGERS_STREAM, - USAGE_STREAM, + use config::meta::{ + pipeline::{ + Pipeline, PipelineKind, + components::{Edge, Node, NodeData, PipelineSource}, + }, + self_reporting::{ + redaction::REDACTION_EVIDENCE_STREAM, + usage::{ + AUDIT_STREAM, DATA_RETENTION_USAGE_STREAM, ERROR_STREAM, STATS_STREAM, + TRIGGERS_STREAM, USAGE_STREAM, + }, }, }; use ingestion_common::{IngestUser, SystemJobType}; use super::*; + use crate::cache::STREAM_EXECUTABLE_PIPELINES; /// A bulk ingest returned 200 while dropping every record as "Too old data", and /// nothing server-side said so. The status code has to stay 200 — clients and @@ -2065,4 +2194,69 @@ mod tests { } assert_eq!(decoded, body_cap / (1024 * 1024) + 2); } + + fn realtime_pipeline(source_stream: &StreamParams) -> Pipeline { + let node = |id: &str, stream: StreamParams| { + Node::new( + id.to_string(), + NodeData::Stream(stream), + 0.0, + 0.0, + "input".to_string(), + ) + }; + let dest = StreamParams::new(&source_stream.org_id, "dest_stream", StreamType::Logs); + Pipeline { + id: "bulk_snapshot_pl".to_string(), + version: 1, + enabled: true, + org: source_stream.org_id.to_string(), + name: "bulk_snapshot_pl".to_string(), + description: String::new(), + source: PipelineSource::Realtime(source_stream.clone()), + nodes: vec![node("source", source_stream.clone()), node("dest", dest)], + edges: vec![Edge::new("source".to_string(), "dest".to_string())], + kind: PipelineKind::User, + } + } + + // regression: a pipeline enabled between a `_bulk` plan and its ingest reached only the + // fallback rows + #[tokio::test] + async fn test_bulk_batch_keeps_the_snapshot_it_was_planned_under() { + let stream = StreamParams::new( + "bulk_snapshot_org", + "bulk_snapshot_stream", + StreamType::Logs, + ); + let needs = StreamRecordNeeds::load(stream.clone()).await; + assert!(!needs.json_records(&stream.stream_name)); + let schema = arrow_schema::Schema::new(vec![arrow_schema::Field::new( + TIMESTAMP_COL_NAME, + arrow_schema::DataType::Int64, + false, + )]); + let columnar = + JsonColumnar::for_schema(&stream.stream_name, &schema, 0, (0, i64::MAX), false) + .unwrap(); + let plan = BulkPlan { columnar, needs }; + + let pipeline = ExecutablePipeline::new(&realtime_pipeline(&stream)) + .await + .unwrap(); + STREAM_EXECUTABLE_PIPELINES + .write() + .await + .insert(stream.clone(), vec![pipeline]); + let reloaded = StreamRecordNeeds::load(stream.clone()).await; + let (needs, columnar) = request_needs(stream.clone(), Some(plan)).await; + STREAM_EXECUTABLE_PIPELINES.write().await.remove(&stream); + + assert!(reloaded.json_records(&stream.stream_name)); + assert!(columnar.is_some()); + assert!(needs.executable_pipelines.is_empty()); + let (needs, columnar) = request_needs(stream.clone(), None).await; + assert!(columnar.is_none()); + assert!(needs.executable_pipelines.is_empty()); + } } diff --git a/src/core/src/logs/mod.rs b/src/core/src/logs/mod.rs index 3124c8af84..66192c7501 100644 --- a/src/core/src/logs/mod.rs +++ b/src/core/src/logs/mod.rs @@ -598,15 +598,13 @@ async fn write_logs( status.successful += 1; } IngestionStatus::Bulk(bulk_res) => { - bulk::add_record_status( - stream_name.to_string(), - doc_id, - "".to_string(), - None, - bulk_res, - None, - None, - ); + if !cfg.common.bulk_api_response_errors_only { + bulk_res.items.push(bulk::success_item( + "index".to_string(), + stream_name, + doc_id.unwrap_or_default(), + )); + } } } } @@ -636,10 +634,14 @@ async fn write_logs( fn columnar_entries( org_id: &str, status: &mut IngestionStatus, - columnar: columnar::JsonColumnar, + mut columnar: columnar::JsonColumnar, ) -> Result> { - if let IngestionStatus::Record(status) = status { - status.successful += columnar.rows() as u32; + match status { + IngestionStatus::Record(status) => status.successful += columnar.rows() as u32, + IngestionStatus::Bulk(bulk_res) => { + let doc_ids = columnar.take_bulk_ids(); + bulk::add_success_items(columnar.stream_name(), doc_ids, bulk_res); + } } columnar.into_entries(org_id) }