From ebdc19d02e1e3860dfad73a69b95e0ff059abd21 Mon Sep 17 00:00:00 2001 From: Ashish Kolhe <35160958+oasisk@users.noreply.github.com> Date: Fri, 19 Apr 2024 06:00:56 +0530 Subject: [PATCH] feat: routing & user defined schema restrictions (#3254) Co-authored-by: omkark06 --- src/config/src/config.rs | 2 + src/config/src/meta/stream.rs | 181 ++++++++++ src/handler/http/request/stream/mod.rs | 4 +- src/infra/src/schema/mod.rs | 17 + src/service/ingestion/mod.rs | 118 +++++-- src/service/logs/bulk.rs | 148 ++++++--- src/service/logs/ingest.rs | 12 +- src/service/logs/mod.rs | 20 ++ src/service/logs/multi.rs | 12 +- src/service/logs/otlp_grpc.rs | 20 +- src/service/logs/otlp_http.rs | 12 +- src/service/logs/syslog.rs | 12 +- src/service/metadata/trace_list_index.rs | 3 + src/service/metrics/json.rs | 4 +- src/service/metrics/otlp_grpc.rs | 27 +- src/service/metrics/otlp_http.rs | 24 +- src/service/metrics/prom.rs | 19 +- src/service/schema.rs | 3 + src/service/traces/mod.rs | 14 +- src/service/traces/otlp_http.rs | 14 +- web/src/components/logstream/AddStream.vue | 9 + .../logstream/StreamFieldInputs.vue | 20 +- web/src/components/logstream/schema.vue | 308 ++++++++++++------ web/src/locales/languages/en.json | 3 +- 24 files changed, 760 insertions(+), 246 deletions(-) diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 8a4094649a..fab4e6d542 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -626,6 +626,8 @@ pub struct Common { pub report_user_name: String, #[env_config(name = "ZO_REPORT_USER_PASSWORD", default = "")] pub report_user_password: String, + #[env_config(name = "ZO_CONCATENATED_SCHEMA_FIELD_NAME", default = "_all")] + pub all_fields_name: String, } #[derive(EnvConfig)] diff --git a/src/config/src/meta/stream.rs b/src/config/src/meta/stream.rs index ec6ab63eea..b417f046f1 100644 --- a/src/config/src/meta/stream.rs +++ b/src/config/src/meta/stream.rs @@ -27,6 +27,7 @@ use crate::{ utils::{ hash::{gxhash, Sum64}, json, + json::{Map, Value}, }, CONFIG, }; @@ -410,6 +411,12 @@ pub struct StreamSettings { pub bloom_filter_fields: Vec, #[serde(default)] pub data_retention: i64, + #[serde(skip_serializing_if = "Option::None")] + pub routing: Option>>, + #[serde(skip_serializing_if = "Option::None")] + pub flatten_level: Option, + #[serde(skip_serializing_if = "Option::None")] + pub defined_schema_fields: Option>, } impl Serialize for StreamSettings { @@ -430,6 +437,30 @@ impl Serialize for StreamSettings { state.serialize_field("full_text_search_keys", &self.full_text_search_keys)?; state.serialize_field("bloom_filter_fields", &self.bloom_filter_fields)?; state.serialize_field("data_retention", &self.data_retention)?; + match self.routing.as_ref() { + Some(routing) => { + state.serialize_field("routing", routing)?; + } + None => { + state.skip_field("routing")?; + } + } + match self.defined_schema_fields.as_ref() { + Some(defined_schema_fields) => { + state.serialize_field("defined_schema_fields", defined_schema_fields)?; + } + None => { + state.skip_field("defined_schema_fields")?; + } + } + match self.flatten_level.as_ref() { + Some(flatten_level) => { + state.serialize_field("flatten_level", flatten_level)?; + } + None => { + state.skip_field("flatten_level")?; + } + } state.end() } } @@ -485,12 +516,41 @@ impl From<&str> for StreamSettings { data_retention = v.as_i64().unwrap(); }; + let mut defined_schema_fields: Option> = None; + if let Some(value) = settings.get("defined_schema_fields") { + defined_schema_fields = Some( + value + .as_array() + .unwrap() + .iter() + .map(|item| item.as_str().unwrap().to_string()) + .collect(), + ); + } + + let mut routing: HashMap> = HashMap::new(); + let routes = settings.get("routing"); + if let Some(value) = routes { + let v: Vec<_> = value.as_object().unwrap().iter().collect(); + for item in v { + routing.insert( + item.0.to_string(), + json::from_value(item.1.clone()).unwrap(), + ); + } + } + + let flatten_level = settings.get("flatten_level").map(|v| v.as_i64().unwrap()); + Self { partition_keys, partition_time_level, full_text_search_keys, bloom_filter_fields, data_retention, + routing: Some(routing), + flatten_level, + defined_schema_fields, } } } @@ -561,6 +621,127 @@ pub struct PartitioningDetails { pub partition_time_level: Option, } +pub struct Routing { + pub destination: String, + pub routing: Vec, +} + +// Code Duplicated from alerts +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +pub struct RoutingCondition { + pub column: String, + pub operator: Operator, + #[schema(value_type = Object)] + pub value: Value, + #[serde(default)] + pub ignore_case: bool, +} +// Code Duplicated from alerts +impl RoutingCondition { + pub async fn evaluate(&self, row: &Map) -> bool { + let val = match row.get(&self.column) { + Some(val) => val, + None => { + return false; + } + }; + match val { + Value::String(v) => { + let val = v.as_str(); + let con_val = self.value.as_str().unwrap_or_default(); + match self.operator { + Operator::EqualTo => val == con_val, + Operator::NotEqualTo => val != con_val, + Operator::GreaterThan => val > con_val, + Operator::GreaterThanEquals => val >= con_val, + Operator::LessThan => val < con_val, + Operator::LessThanEquals => val <= con_val, + Operator::Contains => val.contains(con_val), + Operator::NotContains => !val.contains(con_val), + } + } + Value::Number(_) => { + let val = val.as_f64().unwrap_or_default(); + let con_val = if self.value.is_number() { + self.value.as_f64().unwrap_or_default() + } else { + self.value + .as_str() + .unwrap_or_default() + .parse() + .unwrap_or_default() + }; + match self.operator { + Operator::EqualTo => val == con_val, + Operator::NotEqualTo => val != con_val, + Operator::GreaterThan => val > con_val, + Operator::GreaterThanEquals => val >= con_val, + Operator::LessThan => val < con_val, + Operator::LessThanEquals => val <= con_val, + _ => false, + } + } + Value::Bool(v) => { + let val = v.to_owned(); + let con_val = if self.value.is_boolean() { + self.value.as_bool().unwrap_or_default() + } else { + self.value + .as_str() + .unwrap_or_default() + .parse() + .unwrap_or_default() + }; + match self.operator { + Operator::EqualTo => val == con_val, + Operator::NotEqualTo => val != con_val, + _ => false, + } + } + _ => false, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +pub enum Operator { + #[serde(rename = "=")] + EqualTo, + #[serde(rename = "!=")] + NotEqualTo, + #[serde(rename = ">")] + GreaterThan, + #[serde(rename = ">=")] + GreaterThanEquals, + #[serde(rename = "<")] + LessThan, + #[serde(rename = "<=")] + LessThanEquals, + Contains, + NotContains, +} + +impl Default for Operator { + fn default() -> Self { + Self::EqualTo + } +} + +impl std::fmt::Display for Operator { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Operator::EqualTo => write!(f, "="), + Operator::NotEqualTo => write!(f, "!="), + Operator::GreaterThan => write!(f, ">"), + Operator::GreaterThanEquals => write!(f, ">="), + Operator::LessThan => write!(f, "<"), + Operator::LessThanEquals => write!(f, "<="), + Operator::Contains => write!(f, "contains"), + Operator::NotContains => write!(f, "not contains"), + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/handler/http/request/stream/mod.rs b/src/handler/http/request/stream/mod.rs index ebc00ddff8..a9ba259812 100644 --- a/src/handler/http/request/stream/mod.rs +++ b/src/handler/http/request/stream/mod.rs @@ -101,11 +101,11 @@ async fn settings( let stream_type = match get_stream_type_from_request(&query) { Ok(v) => { if let Some(s_type) = v { - if s_type == StreamType::EnrichmentTables { + if s_type == StreamType::EnrichmentTables || s_type == StreamType::Index { return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( http::StatusCode::BAD_REQUEST.into(), - "Stream type 'EnrichmentTable' not allowed".to_string(), + format!("Stream type '{}' not allowed", s_type), )), ); } diff --git a/src/infra/src/schema/mod.rs b/src/infra/src/schema/mod.rs index da1bf8953a..a264eefe75 100644 --- a/src/infra/src/schema/mod.rs +++ b/src/infra/src/schema/mod.rs @@ -498,6 +498,23 @@ pub async fn delete_fields( } }) .collect::>(); + + let mut settings = unwrap_stream_settings(&latest_schema).unwrap_or_default(); + + if let Some(schema_fields) = settings.defined_schema_fields { + let defined_schema_fields = schema_fields + .iter() + .filter_map(|f| { + if deleted_fields.contains(f) { + None + } else { + Some(f.clone()) + } + }) + .collect::>(); + settings.defined_schema_fields = Some(defined_schema_fields); + }; + new_metadata.insert("settings".to_string(), json::to_string(&settings).unwrap()); let new_schema = vec![Schema::new_with_metadata(fields, new_metadata)]; Ok(Some(( Some(json::to_vec(&prev_schema).unwrap().into()), diff --git a/src/service/ingestion/mod.rs b/src/service/ingestion/mod.rs index ffe70104e5..3def98ae6a 100644 --- a/src/service/ingestion/mod.rs +++ b/src/service/ingestion/mod.rs @@ -23,7 +23,7 @@ use chrono::{TimeZone, Utc}; use config::{ cluster, meta::{ - stream::{PartitionTimeLevel, PartitioningDetails, StreamPartition, StreamType}, + stream::{PartitionTimeLevel, PartitioningDetails, Routing, StreamPartition, StreamType}, usage::RequestStats, }, utils::{ @@ -44,7 +44,7 @@ use crate::{ meta::{ alerts::Alert, functions::{StreamTransform, VRLResultResolver, VRLRuntimeConfig}, - stream::SchemaRecords, + stream::{SchemaRecords, StreamParams}, }, utils::functions::get_vrl_compiler_config, }, @@ -115,21 +115,27 @@ pub fn apply_vrl_fn(runtime: &mut Runtime, vrl_runtime: &VRLResultResolver, row: } } -pub async fn get_stream_transforms<'a>( - org_id: &str, - stream_type: &StreamType, - stream_name: &str, - stream_transform_map: &mut HashMap>, +pub async fn get_stream_functions<'a>( + streams: &[StreamParams], + stream_functions_map: &mut HashMap>, stream_vrl_map: &mut HashMap, ) { - let key = format!("{}/{}/{}", org_id, stream_type, stream_name); - if stream_transform_map.contains_key(&key) { - return; + for stream in streams { + let key = format!( + "{}/{}/{}", + stream.org_id, stream.stream_type, stream.stream_name + ); + if stream_functions_map.contains_key(&key) { + return; + } + let mut _local_trans: Vec = vec![]; + (_local_trans, *stream_vrl_map) = crate::service::ingestion::register_stream_functions( + &stream.org_id, + &stream.stream_type, + &stream.stream_name, + ); + stream_functions_map.insert(key, _local_trans); } - let mut _local_trans: Vec = vec![]; - (_local_trans, *stream_vrl_map) = - crate::service::ingestion::register_stream_transforms(org_id, stream_type, stream_name); - stream_transform_map.insert(key, _local_trans); } pub async fn get_stream_partition_keys( @@ -147,29 +153,32 @@ pub async fn get_stream_partition_keys( } pub async fn get_stream_alerts( - org_id: &str, - stream_type: &StreamType, - stream_name: &str, + streams: &[StreamParams], stream_alerts_map: &mut HashMap>, ) { - let key = format!("{}/{}/{}", org_id, stream_type, stream_name); - if stream_alerts_map.contains_key(&key) { - return; - } + for stream in streams { + let key = format!( + "{}/{}/{}", + stream.org_id, stream.stream_type, stream.stream_name + ); + if stream_alerts_map.contains_key(&key) { + return; + } - let alerts_cacher = STREAM_ALERTS.read().await; - let alerts_list = alerts_cacher.get(&key); - if alerts_list.is_none() { - return; - } - let alerts = alerts_list - .unwrap() - .iter() - .filter(|alert| alert.enabled && alert.is_real_time) - .cloned() - .collect::>(); + let alerts_cacher = STREAM_ALERTS.read().await; + let alerts_list = alerts_cacher.get(&key); + if alerts_list.is_none() { + return; + } + let alerts = alerts_list + .unwrap() + .iter() + .filter(|alert| alert.enabled && alert.is_real_time) + .cloned() + .collect::>(); - stream_alerts_map.insert(key, alerts); + stream_alerts_map.insert(key, alerts); + } } pub async fn evaluate_trigger(trigger: Option) { @@ -223,7 +232,7 @@ pub fn get_wal_time_key( time_key } -pub fn register_stream_transforms( +pub fn register_stream_functions( org_id: &str, stream_type: &StreamType, stream_name: &str, @@ -258,7 +267,7 @@ pub fn register_stream_transforms( (local_trans, stream_vrl_map) } -pub fn apply_stream_transform( +pub fn apply_stream_functions( local_trans: &[StreamTransform], mut value: Value, stream_vrl_map: &HashMap, @@ -451,6 +460,45 @@ pub fn get_val_with_type_retained(val: &Value) -> Value { } } +pub async fn get_stream_routing( + stream_params: StreamParams, + stream_routing_map: &mut HashMap>, +) { + let stream_settings = infra::schema::get_settings( + &stream_params.org_id, + &stream_params.stream_name, + stream_params.stream_type, + ) + .await + .unwrap_or_default(); + let res: Vec = stream_settings + .routing + .unwrap_or_default() + .iter() + .map(|(k, v)| Routing { + destination: k.to_string(), + routing: v.clone(), + }) + .collect(); + + stream_routing_map.insert(stream_params.stream_name.to_string(), res); +} + +pub async fn get_user_defined_schema( + streams: &[StreamParams], + user_defined_schema_map: &mut HashMap>, +) { + for stream in streams { + let stream_settings = + infra::schema::get_settings(&stream.org_id, &stream.stream_name, stream.stream_type) + .await + .unwrap_or_default(); + if let Some(fields) = stream_settings.defined_schema_fields { + user_defined_schema_map.insert(stream.stream_name.to_string(), fields); + } + } +} + #[cfg(test)] mod tests { use infra::schema::{unwrap_stream_settings, STREAM_SETTINGS}; diff --git a/src/service/logs/bulk.rs b/src/service/logs/bulk.rs index a20288b080..fd2bf70519 100644 --- a/src/service/logs/bulk.rs +++ b/src/service/logs/bulk.rs @@ -25,7 +25,7 @@ use chrono::{Duration, Utc}; use config::{ cluster, meta::{ - stream::{PartitioningDetails, StreamType}, + stream::{PartitioningDetails, Routing, StreamType}, usage::UsageType, }, metrics, @@ -97,7 +97,7 @@ pub async fn ingest( let mut stream_schema_map: HashMap = HashMap::new(); let mut stream_data_map = HashMap::new(); - let mut stream_transform_map: HashMap> = HashMap::new(); + let mut stream_functions_map: HashMap> = HashMap::new(); let mut stream_partition_keys_map: HashMap = HashMap::new(); let mut stream_alerts_map: HashMap> = HashMap::new(); @@ -110,6 +110,10 @@ pub async fn ingest( let mut blocked_stream_warnings: HashMap = HashMap::new(); + let mut stream_routing_map: HashMap> = HashMap::new(); + + let mut user_defined_schema_map: HashMap> = HashMap::new(); + let mut next_line_is_data = false; let reader = BufReader::new(body.as_ref()); for line in reader.lines() { @@ -139,46 +143,76 @@ pub async fn ingest( continue; // skip } + // Start get routing keys + crate::service::ingestion::get_stream_routing( + StreamParams { + org_id: org_id.to_owned().into(), + stream_type: StreamType::Logs, + stream_name: stream_name.to_owned().into(), + }, + &mut stream_routing_map, + ) + .await; + + let mut streams = vec![StreamParams { + org_id: org_id.to_owned().into(), + stream_type: StreamType::Logs, + stream_name: stream_name.to_owned().into(), + }]; + + if let Some(routes) = stream_routing_map.get(&stream_name) { + for route in routes { + streams.push(StreamParams { + org_id: org_id.to_owned().into(), + stream_type: StreamType::Logs, + stream_name: route.destination.clone().into(), + }); + } + } + + // End get stream keys + + crate::service::ingestion::get_user_defined_schema( + &streams, + &mut user_defined_schema_map, + ) + .await; + next_line_is_data = true; - // Start Register Transfoms for stream - - crate::service::ingestion::get_stream_transforms( - org_id, - &StreamType::Logs, - &stream_name, - &mut stream_transform_map, + // Start Register functions for stream + crate::service::ingestion::get_stream_functions( + &streams, + &mut stream_functions_map, &mut stream_vrl_map, ) .await; - // End Register Transfoms for index + // End Register functions for index // Start get stream alerts - crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Logs, - &stream_name, - &mut stream_alerts_map, - ) - .await; + crate::service::ingestion::get_stream_alerts(&streams, &mut stream_alerts_map).await; // End get stream alert - if !stream_partition_keys_map.contains_key(&stream_name) { - let stream_schema = stream_schema_exists( - org_id, - &stream_name, - StreamType::Logs, - &mut stream_schema_map, - ) - .await; - let partition_det = crate::service::ingestion::get_stream_partition_keys( - org_id, - &StreamType::Logs, - &stream_name, - ) - .await; - stream_partition_keys_map - .insert(stream_name.clone(), (stream_schema, partition_det)); + for stream in streams { + let local_stream_name = stream.stream_name.to_string(); + if let std::collections::hash_map::Entry::Vacant(e) = + stream_partition_keys_map.entry(local_stream_name.to_owned()) + { + let stream_schema = stream_schema_exists( + org_id, + &local_stream_name, + StreamType::Logs, + &mut stream_schema_map, + ) + .await; + let partition_det = crate::service::ingestion::get_stream_partition_keys( + org_id, + &StreamType::Logs, + &local_stream_name, + ) + .await; + e.insert((stream_schema, partition_det)); + } } stream_data_map @@ -189,19 +223,43 @@ pub async fn ingest( } else { next_line_is_data = false; - let stream_data = stream_data_map.get_mut(&stream_name).unwrap(); - let buf = &mut stream_data.data; - - // Start row based transform - - let key = format!("{org_id}/{}/{stream_name}", StreamType::Logs); - // JSON Flattening let mut value = flatten::flatten_with_level(value, CONFIG.limit.ingest_flatten_level)?; - if let Some(transforms) = stream_transform_map.get(&key) { + if let Some(routing) = stream_routing_map.get(&stream_name) { + if !routing.is_empty() { + for route in routing { + let mut is_routed = true; + let val = &route.routing; + for q_condition in val.iter() { + is_routed = + is_routed && q_condition.evaluate(value.as_object().unwrap()).await; + } + if is_routed && !val.is_empty() { + stream_name = route.destination.clone(); + if !stream_data_map.contains_key(&stream_name) { + stream_data_map.insert( + stream_name.clone(), + BulkStreamData { + data: HashMap::new(), + }, + ); + } + break; + } + } + } + } + + let stream_data = stream_data_map.get_mut(&stream_name).unwrap(); + let buf = &mut stream_data.data; + + let key = format!("{org_id}/{}/{stream_name}", StreamType::Logs); + + // Start row based transform + if let Some(transforms) = stream_functions_map.get(&key) { let mut ret_value = value.clone(); - ret_value = crate::service::ingestion::apply_stream_transform( + ret_value = crate::service::ingestion::apply_stream_functions( transforms, ret_value, &stream_vrl_map, @@ -233,6 +291,10 @@ pub async fn ingest( _ => unreachable!(), }; + if let Some(fields) = user_defined_schema_map.get(&stream_name) { + crate::service::logs::refactor_map(&mut local_val, fields); + } + // set _id if !doc_id.is_empty() { local_val.insert("_id".to_string(), json::Value::String(doc_id.clone())); @@ -423,7 +485,7 @@ pub async fn ingest( req_stats.response_time += time; req_stats.user_email = Some(user_email.to_string()); // metric + data usage - let fns_length: usize = stream_transform_map.values().map(|v| v.len()).sum(); + let fns_length: usize = stream_functions_map.values().map(|v| v.len()).sum(); report_request_usage_stats( req_stats, org_id, diff --git a/src/service/logs/ingest.rs b/src/service/logs/ingest.rs index d26cd4d79f..c8e6e40195 100644 --- a/src/service/logs/ingest.rs +++ b/src/service/logs/ingest.rs @@ -79,7 +79,7 @@ pub async fn ingest( // Start Register Transforms for stream let mut runtime = crate::service::ingestion::init_functions_runtime(); - let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms( + let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_functions( org_id, &StreamType::Logs, stream_name, @@ -89,9 +89,11 @@ pub async fn ingest( // Start get stream alerts let mut stream_alerts_map: HashMap> = HashMap::new(); crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Logs, - stream_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: stream_name.to_owned().into(), + stream_type: StreamType::Logs, + }], &mut stream_alerts_map, ) .await; @@ -285,7 +287,7 @@ pub fn apply_functions<'a>( let mut value = flatten::flatten_with_level(item, CONFIG.limit.ingest_flatten_level)?; if !local_trans.is_empty() { - value = crate::service::ingestion::apply_stream_transform( + value = crate::service::ingestion::apply_stream_functions( local_trans, value, stream_vrl_map, diff --git a/src/service/logs/mod.rs b/src/service/logs/mod.rs index 36e13a442b..7b84036603 100644 --- a/src/service/logs/mod.rs +++ b/src/service/logs/mod.rs @@ -447,6 +447,26 @@ struct StreamMeta<'a> { stream_alerts_map: &'a HashMap>, } +pub fn refactor_map(original_map: &mut Map, defined_schema_keys: &[String]) { + let mut non_schema_map = HashMap::new(); + + for (key, value) in original_map.iter() { + if !defined_schema_keys.contains(key) { + non_schema_map.insert(key.clone(), get_string_value(value)); + } + } + + original_map.retain(|key, _| defined_schema_keys.contains(key)); + + if !non_schema_map.is_empty() { + let non_schema_json = serde_json::to_string(&non_schema_map).unwrap_or_default(); + original_map.insert( + CONFIG.common.all_fields_name.to_string(), + Value::String(non_schema_json), + ); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/service/logs/multi.rs b/src/service/logs/multi.rs index d533aedc55..3f689229d7 100644 --- a/src/service/logs/multi.rs +++ b/src/service/logs/multi.rs @@ -87,7 +87,7 @@ async fn ingest_inner( let mut trigger: Option = None; // Start Register Transforms for stream - let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms( + let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_functions( org_id, &StreamType::Logs, stream_name, @@ -105,9 +105,11 @@ async fn ingest_inner( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Logs, - stream_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: stream_name.to_owned().into(), + stream_type: StreamType::Logs, + }], &mut stream_alerts_map, ) .await; @@ -132,7 +134,7 @@ async fn ingest_inner( // Start row based transform if !local_trans.is_empty() { - value = crate::service::ingestion::apply_stream_transform( + value = crate::service::ingestion::apply_stream_functions( &local_trans, value, &stream_vrl_map, diff --git a/src/service/logs/otlp_grpc.rs b/src/service/logs/otlp_grpc.rs index aee8441e1e..b8646b68a9 100644 --- a/src/service/logs/otlp_grpc.rs +++ b/src/service/logs/otlp_grpc.rs @@ -102,9 +102,11 @@ pub async fn usage_ingest( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Logs, - stream_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: stream_name.to_owned().into(), + stream_type: StreamType::Logs, + }], &mut stream_alerts_map, ) .await; @@ -309,16 +311,18 @@ pub async fn handle_grpc_request( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Logs, - stream_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: stream_name.to_owned().into(), + stream_type: StreamType::Logs, + }], &mut stream_alerts_map, ) .await; // End get stream alert // Start Register Transforms for stream - let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms( + let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_functions( org_id, &StreamType::Logs, stream_name, @@ -415,7 +419,7 @@ pub async fn handle_grpc_request( rec = flatten::flatten_with_level(rec, CONFIG.limit.ingest_flatten_level)?; if !local_trans.is_empty() { - rec = crate::service::ingestion::apply_stream_transform( + rec = crate::service::ingestion::apply_stream_functions( &local_trans, rec, &stream_vrl_map, diff --git a/src/service/logs/otlp_http.rs b/src/service/logs/otlp_http.rs index fc6ed79c99..d3f7a94d22 100644 --- a/src/service/logs/otlp_http.rs +++ b/src/service/logs/otlp_http.rs @@ -160,16 +160,18 @@ pub async fn logs_json_handler( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Logs, - stream_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: stream_name.to_owned().into(), + stream_type: StreamType::Logs, + }], &mut stream_alerts_map, ) .await; // End get stream alert // Start Register Transforms for stream - let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms( + let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_functions( org_id, &StreamType::Logs, stream_name, @@ -356,7 +358,7 @@ pub async fn logs_json_handler( flatten::flatten_with_level(value, CONFIG.limit.ingest_flatten_level).unwrap(); if !local_trans.is_empty() { - value = crate::service::ingestion::apply_stream_transform( + value = crate::service::ingestion::apply_stream_functions( &local_trans, value, &stream_vrl_map, diff --git a/src/service/logs/syslog.rs b/src/service/logs/syslog.rs index 19e7cd909f..fd5b5988d9 100644 --- a/src/service/logs/syslog.rs +++ b/src/service/logs/syslog.rs @@ -108,16 +108,18 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result { // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Logs, - stream_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: stream_name.to_owned().into(), + stream_type: StreamType::Logs, + }], &mut stream_alerts_map, ) .await; // End get stream alert // Start Register Transforms for stream - let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms( + let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_functions( org_id, &StreamType::Logs, stream_name, @@ -131,7 +133,7 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result { value = flatten::flatten_with_level(value, CONFIG.limit.ingest_flatten_level).unwrap(); if !local_trans.is_empty() { - value = crate::service::ingestion::apply_stream_transform( + value = crate::service::ingestion::apply_stream_functions( &local_trans, value, &stream_vrl_map, diff --git a/src/service/metadata/trace_list_index.rs b/src/service/metadata/trace_list_index.rs index 9f26000300..1494984f23 100644 --- a/src/service/metadata/trace_list_index.rs +++ b/src/service/metadata/trace_list_index.rs @@ -184,6 +184,9 @@ impl TraceListIndex { full_text_search_keys: vec![], bloom_filter_fields: vec!["trace_id".to_string()], data_retention: 0, + routing: None, + flatten_level: None, + defined_schema_fields: None, }; stream::save_stream_settings(org_id, STREAM_NAME, StreamType::Metadata, settings) diff --git a/src/service/metrics/json.rs b/src/service/metrics/json.rs index f1367442c0..c27e7f1734 100644 --- a/src/service/metrics/json.rs +++ b/src/service/metrics/json.rs @@ -357,13 +357,13 @@ fn apply_func( metric_name: &str, value: json::Value, ) -> Result { - let (local_tans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms( + let (local_tans, stream_vrl_map) = crate::service::ingestion::register_stream_functions( org_id, &StreamType::Metrics, metric_name, ); - crate::service::ingestion::apply_stream_transform( + crate::service::ingestion::apply_stream_functions( &local_tans, value, &stream_vrl_map, diff --git a/src/service/metrics/otlp_grpc.rs b/src/service/metrics/otlp_grpc.rs index 5c42139d62..3894610ffc 100644 --- a/src/service/metrics/otlp_grpc.rs +++ b/src/service/metrics/otlp_grpc.rs @@ -38,7 +38,10 @@ use prost::Message; use crate::{ common::meta::{ - alerts, http::HttpResponse as MetaHttpResponse, prom::*, stream::SchemaRecords, + alerts, + http::HttpResponse as MetaHttpResponse, + prom::*, + stream::{SchemaRecords, StreamParams}, }, service::{ db, format_stream_name, @@ -127,9 +130,11 @@ pub async fn handle_grpc_request( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Metrics, - metric_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: metric_name.to_owned().into(), + stream_type: StreamType::Metrics, + }], &mut stream_alerts_map, ) .await; @@ -137,7 +142,7 @@ pub async fn handle_grpc_request( // Start Register Transforms for stream let (mut local_trans, mut stream_vrl_map) = - crate::service::ingestion::register_stream_transforms( + crate::service::ingestion::register_stream_functions( org_id, &StreamType::Metrics, metric_name, @@ -248,9 +253,11 @@ pub async fn handle_grpc_request( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Metrics, - local_metric_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: local_metric_name.to_owned().into(), + stream_type: StreamType::Metrics, + }], &mut stream_alerts_map, ) .await; @@ -258,7 +265,7 @@ pub async fn handle_grpc_request( // Start Register Transforms for stream (local_trans, stream_vrl_map) = - crate::service::ingestion::register_stream_transforms( + crate::service::ingestion::register_stream_functions( org_id, &StreamType::Metrics, local_metric_name, @@ -267,7 +274,7 @@ pub async fn handle_grpc_request( } if !local_trans.is_empty() { - rec = crate::service::ingestion::apply_stream_transform( + rec = crate::service::ingestion::apply_stream_functions( &local_trans, rec, &stream_vrl_map, diff --git a/src/service/metrics/otlp_http.rs b/src/service/metrics/otlp_http.rs index 2f44fbb22d..0536a6002e 100644 --- a/src/service/metrics/otlp_http.rs +++ b/src/service/metrics/otlp_http.rs @@ -42,7 +42,7 @@ use crate::{ alerts::Alert, http::HttpResponse as MetaHttpResponse, prom::{self, MetricType, HASH_LABEL, NAME_LABEL, VALUE_LABEL}, - stream::SchemaRecords, + stream::{SchemaRecords, StreamParams}, }, handler::http::request::CONTENT_TYPE_JSON, service::{ @@ -209,9 +209,11 @@ pub async fn metrics_json_handler( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Metrics, - metric_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: metric_name.to_owned().into(), + stream_type: StreamType::Metrics, + }], &mut stream_alerts_map, ) .await; @@ -219,7 +221,7 @@ pub async fn metrics_json_handler( // Start Register Transforms for stream let (mut local_trans, mut stream_vrl_map) = - crate::service::ingestion::register_stream_transforms( + crate::service::ingestion::register_stream_functions( org_id, &StreamType::Metrics, metric_name, @@ -349,9 +351,11 @@ pub async fn metrics_json_handler( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Metrics, - local_metric_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: local_metric_name.to_owned().into(), + stream_type: StreamType::Metrics, + }], &mut stream_alerts_map, ) .await; @@ -359,7 +363,7 @@ pub async fn metrics_json_handler( // Start Register Transforms for stream (local_trans, stream_vrl_map) = - crate::service::ingestion::register_stream_transforms( + crate::service::ingestion::register_stream_functions( org_id, &StreamType::Metrics, local_metric_name, @@ -368,7 +372,7 @@ pub async fn metrics_json_handler( } if !local_trans.is_empty() { - rec = crate::service::ingestion::apply_stream_transform( + rec = crate::service::ingestion::apply_stream_functions( &local_trans, rec, &stream_vrl_map, diff --git a/src/service/metrics/prom.rs b/src/service/metrics/prom.rs index 730d115938..64aea38543 100644 --- a/src/service/metrics/prom.rs +++ b/src/service/metrics/prom.rs @@ -40,7 +40,12 @@ use proto::prometheus_rpc; use crate::{ common::{ infra::config::{METRIC_CLUSTER_LEADER, METRIC_CLUSTER_MAP}, - meta::{alerts, functions::StreamTransform, prom::*, stream::SchemaRecords}, + meta::{ + alerts, + functions::StreamTransform, + prom::*, + stream::{SchemaRecords, StreamParams}, + }, }, service::{ db, format_stream_name, @@ -275,9 +280,11 @@ pub async fn remote_write( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Metrics, - &metric_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: metric_name.to_owned().into(), + stream_type: StreamType::Metrics, + }], &mut stream_alerts_map, ) .await; @@ -287,7 +294,7 @@ pub async fn remote_write( // Start Register Transforms for stream let (local_trans, stream_vrl_map) = - crate::service::ingestion::register_stream_transforms( + crate::service::ingestion::register_stream_functions( org_id, &StreamType::Metrics, &metric_name, @@ -299,7 +306,7 @@ pub async fn remote_write( let mut value: json::Value = json::to_value(&metric).unwrap(); // Start row based transform - value = crate::service::ingestion::apply_stream_transform( + value = crate::service::ingestion::apply_stream_functions( &local_trans, value, &stream_vrl_map, diff --git a/src/service/schema.rs b/src/service/schema.rs index 17e0b79355..9feccd85f8 100644 --- a/src/service/schema.rs +++ b/src/service/schema.rs @@ -628,6 +628,9 @@ pub async fn add_stream_schema( full_text_search_keys: vec![], bloom_filter_fields: vec![], data_retention: 0, + routing: None, + flatten_level: None, + defined_schema_fields: None, }; metadata.insert( "settings".to_string(), diff --git a/src/service/traces/mod.rs b/src/service/traces/mod.rs index 1352731569..87d60dacc7 100644 --- a/src/service/traces/mod.rs +++ b/src/service/traces/mod.rs @@ -42,7 +42,7 @@ use crate::{ common::meta::{ alerts::Alert, http::HttpResponse as MetaHttpResponse, - stream::SchemaRecords, + stream::{SchemaRecords, StreamParams}, traces::{Event, Span, SpanRefType}, }, service::{ @@ -138,16 +138,18 @@ pub async fn handle_trace_request( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Traces, - &traces_stream_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: traces_stream_name.to_owned().into(), + stream_type: StreamType::Traces, + }], &mut stream_alerts_map, ) .await; // End get stream alert // Start Register Transforms for stream - let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms( + let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_functions( org_id, &StreamType::Traces, &traces_stream_name, @@ -261,7 +263,7 @@ pub async fn handle_trace_request( })?; if !local_trans.is_empty() { - value = crate::service::ingestion::apply_stream_transform( + value = crate::service::ingestion::apply_stream_functions( &local_trans, value, &stream_vrl_map, diff --git a/src/service/traces/otlp_http.rs b/src/service/traces/otlp_http.rs index c32cb92e3b..af608c90ff 100644 --- a/src/service/traces/otlp_http.rs +++ b/src/service/traces/otlp_http.rs @@ -35,7 +35,7 @@ use crate::{ common::meta::{ alerts::Alert, http::HttpResponse as MetaHttpResponse, - stream::SchemaRecords, + stream::{SchemaRecords, StreamParams}, traces::{Event, ExportTracePartialSuccess, ExportTraceServiceResponse, Span, SpanRefType}, }, service::{ @@ -146,16 +146,18 @@ pub async fn traces_json( // Start get stream alerts crate::service::ingestion::get_stream_alerts( - org_id, - &StreamType::Traces, - &traces_stream_name, + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: traces_stream_name.to_owned().into(), + stream_type: StreamType::Traces, + }], &mut stream_alerts_map, ) .await; // End get stream alert // Start Register Transforms for stream - let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms( + let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_functions( org_id, &StreamType::Traces, &traces_stream_name, @@ -320,7 +322,7 @@ pub async fn traces_json( })?; if !local_trans.is_empty() { - value = crate::service::ingestion::apply_stream_transform( + value = crate::service::ingestion::apply_stream_functions( &local_trans, value, &stream_vrl_map, diff --git a/web/src/components/logstream/AddStream.vue b/web/src/components/logstream/AddStream.vue index fbad3ecd10..34f5018968 100644 --- a/web/src/components/logstream/AddStream.vue +++ b/web/src/components/logstream/AddStream.vue @@ -235,10 +235,12 @@ const getStreamPayload = () => { full_text_search_keys: any[]; bloom_filter_fields: any[]; data_retention?: number; + defined_schema_fields: any[]; } = { partition_keys: [], full_text_search_keys: [], bloom_filter_fields: [], + defined_schema_fields: [], }; if (showDataRetention.value && streamInputs.value.dataRetentionDays < 1) { @@ -256,6 +258,12 @@ const getStreamPayload = () => { } fields.value.forEach((field) => { + field.name = field.name + .trim() + .toLowerCase() + .replace(/ /g, "_") + .replace(/-/g, "_"); + field.index_type?.forEach((index: string) => { if (index === "fullTextSearchKey") { settings.full_text_search_keys.push(field.name); @@ -282,6 +290,7 @@ const getStreamPayload = () => { settings.bloom_filter_fields.push(field.name); } }); + settings.defined_schema_fields.push(field.name); }); return settings; diff --git a/web/src/components/logstream/StreamFieldInputs.vue b/web/src/components/logstream/StreamFieldInputs.vue index 3770606e88..251916fb53 100644 --- a/web/src/components/logstream/StreamFieldInputs.vue +++ b/web/src/components/logstream/StreamFieldInputs.vue @@ -1,6 +1,8 @@