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

fix: Update enrichment table cache with correct start time after meta table update and pipeline delayed timerange fix (#6844)

Browse Source 
- when creating an enrichment table, the `notify_update` is called
before the enrichment table stats is stored in the meta table.
- When pipeline reaches max retries and gets triggered again in the next
run, it uses one single time window for the whole delayed timerange.
This commit is contained in:
Subhra264
2025-05-20 17:15:20 +05:30
committed by GitHub
parent c84792d3eb
commit e268fe6f64
5 changed files with 98 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config/src/meta/triggers.rs
+10
View File
@@ -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<Self, serde_json::Error> {
json::from_str(s)
}
}
src/service/alerts/scheduler/handlers.rs
+44 -36
View File
@@ -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<ScheduledTriggerData> = 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 => {
src/service/db/enrichment_table.rs
+11
View File
@@ -42,6 +42,7 @@ pub async fn get(org_id: &str, name: &str) -> Result<Vec<vrl::value::Value>, 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<Vec<vrl::value::Value>, 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 {
src/service/enrichment_table/mod.rs
+19 -23
View File
@@ -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(),
src/super_cluster_queue/meta.rs
+14
View File
@@ -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::<Vec<&str>>();
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)