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

feat: routing & user defined schema restrictions (#3254)

Browse Source 
Co-authored-by: omkark06 <omkar@zinclabs.io>
This commit is contained in:
Ashish Kolhe
2024-04-19 06:00:56 +05:30
committed by GitHub
co-authored by omkark06
parent 015f3db59d
commit ebdc19d02e
24 changed files with 760 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config/src/config.rs
+2
View File
@@ -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)]
src/config/src/meta/stream.rs
+181
View File
@@ -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<String>,
#[serde(default)]
pub data_retention: i64,
#[serde(skip_serializing_if = "Option::None")]
pub routing: Option<HashMap<String, Vec<RoutingCondition>>>,
#[serde(skip_serializing_if = "Option::None")]
pub flatten_level: Option<i64>,
#[serde(skip_serializing_if = "Option::None")]
pub defined_schema_fields: Option<Vec<String>>,
}
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<Vec<String>> = 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<String, Vec<RoutingCondition>> = 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<PartitionTimeLevel>,
}
pub struct Routing {
pub destination: String,
pub routing: Vec<RoutingCondition>,
}
// 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<String, Value>) -> 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::*;
src/handler/http/request/stream/mod.rs
+2 -2
View File
@@ -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),
)),
);
}
src/infra/src/schema/mod.rs
+17
View File
@@ -498,6 +498,23 @@ pub async fn delete_fields(
}
})
.collect::<Vec<_>>();
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::<Vec<_>>();
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()),
src/service/ingestion/mod.rs
+83 -35
View File
@@ -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<String, Vec<StreamTransform>>,
pub async fn get_stream_functions<'a>(
streams: &[StreamParams],
stream_functions_map: &mut HashMap<String, Vec<StreamTransform>>,
stream_vrl_map: &mut HashMap<String, VRLResultResolver>,
) {
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<StreamTransform> = 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<StreamTransform> = 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<String, Vec<Alert>>,
) {
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::<Vec<_>>();
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::<Vec<_>>();
stream_alerts_map.insert(key, alerts);
stream_alerts_map.insert(key, alerts);
}
}
pub async fn evaluate_trigger(trigger: Option<TriggerAlertData>) {
@@ -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<String, VRLResultResolver>,
@@ -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<String, Vec<Routing>>,
) {
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<Routing> = 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<String, Vec<String>>,
) {
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};
src/service/logs/bulk.rs
+105 -43
View File
@@ -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<String, SchemaCache> = HashMap::new();
let mut stream_data_map = HashMap::new();
let mut stream_transform_map: HashMap<String, Vec<StreamTransform>> = HashMap::new();
let mut stream_functions_map: HashMap<String, Vec<StreamTransform>> = HashMap::new();
let mut stream_partition_keys_map: HashMap<String, (StreamSchemaChk, PartitioningDetails)> =
HashMap::new();
let mut stream_alerts_map: HashMap<String, Vec<Alert>> = HashMap::new();
@@ -110,6 +110,10 @@ pub async fn ingest(
let mut blocked_stream_warnings: HashMap<String, bool> = HashMap::new();
let mut stream_routing_map: HashMap<String, Vec<Routing>> = HashMap::new();
let mut user_defined_schema_map: HashMap<String, Vec<String>> = 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,
src/service/logs/ingest.rs
+7 -5
View File
@@ -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<String, Vec<Alert>> = 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,
src/service/logs/mod.rs
+20
View File
@@ -447,6 +447,26 @@ struct StreamMeta<'a> {
stream_alerts_map: &'a HashMap<String, Vec<Alert>>,
}
pub fn refactor_map(original_map: &mut Map<String, Value>, 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::*;
src/service/logs/multi.rs
+7 -5
View File
@@ -87,7 +87,7 @@ async fn ingest_inner(
let mut trigger: Option<TriggerAlertData> = 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,
src/service/logs/otlp_grpc.rs
+12 -8
View File
@@ -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,
src/service/logs/otlp_http.rs
+7 -5
View File
@@ -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,
src/service/logs/syslog.rs
+7 -5
View File
@@ -108,16 +108,18 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result<HttpResponse> {
// 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<HttpResponse> {
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,
src/service/metadata/trace_list_index.rs
+3
View File
@@ -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)
src/service/metrics/json.rs
+2 -2
View File
@@ -357,13 +357,13 @@ fn apply_func(
metric_name: &str,
value: json::Value,
) -> Result<json::Value> {
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,
src/service/metrics/otlp_grpc.rs
+17 -10
View File
@@ -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,
src/service/metrics/otlp_http.rs
+14 -10
View File
@@ -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,
src/service/metrics/prom.rs
+13 -6
View File
@@ -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,
src/service/schema.rs
+3
View File
@@ -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(),
src/service/traces/mod.rs
+8 -6
View File
@@ -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,
src/service/traces/otlp_http.rs
+8 -6
View File
@@ -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,
web/src/components/logstream/AddStream.vue
+9
View File
@@ -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;
web/src/components/logstream/StreamFieldInputs.vue
+18 -2
View File
@@ -1,6 +1,8 @@
<template>
<div data-test="add-stream-fields-section">
<div data-test="alert-conditions-text" class="text-bold">Fields *</div>
<div v-if="showHeader" data-test="alert-conditions-text" class="text-bold">
Fields *
</div>
<template v-if="!fields.length">
<q-btn
data-test="add-stream-add-field-btn"
@@ -42,6 +44,7 @@
/>
</div>
<!-- <div
v-if="visibleInputs.type"
data-test="alert-conditions-operator-select"
class="q-ml-none o2-input"
>
@@ -62,6 +65,7 @@
/>
</div> -->
<div
v-if="visibleInputs.index_type"
data-test="add-stream-field-type-select-input"
class="q-ml-none flex items-end o2-input"
>
@@ -75,7 +79,7 @@
multiple
:max-values="2"
map-options
:option-disable="(_option) => disableOptions(field, _option)"
:option-disable="(_option: any) => disableOptions(field, _option)"
emit-value
clearable
stack-label
@@ -138,6 +142,18 @@ defineProps({
default: () => [],
required: true,
},
showHeader: {
type: Boolean,
default: true,
},
visibleInputs: {
type: Object,
default: () => ({
name: true,
type: true,
index_type: true,
}),
},
});
const emits = defineEmits(["add", "remove", "input:update"]);
web/src/components/logstream/schema.vue
+213 -95
View File
@@ -45,8 +45,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
>
No data available.
</div>
<div v-else
class="indexDetailsContainer" style="height: 100vh">
<div v-else class="indexDetailsContainer" style="height: 100vh">
<div class="title" data-test="schema-stream-title-text">
{{ indexData.name }}
</div>
@@ -116,6 +115,30 @@ class="indexDetailsContainer" style="height: 100vh">
</div>
</div>
</template>
<template v-if="indexData.defined_schema_fields.length">
<q-separator
id="schema-add-fields-section"
class="q-mt-lg q-mb-lg"
/>
<div
data-test="schema-add-fields-title"
class="q-pr-sm text-bold q-mb-sm"
>
Add Fields
</div>
<StreamFieldsInputs
:fields="newSchemaFields"
:showHeader="false"
:visibleInputs="{
name: true,
type: false,
index_type: false,
}"
@add="addSchemaField"
@remove="removeSchemaField"
/>
</template>
<q-separator class="q-mt-lg q-mb-lg" />
<div class="title" data-test="schema-log-stream-mapping-title-text">
@@ -185,8 +208,7 @@ class="indexDetailsContainer" style="height: 100vh">
<q-td>{{ props.row.type }}</q-td>
</template>
<template v-slot:body-cell-index_type="props">
<q-td
data-test="schema-stream-index-select"
<q-td data-test="schema-stream-index-select"
><q-select
v-if="
props.row.name !== store.state.zoConfig.timestamp_column
@@ -211,7 +233,7 @@ class="indexDetailsContainer" style="height: 100vh">
outlined
filled
dense
style="width: 300px;"
style="width: 300px"
@update:model-value="markFormDirty(props.row.name, 'fts')"
/>
</q-td>
@@ -231,6 +253,18 @@ class="indexDetailsContainer" style="height: 100vh">
no-caps
@click="confirmQueryModeChangeDialog = true"
/>
<q-btn
v-if="indexData.defined_schema_fields.length"
data-test="schema-add-field-button"
class="q-my-sm no-border text-bold q-ml-md"
:label="t('logStream.addField')"
padding="sm md"
color="secondary"
no-caps
@click="scrollToAddFields"
/>
<q-btn
v-bind:disable="!formDirtyFlag"
data-test="schema-update-settings-button"
@@ -283,6 +317,8 @@ import { formatSizeFromMB, getImageURL } from "@/utils/zincutils";
import config from "@/aws-exports";
import ConfirmDialog from "@/components/ConfirmDialog.vue";
import useStreams from "@/composables/useStreams";
import { useRouter } from "vue-router";
import StreamFieldsInputs from "@/components/logstream/StreamFieldInputs.vue";
const defaultValue: any = () => {
return {
@@ -304,6 +340,7 @@ export default defineComponent({
},
components: {
ConfirmDialog,
StreamFieldsInputs,
},
setup({ modelValue }) {
const { t } = useI18n();
@@ -319,6 +356,8 @@ export default defineComponent({
const loadingState = ref(true);
const rowsPerPage = ref(250);
const filterField = ref("");
const router = useRouter();
const newSchemaFields = ref([]);
const streamIndexType = [
{ label: "Inverted Index", value: "fullTextSearchKey" },
@@ -394,6 +433,126 @@ export default defineComponent({
});
};
const getFieldIndices = (property, settings) => {
const fieldIndices = [];
if (
(settings.full_text_search_keys.length > 0 &&
settings.full_text_search_keys.includes(property.name)) ||
(settings.full_text_search_keys.length == 0 &&
store.state.zoConfig.default_fts_keys.includes(property.name))
) {
fieldIndices.push("fullTextSearchKey");
}
if (
settings.bloom_filter_fields.length > 0 &&
settings.bloom_filter_fields.includes(property.name)
) {
fieldIndices.push("bloomFilterKey");
}
property["delete"] = false;
if (
settings.partition_keys &&
Object.values(settings.partition_keys).some(
(v) => !v.disabled && v.field === property.name
)
) {
const [level, partition] = Object.entries(settings.partition_keys).find(
([, partition]) => partition["field"] === property.name
);
property.level = level;
if (partition.types === "value") fieldIndices.push("keyPartition");
if (partition.types?.hash)
fieldIndices.push(`hashPartition_${partition.types.hash}`);
}
property.index_type = [...fieldIndices];
return fieldIndices;
};
const setSchema = (streamResponse) => {
const schemaMapping = new Set([]);
if (!streamResponse.schema?.length) {
streamResponse.schema = [];
streamResponse.settings.defined_schema_fields.forEach((field) => {
streamResponse.schema.push({
name: field,
delete: false,
index_type: [],
});
});
}
if (
streamResponse.settings.full_text_search_keys.length == 0 &&
(showFullTextSearchColumn.value || showPartitionColumn.value)
) {
indexData.value.defaultFts = true;
} else {
indexData.value.defaultFts = false;
}
indexData.value.schema = streamResponse.schema || [];
indexData.value.stats = JSON.parse(JSON.stringify(streamResponse.stats));
indexData.value.stats.doc_time_max = date.formatDate(
parseInt(streamResponse.stats.doc_time_max) / 1000,
"YYYY-MM-DDTHH:mm:ss:SSZ"
);
indexData.value.stats.doc_time_min = date.formatDate(
parseInt(streamResponse.stats.doc_time_min) / 1000,
"YYYY-MM-DDTHH:mm:ss:SSZ"
);
indexData.value.defined_schema_fields =
streamResponse.settings.defined_schema_fields || [];
if (showDataRetention.value)
dataRetentionDays.value =
streamResponse.settings.data_retention ||
store.state.zoConfig.data_retention_days;
if (!streamResponse.schema) {
loadingState.value = false;
dismiss();
return;
}
let fieldIndices = [];
for (var property of streamResponse.schema) {
schemaMapping.add(property.name);
fieldIndices = getFieldIndices(property, streamResponse.settings);
property.index_type = [...fieldIndices];
fieldIndices.length = 0;
}
indexData.value.defined_schema_fields.forEach((field) => {
if (!schemaMapping.has(field)) {
const property = {
name: field,
delete: false,
index_type: [],
};
fieldIndices = getFieldIndices(property, streamResponse.settings);
property.index_type = [...fieldIndices];
fieldIndices.length = 0;
indexData.value.schema.push(property);
}
});
};
const getSchema = async () => {
const dismiss = q.notify({
spinner: true,
@@ -402,88 +561,7 @@ export default defineComponent({
await getStream(indexData.value.name, indexData.value.stream_type, true)
.then((streamResponse) => {
if (
streamResponse.settings.full_text_search_keys.length == 0 &&
(showFullTextSearchColumn.value || showPartitionColumn.value)
) {
indexData.value.defaultFts = true;
} else {
indexData.value.defaultFts = false;
}
indexData.value.schema = streamResponse.schema || [];
indexData.value.stats = JSON.parse(
JSON.stringify(streamResponse.stats)
);
indexData.value.stats.doc_time_max = date.formatDate(
parseInt(streamResponse.stats.doc_time_max) / 1000,
"YYYY-MM-DDTHH:mm:ss:SSZ"
);
indexData.value.stats.doc_time_min = date.formatDate(
parseInt(streamResponse.stats.doc_time_min) / 1000,
"YYYY-MM-DDTHH:mm:ss:SSZ"
);
if (showDataRetention.value)
dataRetentionDays.value =
streamResponse.settings.data_retention ||
store.state.zoConfig.data_retention_days;
if (!streamResponse.schema) {
loadingState.value = false;
dismiss();
return;
}
const fieldIndices = [];
for (var property of streamResponse.schema) {
if (
(streamResponse.settings.full_text_search_keys.length > 0 &&
streamResponse.settings.full_text_search_keys.includes(
property.name
)) ||
(streamResponse.settings.full_text_search_keys.length == 0 &&
store.state.zoConfig.default_fts_keys.includes(property.name))
) {
fieldIndices.push("fullTextSearchKey");
}
if (
streamResponse.settings.bloom_filter_fields.length > 0 &&
streamResponse.settings.bloom_filter_fields.includes(
property.name
)
) {
fieldIndices.push("bloomFilterKey");
}
property["delete"] = false;
if (
streamResponse.settings.partition_keys &&
Object.values(streamResponse.settings.partition_keys).some(
(v) => !v.disabled && v.field === property.name
)
) {
const [level, partition] = Object.entries(
streamResponse.settings.partition_keys
).find(([, partition]) => partition["field"] === property.name);
property.level = level;
if (partition.types === "value")
fieldIndices.push("keyPartition");
if (partition.types?.hash)
fieldIndices.push(`hashPartition_${partition.types.hash}`);
}
property.index_type = [...fieldIndices];
fieldIndices.length = 0;
}
setSchema(streamResponse);
loadingState.value = false;
dismiss();
})
@@ -498,6 +576,7 @@ export default defineComponent({
partition_keys: [],
full_text_search_keys: [],
bloom_filter_fields: [],
defined_schema_fields: [...indexData.value.defined_schema_fields],
};
if (showDataRetention.value && dataRetentionDays.value < 1) {
@@ -514,6 +593,17 @@ export default defineComponent({
settings["data_retention"] = Number(dataRetentionDays.value);
}
settings.defined_schema_fields.push(
...newSchemaFields.value.map((field) => {
field.name = field.name
.trim()
.toLowerCase()
.replace(/ /g, "_")
.replace(/-/g, "_");
return field.name;
})
);
let added_part_keys = [];
for (var property of indexData.value.schema) {
property.index_type?.forEach((index: string) => {
@@ -568,6 +658,9 @@ export default defineComponent({
}
loadingState.value = true;
newSchemaFields.value = [];
await streamService
.updateSettings(
store.state.selectedOrganization.identifier,
@@ -576,19 +669,20 @@ export default defineComponent({
settings
)
.then(async (res) => {
loadingState.value = false;
q.notify({
color: "positive",
message: "Stream settings updated successfully.",
timeout: 2000,
});
await getStream(
indexData.value.name,
indexData.value.stream_type,
true,
true
);
).then((streamResponse) => {
setSchema(streamResponse);
loadingState.value = false;
q.notify({
color: "positive",
message: "Stream settings updated successfully.",
timeout: 2000,
});
});
segment.track("Button Click", {
button: "Update Settings",
@@ -693,6 +787,26 @@ export default defineComponent({
},
];
const addSchemaField = () => {
newSchemaFields.value.push({
name: "",
type: "",
index_type: [],
});
formDirtyFlag.value = true;
};
const removeSchemaField = (field: any, index: number) => {
newSchemaFields.value.splice(index, 1);
};
const scrollToAddFields = () => {
const el = document.getElementById("schema-add-fields-section");
if (el) {
el.scrollIntoView({ behavior: "smooth" });
}
};
return {
t,
q,
@@ -722,6 +836,10 @@ export default defineComponent({
rowsPerPage,
filterField,
columns,
addSchemaField,
removeSchemaField,
newSchemaFields,
scrollToAddFields,
};
},
created() {
web/src/locales/languages/en.json
+2 -1
View File
@@ -371,7 +371,8 @@
"dataRetention": "Data Retention (in days)",
"indexing": "Indexing",
"createStream": "Create Stream",
"indexType": "Index Type"
"indexType": "Index Type",
"addField": "Add Field"
},
"function": {
"header": "Functions",