diff --git a/src/config/src/meta/triggers.rs b/src/config/src/meta/triggers.rs index a75101975e..9db5fcb90a 100644 --- a/src/config/src/meta/triggers.rs +++ b/src/config/src/meta/triggers.rs @@ -15,6 +15,8 @@ use serde::{Deserialize, Serialize}; +use crate::utils::json; + #[derive(Debug, Clone, sqlx::Type, PartialEq, Serialize, Deserialize, Default)] #[repr(i32)] pub enum TriggerStatus { @@ -88,4 +90,12 @@ impl ScheduledTriggerData { self.period_end_time = None; self.tolerance = 0; } + + pub fn to_json_string(&self) -> String { + json::to_string(self).unwrap() + } + + pub fn from_json_string(s: &str) -> Result { + json::from_str(s) + } } diff --git a/src/service/alerts/scheduler/handlers.rs b/src/service/alerts/scheduler/handlers.rs index f956ada35c..85e5020741 100644 --- a/src/service/alerts/scheduler/handlers.rs +++ b/src/service/alerts/scheduler/handlers.rs @@ -864,7 +864,11 @@ async fn handle_derived_stream_triggers( status: db::scheduler::TriggerStatus::Waiting, ..trigger.clone() }; - + let mut new_trigger_data = if trigger.data.is_empty() { + ScheduledTriggerData::default() + } else { + ScheduledTriggerData::from_json_string(&trigger.data).unwrap() + }; let Ok(pipeline) = db::pipeline::get_by_id(pipeline_id).await else { let err_msg = format!( "Pipeline associated with trigger not found: {}/{}/{}/{}. Checking after 5 mins.", @@ -899,6 +903,8 @@ async fn handle_derived_stream_triggers( }; log::error!("[SCHEDULER trace_id {scheduler_trace_id}] {}", err_msg); + new_trigger_data.reset(); + new_trigger.data = new_trigger_data.to_json_string(); db::scheduler::update_trigger(new_trigger).await?; publish_triggers_usage(trigger_data_stream).await; return Err(anyhow::anyhow!( @@ -940,6 +946,8 @@ async fn handle_derived_stream_triggers( time_in_queue_ms: Some(time_in_queue), }; log::info!("[SCHEDULER trace_id {scheduler_trace_id}] {}", msg); + new_trigger_data.reset(); + new_trigger.data = new_trigger_data.to_json_string(); db::scheduler::update_trigger(new_trigger).await?; publish_triggers_usage(trigger_data_stream).await; return Ok(()); @@ -977,6 +985,8 @@ async fn handle_derived_stream_triggers( time_in_queue_ms: Some(time_in_queue), }; log::error!("[SCHEDULER trace_id {scheduler_trace_id}] {}", err_msg); + new_trigger_data.reset(); + new_trigger.data = new_trigger_data.to_json_string(); db::scheduler::update_trigger(new_trigger).await?; publish_triggers_usage(trigger_data_stream).await; return Err(anyhow::anyhow!( @@ -984,18 +994,9 @@ async fn handle_derived_stream_triggers( err_msg )); }; - let start_time = if trigger.data.is_empty() { - None - } else { - let trigger_data: Option = json::from_str(&new_trigger.data).ok(); - if let Some(trigger_data) = trigger_data { - trigger_data - .period_end_time - .map(|period_end_time| period_end_time + 1) - } else { - None - } - }; + let start_time = new_trigger_data + .period_end_time + .map(|period_end_time| period_end_time + 1); // in case the range [start_time, end_time] is greater than querying period, it needs to // evaluate and ingest 1 period at a time. @@ -1006,7 +1007,6 @@ async fn handle_derived_stream_triggers( }) .unwrap_or_default(); let supposed_to_be_run_at = trigger.next_run_at - user_defined_delay; - let delay = current_time - trigger.next_run_at; // delay is in microseconds let mut final_end_time = if !derived_stream.trigger_condition.align_time { current_time - user_defined_delay } else { @@ -1025,27 +1025,33 @@ async fn handle_derived_stream_triggers( // is unnecessary. Hence, we need to check how big the delay is. // Note: For pipeline, period and frequency both have the same value. - // If the delay is equal to or greater than the frequency, we need to ingest data one by one - // If the delay is less than the frequency, we need to ingest data for the "next run at" - // period, For example, if the current time is 5:19pm, frequency is 5 mins, and - // delay is 4mins (supposed to be run at 5:15pm), we need to ingest data for the - // period from 5:10pm to 5:15pm only. The next run at will be 5:20pm which will - // query for the period from 5:15pm to 5:20pm. But, if the suppossed to be run at is - // 5:10pm, then we need ingest data for the period from 5:05pm to 5:15pm. - // Which is to cover the skipped period from 5:05pm to 5:15pm. - let end = if delay >= period_num_microseconds { - // final_end_time is the last multiple of given frequency after the "suppossed to be run - // at" timestamp - let frequency_count = delay / period_num_microseconds; - if derived_stream.trigger_condition.align_time { - final_end_time = supposed_to_be_run_at + (frequency_count * period_num_microseconds) - } - std::cmp::min(supposed_to_be_run_at, t0 + period_num_microseconds) - } else { - supposed_to_be_run_at - }; - - (Some(t0), end) + // For derived stream, period is in minutes, so we need to convert it to seconds for + // align_time + if derived_stream.trigger_condition.align_time { + let aligned_curr_time = TriggerCondition::align_time( + current_time, + derived_stream.tz_offset, + derived_stream.trigger_condition.period * 60, + ); + final_end_time = if aligned_curr_time > t0 { + aligned_curr_time - user_defined_delay + } else { + supposed_to_be_run_at + }; + } + // If the delay is equal to or greater than the frequency, we need to ingest data one by + // one If the delay is less than the frequency, we need to ingest data for + // the "next run at" period, For example, if the current time is 5:19pm, + // frequency is 5 mins, and delay is 4mins (supposed to be run at 5:15pm), + // we need to ingest data for the period from 5:10pm to 5:15pm only. The + // next run at will be 5:20pm which will query for the period from 5:15pm to + // 5:20pm. But, if the suppossed to be run at is 5:10pm, then we need ingest + // data for the period from 5:05pm to 5:15pm. Which is to cover the skipped + // period from 5:05pm to 5:15pm. + ( + Some(t0), + std::cmp::min(supposed_to_be_run_at, t0 + period_num_microseconds), + ) } else { (None, supposed_to_be_run_at) }; @@ -1232,6 +1238,7 @@ async fn handle_derived_stream_triggers( dest_stream.stream_type.to_string(), ) }; + let records_len = records.len(); let req = cluster_rpc::IngestionRequest { org_id: org_id.clone(), stream_name: stream_name.clone(), @@ -1243,7 +1250,8 @@ async fn handle_derived_stream_triggers( match ingestion_service::ingest(req).await { Ok(resp) if resp.status_code == 200 => { log::info!( - "[SCHEDULER trace_id {scheduler_trace_id}] DerivedStream result ingested to destination {org_id}/{stream_name}/{stream_type}", + "[SCHEDULER trace_id {scheduler_trace_id}] DerivedStream result ingested to destination {org_id}/{stream_name}/{stream_type}, records: {}", + records_len ); } error => { diff --git a/src/service/db/enrichment_table.rs b/src/service/db/enrichment_table.rs index 9910bc4a18..51df99d963 100644 --- a/src/service/db/enrichment_table.rs +++ b/src/service/db/enrichment_table.rs @@ -42,6 +42,7 @@ pub async fn get(org_id: &str, name: &str) -> Result, any sql: format!("SELECT * FROM \"{name}\""), start_time, end_time, + size: -1, // -1 means no limit, enrichment table should not be limited ..Default::default() }; @@ -56,6 +57,11 @@ pub async fn get(org_id: &str, name: &str) -> Result, any use_cache: None, local_mode: Some(true), }; + log::debug!( + "get enrichment table {} data req start time: {}", + name, + start_time + ); // do search match SearchService::search("", org_id, StreamType::EnrichmentTables, None, &req).await { Ok(res) => { @@ -231,6 +237,11 @@ pub async fn watch() -> Result<(), anyhow::Error> { let data = super::enrichment_table::get(org_id, stream_name) .await .unwrap(); + log::debug!( + "enrichment table: {} cache data length: {}", + item_key, + data.len() + ); ENRICHMENT_TABLES.insert( item_key.to_owned(), StreamTable { diff --git a/src/service/enrichment_table/mod.rs b/src/service/enrichment_table/mod.rs index 9e7c234472..edc79d016b 100644 --- a/src/service/enrichment_table/mod.rs +++ b/src/service/enrichment_table/mod.rs @@ -122,15 +122,11 @@ pub async fn save_enrichment_data( 0.0 }; let enrichment_table_max_size = cfg.limit.enrichment_table_max_size as f64; - log::info!( - "enrichment table [{stream_name}] saving stats: {:?} vs max_table_size {}", - current_size_in_bytes, - enrichment_table_max_size - ); let total_expected_size_in_bytes = current_size_in_bytes + bytes_in_payload as f64; let total_expected_size_in_mb = total_expected_size_in_bytes / SIZE_IN_MB; - log::debug!( - "enrichment table [{stream_name}] total expected storage size in mb: {} and max storage size in mb: {}", + log::info!( + "enrichment table [{stream_name}] current stats in bytes: {:?} vs total expected size in mb: {} vs max_table_size in mb: {}", + current_size_in_bytes, total_expected_size_in_mb, enrichment_table_max_size ); @@ -261,6 +257,22 @@ pub async fn save_enrichment_data( } }; + let mut enrich_meta_stats = enrichment_table::get_meta_table_stats(org_id, stream_name) + .await + .unwrap_or_default(); + + if !append_data { + enrich_meta_stats.start_time = started_at; + } + if enrich_meta_stats.start_time == 0 { + enrich_meta_stats.start_time = enrichment_table::get_start_time(org_id, stream_name).await; + } + enrich_meta_stats.end_time = now_micros(); + enrich_meta_stats.size = total_expected_size_in_bytes as i64; + // The stream_stats table takes some time to update, so we need to update the enrichment table + // size in the meta table to avoid exceeding the `ZO_ENRICHMENT_TABLE_LIMIT`. + let _ = enrichment_table::update_meta_table_stats(org_id, stream_name, enrich_meta_stats).await; + // notify update if !schema.fields().is_empty() { if let Err(e) = super::db::enrichment_table::notify_update(org_id, stream_name).await { @@ -289,22 +301,6 @@ pub async fn save_enrichment_data( ) .await; - let mut enrich_meta_stats = enrichment_table::get_meta_table_stats(org_id, stream_name) - .await - .unwrap_or_default(); - - if !append_data { - enrich_meta_stats.start_time = started_at; - } - if enrich_meta_stats.start_time == 0 { - enrich_meta_stats.start_time = enrichment_table::get_start_time(org_id, stream_name).await; - } - enrich_meta_stats.end_time = now_micros(); - enrich_meta_stats.size = total_expected_size_in_bytes as i64; - // The stream_stats table takes some time to update, so we need to update the enrichment table - // size in the meta table to avoid exceeding the `ZO_ENRICHMENT_TABLE_LIMIT`. - let _ = enrichment_table::update_meta_table_stats(org_id, stream_name, enrich_meta_stats).await; - Ok(HttpResponse::Ok().json(MetaHttpResponse::error( StatusCode::OK.into(), "Saved enrichment table".to_string(), diff --git a/src/super_cluster_queue/meta.rs b/src/super_cluster_queue/meta.rs index 0588672041..6a0e69ccd2 100644 --- a/src/super_cluster_queue/meta.rs +++ b/src/super_cluster_queue/meta.rs @@ -16,12 +16,26 @@ use infra::errors::{Error, Result}; use o2_enterprise::enterprise::super_cluster::queue::{Message, MessageType}; +use crate::service::db::enrichment_table::{ENRICHMENT_TABLE_META_STREAM_STATS_KEY, notify_update}; + pub(crate) async fn process(msg: Message) -> Result<()> { let db = infra::db::get_db().await; match msg.message_type { MessageType::Put => { db.put(&msg.key, msg.value.unwrap(), msg.need_watch, None) .await?; + // hack: notify the nodes to update the meta table stats + if msg.key.starts_with(ENRICHMENT_TABLE_META_STREAM_STATS_KEY) { + let key_parts = msg.key.split('/').collect::>(); + let org_id = key_parts[1]; + let name = key_parts[2]; + if let Err(e) = notify_update(org_id, name).await { + log::error!( + "super cluster meta queue enrichment table notify_update error: {:?}", + e + ); + } + } } MessageType::Delete(with_prefix) => { db.delete(&msg.key, with_prefix, msg.need_watch, None)