starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-04 16:09:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(logs): route _bulk data lines through the direct-to-Arrow columnar path (#15592)

Browse Source
This commit is contained in:
Hengfei Yang
2026-10-21 12:49:28 +00:00
committed by GitHub
parent d7e9aaf6c7
commit 8a4c5752b9
5 changed files with 697 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/src/logs/bulk.rs
+165 -75
View File
@@ -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<String> = None;
let stream_type = StreamType::Logs;
let mut stream_key_cache: HashMap<String, String> = HashMap::new();
let mut streams_data: HashMap<String, Vec<json::Value>> = HashMap::new();
let mut plans: HashMap<String, Option<BulkPlan>> = 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<String, BulkResponseItem> {
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<Option<String>>,
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<String, json::Value>,
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<anyhow::Result<(i64, bool)>>,
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<String, Option<BulkPlan>>,
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
src/core/src/logs/columnar.rs
+248 -7
View File
@@ -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<Option<String>>,
bulk_id_col: Option<usize>,
}
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<Option<String>> {
std::mem::take(&mut self.bulk_ids)
}
pub fn into_entries(self, org_id: &str) -> Result<Vec<ingester::Entry>> {
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<String, usize> = 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<u32>,
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<A: MapAccess<'de>>(
&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::Value> =
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<json::Value>,
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 [
src/core/src/logs/hec.rs
+6 -19
View File
@@ -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<i64, HecParseError
Ok(micros.round() as i64)
}
/// Resolve an index to the stream name the writer will actually use.
///
/// Note the branch reads inverted against its own name: `skip_formatting_stream_name`
/// selects `get_formatted_stream_name`. This mirrors `logs::ingest::ingest` so the
/// grouping key matches what is written.
/// Resolve an index to the stream name the writer will actually use, so grouping matches it.
async fn resolve_stream_name(
org_id: &str,
index: &str,
) -> std::result::Result<String, HecParseError> {
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);
}
src/core/src/logs/ingest.rs
+264 -70
View File
@@ -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<ExecutablePipeline>,
user_defined_schema_map: HashMap<String, Option<HashSet<String>>>,
streams_need_original_map: HashMap<String, bool>,
streams_need_all_values_map: HashMap<String, bool>,
}
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<String, serde_json::Value>>,
is_derived: bool,
) -> Result<IngestionResponse> {
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<json::Value>,
plan: Option<BulkPlan>,
user: IngestUser,
) -> Result<IngestionResponse> {
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<BulkPlan> {
// 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<String, serde_json::Value>>,
is_derived: bool,
bulk_plan: Option<BulkPlan>,
) -> Result<IngestionResponse> {
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<String, Option<HashSet<String>>> = HashMap::new();
let mut streams_need_original_map: HashMap<String, bool> = HashMap::new();
let mut streams_need_all_values_map: HashMap<String, bool> = 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<BulkPlan>,
) -> (StreamRecordNeeds, Option<JsonColumnar>) {
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<String> {
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<Vec<json::Value>> {
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());
}
}
src/core/src/logs/mod.rs
+14 -12
View File
@@ -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<Vec<ingester::Entry>> {
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)
}