diff --git a/src/common/utils/auth.rs b/src/common/utils/auth.rs index d7bf017116..7a8e87f876 100644 --- a/src/common/utils/auth.rs +++ b/src/common/utils/auth.rs @@ -260,12 +260,12 @@ impl FromRequest for AuthExtractor { } else { path_columns[0].to_string() } - } else if url_len == 2 || (url_len > 2 && path_columns[1].starts_with("settings")) { + } else if url_len == 2 || (url_len > 2 && path_columns[1].eq("settings")) { // for settings, the post/delete require PUT permissions, GET needs LIST permissions // also the special settings exception is for 3-part urls for logo /text // which are of path /org/settings/logo , which need permission of operating // on permission in general - if path_columns[1].starts_with("settings") { + if path_columns[1].eq("settings") { if method.eq("POST") || method.eq("DELETE") { method = "PUT".to_string(); } @@ -280,7 +280,7 @@ impl FromRequest for AuthExtractor { .map_or(path_columns[1], |model| model.key), path_columns[0] ) - } else if path_columns[1].starts_with("groups") || path_columns[1].starts_with("roles") { + } else if path_columns[1].eq("groups") || path_columns[1].eq("roles") { // for groups or roles, path will be of format /org/roles/id , so we need // to check permission on role:org/id for permissions on that specific role format!( @@ -295,9 +295,9 @@ impl FromRequest for AuthExtractor { // for example, alerts are on route /org/stream/alerts // or templates are on route /org/alerts/templates and so on // users/roles is one of the special exception here - if path_columns[2].starts_with("alerts") - || path_columns[2].starts_with("templates") - || path_columns[2].starts_with("destinations") + if path_columns[2].eq("alerts") + || path_columns[2].eq("templates") + || path_columns[2].eq("destinations") || path.ends_with("users/roles") { if method.eq("GET") { @@ -338,11 +338,11 @@ impl FromRequest for AuthExtractor { ) } else if method.eq("PUT") || method.eq("DELETE") - || path_columns[1].starts_with("reports") - || path_columns[1].starts_with("savedviews") - || path_columns[1].starts_with("functions") - || path_columns[1].starts_with("service_accounts") - || path_columns[1].starts_with("cipher_keys") + || path_columns[1].eq("reports") + || path_columns[1].eq("savedviews") + || path_columns[1].eq("functions") + || path_columns[1].eq("service_accounts") + || path_columns[1].eq("cipher_keys") { // Similar to the alerts/templates etc, but for other entities such as specific // pipeline, specific stream, specific alert/destination etc. @@ -360,9 +360,9 @@ impl FromRequest for AuthExtractor { path_columns[2] ) } else if method.eq("GET") - && (path_columns[1].starts_with("dashboards") - || path_columns[1].starts_with("folders") - || path_columns[1].starts_with("actions")) + && (path_columns[1].eq("dashboards") + || path_columns[1].eq("folders") + || path_columns[1].eq("actions")) { format!( "{}:{}", diff --git a/src/config/src/meta/self_reporting/usage.rs b/src/config/src/meta/self_reporting/usage.rs index 9351b44859..27d9d94c76 100644 --- a/src/config/src/meta/self_reporting/usage.rs +++ b/src/config/src/meta/self_reporting/usage.rs @@ -73,6 +73,7 @@ pub struct TriggerData { pub evaluation_took_in_secs: Option, pub source_node: Option, pub query_took: Option, + pub trigger_trace_id: Option, } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] diff --git a/src/infra/src/scheduler/mysql.rs b/src/infra/src/scheduler/mysql.rs index 249be9a2bc..7e436f556d 100644 --- a/src/infra/src/scheduler/mysql.rs +++ b/src/infra/src/scheduler/mysql.rs @@ -16,7 +16,7 @@ use async_trait::async_trait; use bytes::Bytes; use chrono::Duration; -use config::metrics::DB_QUERY_NUMS; +use config::{metrics::DB_QUERY_NUMS, utils::hash::Sum64}; use sqlx::Row; use super::{ @@ -336,7 +336,6 @@ INSERT IGNORE INTO scheduled_jobs (org, module, module_key, is_realtime, is_sile alert_timeout: i64, report_timeout: i64, ) -> Result> { - let pool = CLIENT.clone(); let (include_max, mut max_retries) = get_scheduler_max_retries(); if include_max { max_retries += 1; @@ -354,7 +353,66 @@ INSERT IGNORE INTO scheduled_jobs (org, module, module_key, is_realtime, is_sile .unwrap() .num_microseconds() .unwrap(); - let mut tx = pool.begin().await?; + + let lock_pool = CLIENT.clone(); + let lock_key = "scheduler_pull_lock".to_string(); + let lock_id = config::utils::hash::gxhash::new().sum64(&lock_key); + let lock_sql = format!( + "SELECT GET_LOCK('{}', {})", + lock_id, + config::get_config().limit.meta_transaction_lock_timeout + ); + let unlock_sql = format!("SELECT RELEASE_LOCK('{}')", lock_id); + let mut lock_tx = lock_pool.begin().await?; + DB_QUERY_NUMS + .with_label_values(&["get_lock", "scheduled_jobs"]) + .inc(); + match sqlx::query_scalar::<_, i64>(&lock_sql) + .fetch_one(&mut *lock_tx) + .await + { + Ok(v) => { + if v != 1 { + if let Err(e) = lock_tx.rollback().await { + log::error!( + "[SCHEDULER] rollback lock for pull scheduled_jobs error: {}", + e + ); + } + return Err(Error::from(DbError::DBOperError( + "LockTimeout".to_string(), + lock_key, + ))); + } + } + Err(e) => { + if let Err(e) = lock_tx.rollback().await { + log::error!( + "[SCHEDULER] rollback lock for pull scheduled_jobs error: {}", + e + ); + } + return Err(e.into()); + } + }; + + let pool = CLIENT.clone(); + let mut tx = match pool.begin().await { + Ok(tx) => tx, + Err(e) => { + if let Err(e) = sqlx::query(&unlock_sql).execute(&mut *lock_tx).await { + log::error!("[SCHEDULER] unlock pull scheduled_jobs error: {}", e); + } + if let Err(e) = lock_tx.commit().await { + log::error!( + "[SCHEDULER] commit for unlock pull scheduled_jobs error: {}", + e + ); + } + return Err(e.into()); + } + }; + DB_QUERY_NUMS .with_label_values(&["select", "scheduled_jobs"]) .inc(); @@ -379,7 +437,19 @@ FOR UPDATE; Ok(ids) => ids, Err(e) => { if let Err(e) = tx.rollback().await { - log::error!("[MYSQL] rollback select jobs for update error: {}", e); + log::error!("[SCHEDULER] rollback select jobs for update error: {}", e); + } + DB_QUERY_NUMS + .with_label_values(&["release_lock", "scheduled_jobs"]) + .inc(); + if let Err(e) = sqlx::query(&unlock_sql).execute(&mut *lock_tx).await { + log::error!("[SCHEDULER] unlock pull scheduled_jobs error: {}", e); + } + if let Err(e) = lock_tx.commit().await { + log::error!( + "[SCHEDULER] commit for unlock pull scheduled_jobs error: {}", + e + ); } return Err(e.into()); } @@ -391,8 +461,19 @@ FOR UPDATE; ); if job_ids.is_empty() { if let Err(e) = tx.rollback().await { - log::error!("[MYSQL] rollback scheduler pull error: {}", e); - return Err(e.into()); + log::error!("[SCHEDULER] rollback scheduler pull error: {}", e); + } + DB_QUERY_NUMS + .with_label_values(&["release_lock", "scheduled_jobs"]) + .inc(); + if let Err(e) = sqlx::query(&unlock_sql).execute(&mut *lock_tx).await { + log::error!("[SCHEDULER] unlock pull scheduled_jobs error: {}", e); + } + if let Err(e) = lock_tx.commit().await { + log::error!( + "[SCHEDULER] commit for unlock pull scheduled_jobs error: {}", + e + ); } return Ok(vec![]); } @@ -423,15 +504,52 @@ WHERE id IN ({});", if let Err(e) = tx.rollback().await { log::error!("[MYSQL] rollback update scheduled jobs status error: {}", e); } + DB_QUERY_NUMS + .with_label_values(&["release_lock", "scheduled_jobs"]) + .inc(); + if let Err(e) = sqlx::query(&unlock_sql).execute(&mut *lock_tx).await { + log::error!("[SCHEDULER] unlock pull scheduled_jobs error: {}", e); + } + if let Err(e) = lock_tx.commit().await { + log::error!( + "[SCHEDULER] commit for unlock pull scheduled_jobs error: {}", + e + ); + } return Err(e.into()); } log::debug!("Update scheduled jobs for selected pull job ids"); if let Err(e) = tx.commit().await { - log::error!("[MYSQL] commit scheduler pull update error: {}", e); + log::error!("[SCHEDULER] commit scheduler pull update error: {}", e); + DB_QUERY_NUMS + .with_label_values(&["release_lock", "scheduled_jobs"]) + .inc(); + if let Err(e) = sqlx::query(&unlock_sql).execute(&mut *lock_tx).await { + log::error!("[SCHEDULER] unlock pull scheduled_jobs error: {}", e); + } + if let Err(e) = lock_tx.commit().await { + log::error!( + "[SCHEDULER] commit for unlock pull scheduled_jobs error: {}", + e + ); + } return Err(e.into()); } + DB_QUERY_NUMS + .with_label_values(&["release_lock", "scheduled_jobs"]) + .inc(); + if let Err(e) = sqlx::query(&unlock_sql).execute(&mut *lock_tx).await { + log::error!("[SCHEDULER] unlock pull scheduled_jobs error: {}", e); + } + if let Err(e) = lock_tx.commit().await { + log::error!( + "[SCHEDULER] commit for unlock pull scheduled_jobs error: {}", + e + ); + } + let query = format!( "SELECT * FROM scheduled_jobs WHERE id IN ({});", job_ids.join(",") diff --git a/src/infra/src/scheduler/postgres.rs b/src/infra/src/scheduler/postgres.rs index e2a109bf03..fdd2f02d7f 100644 --- a/src/infra/src/scheduler/postgres.rs +++ b/src/infra/src/scheduler/postgres.rs @@ -16,7 +16,7 @@ use async_trait::async_trait; use bytes::Bytes; use chrono::Duration; -use config::metrics::DB_QUERY_NUMS; +use config::{metrics::DB_QUERY_NUMS, utils::hash::Sum64}; use sqlx::Row; use super::{TRIGGERS_KEY, Trigger, TriggerModule, TriggerStatus, get_scheduler_max_retries}; @@ -350,9 +350,6 @@ INSERT INTO scheduled_jobs (org, module, module_key, is_realtime, is_silenced, s .unwrap() .num_microseconds() .unwrap(); - DB_QUERY_NUMS - .with_label_values(&["update", "scheduled_jobs"]) - .inc(); let query = r#"UPDATE scheduled_jobs SET status = $1, start_time = $2, end_time = CASE @@ -370,6 +367,29 @@ WHERE id IN ( RETURNING *;"#; let mut tx = pool.begin().await?; + + // Lock the table for the duration of the transaction + let lock_key = "scheduler_pull_lock"; + let lock_id = config::utils::hash::gxhash::new().sum64(lock_key); + let lock_id = if lock_id > i64::MAX as u64 { + (lock_id >> 1) as i64 + } else { + lock_id as i64 + }; + let lock_sql = format!("SELECT pg_advisory_xact_lock({lock_id})"); + DB_QUERY_NUMS + .with_label_values(&["get_lock", "scheduled_jobs"]) + .inc(); + if let Err(e) = sqlx::query(&lock_sql).execute(&mut *tx).await { + if let Err(e) = tx.rollback().await { + log::error!("[SCHEDULER] rollback pull scheduled_jobs error: {}", e); + } + return Err(e.into()); + } + + DB_QUERY_NUMS + .with_label_values(&["update", "scheduled_jobs"]) + .inc(); let jobs: Vec = match sqlx::query_as::<_, Trigger>(query) .bind(TriggerStatus::Processing) .bind(now) diff --git a/src/service/alerts/scheduler/handlers.rs b/src/service/alerts/scheduler/handlers.rs index 49cc780a46..51e7acf2d3 100644 --- a/src/service/alerts/scheduler/handlers.rs +++ b/src/service/alerts/scheduler/handlers.rs @@ -224,6 +224,7 @@ async fn handle_alert_triggers( evaluation_took_in_secs: None, source_node: Some(source_node.clone()), query_took: None, + trigger_trace_id: Some(trace_id.to_string()), }) .await; log::info!( @@ -279,6 +280,7 @@ async fn handle_alert_triggers( evaluation_took_in_secs: None, source_node: Some(source_node), query_took: None, + trigger_trace_id: Some(trace_id.to_string()), }; let evaluation_took = Instant::now(); @@ -682,6 +684,7 @@ async fn handle_report_triggers( evaluation_took_in_secs: None, source_node: Some(LOCAL_NODE.name.clone()), query_took: None, + trigger_trace_id: Some(trace_id.to_string()), }; if trigger.retries >= max_retries { @@ -884,6 +887,7 @@ async fn handle_derived_stream_triggers( evaluation_took_in_secs: None, source_node: Some(LOCAL_NODE.name.clone()), query_took: None, + trigger_trace_id: Some(trace_id.to_string()), }; // evaluate trigger and configure trigger next run time diff --git a/src/service/alerts/scheduler/mod.rs b/src/service/alerts/scheduler/mod.rs index ff142d974c..7d391754cc 100644 --- a/src/service/alerts/scheduler/mod.rs +++ b/src/service/alerts/scheduler/mod.rs @@ -13,92 +13,11 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::collections::HashMap; - -use config::{get_config, ider}; -use futures::future::try_join_all; - -use crate::service::db; +use config::get_config; pub mod handlers; pub mod worker; -pub async fn run_old() -> Result<(), anyhow::Error> { - let trace_id = ider::generate(); - log::debug!("[SCHEDULER trace_id {trace_id}] Pulling jobs from scheduler"); - let cfg = get_config(); - - // Scheduler pulls only those triggers that match the conditions- - // - trigger.next_run_at <= now - // - !(trigger.is_realtime && !trigger.is_silenced) - // - trigger.status == "Waiting" - let triggers = db::scheduler::pull( - cfg.limit.alert_schedule_concurrency, - cfg.limit.alert_schedule_timeout, - cfg.limit.report_schedule_timeout, - ) - .await?; - - log::info!( - "[SCHEDULER trace_id {trace_id}] Pulled {} jobs from scheduler", - triggers.len() - ); - - if !triggers.is_empty() { - let mut grouped_triggers: std::collections::HashMap< - db::scheduler::TriggerModule, - Vec<&db::scheduler::Trigger>, - > = HashMap::new(); - - // Group triggers by module - for trigger in &triggers { - grouped_triggers - .entry(trigger.module.clone()) - .or_default() - .push(trigger); - } - - // Print counts for each module - for (module, triggers) in grouped_triggers { - log::info!( - "[SCHEDULER trace_id {trace_id}] Pulled {:?}: {} jobs", - module, - triggers.len() - ); - } - } - - let mut tasks = Vec::new(); - for (i, trigger) in triggers.into_iter().enumerate() { - let trace_id = format!("{}-{}", trace_id, i); - let task = tokio::task::spawn(async move { - let key = format!("{}-{}/{}", trigger.module, trigger.org, trigger.module_key); - log::debug!( - "[SCHEDULER trace_id {trace_id}] start processing trigger: {}", - key - ); - // if let Err(e) = handle_triggers(&trace_id, trigger).await { - // log::error!( - // "[SCHEDULER trace_id {trace_id}] Error handling trigger: {}", - // e - // ); - // } - log::debug!( - "[SCHEDULER trace_id {trace_id}] finished processing trigger: {}", - key - ); - }); - tasks.push(task); - } - if let Err(e) = try_join_all(tasks).await { - log::error!( - "[SCHEDULER trace_id {trace_id}] Error handling triggers: {}", - e - ); - } - Ok(()) -} - // The main function that creates and runs the scheduler pub async fn run() -> Result<(), anyhow::Error> { // Get configuration @@ -106,7 +25,7 @@ pub async fn run() -> Result<(), anyhow::Error> { // Create scheduler config let scheduler_config = worker::SchedulerConfig { - alert_schedule_concurrency: cfg.limit.alert_schedule_concurrency as usize, + alert_schedule_concurrency: cfg.limit.alert_schedule_concurrency, alert_schedule_timeout: cfg.limit.alert_schedule_timeout, report_schedule_timeout: cfg.limit.report_schedule_timeout, poll_interval_secs: cfg.limit.alert_schedule_interval as u64, // Can be configurable diff --git a/src/service/alerts/scheduler/worker.rs b/src/service/alerts/scheduler/worker.rs index c9afc867b5..f76847054f 100644 --- a/src/service/alerts/scheduler/worker.rs +++ b/src/service/alerts/scheduler/worker.rs @@ -16,7 +16,7 @@ use std::{collections::HashMap, sync::Arc}; use anyhow::Result; use tokio::{ - sync::{Mutex, Semaphore, mpsc}, + sync::{Mutex, mpsc}, time, }; @@ -32,7 +32,7 @@ pub struct ScheduledJob { /// Configuration for the scheduler #[derive(Clone)] pub struct SchedulerConfig { - pub alert_schedule_concurrency: usize, + pub alert_schedule_concurrency: i64, pub alert_schedule_timeout: i64, pub report_schedule_timeout: i64, pub poll_interval_secs: u64, @@ -43,14 +43,12 @@ pub struct SchedulerConfig { pub struct SchedulerWorker { pub id: usize, pub rx: Arc>>, - pub available_workers: Arc, } /// Scheduler job puller that fetches jobs from the database #[derive(Clone)] pub struct SchedulerJobPuller { pub tx: mpsc::Sender, - pub available_workers: Arc, pub config: SchedulerConfig, } @@ -61,52 +59,42 @@ pub struct Scheduler { pub job_puller: SchedulerJobPuller, pub tx: mpsc::Sender, pub rx: Arc>>, - pub available_workers: Arc, } impl SchedulerWorker { - pub fn new( - id: usize, - rx: Arc>>, - available_workers: Arc, - ) -> Self { - Self { - id, - rx, - available_workers, - } + pub fn new(id: usize, rx: Arc>>) -> Self { + Self { id, rx } } pub async fn run(&self) -> Result<()> { loop { - // Try to get next trigger - let trigger = { + log::debug!("[SCHEDULER][Worker-{}] Waiting for next job", self.id); + // Try to get next trigger with permit + let job = { let mut rx = self.rx.lock().await; rx.recv().await }; - match trigger { - Some(trigger) => { - // Acquire permit before processing - let _permit = self.available_workers.acquire().await?; + match job { + Some(job) => { + log::debug!( + "[SCHEDULER][Worker-{}] trace_id: {} Processing trigger: {}", + self.id, + job.trace_id, + job.trigger.module_key + ); // Process the trigger - if let Err(e) = self - .handle_trigger(&trigger.trace_id, trigger.trigger) - .await - { + if let Err(e) = self.handle_trigger(&job.trace_id, job.trigger).await { log::error!( "[SCHEDULER][Worker-{}] trace_id: {} Error handling trigger: {}", self.id, - trigger.trace_id, + job.trace_id, e ); } - - // Permit is automatically released when _permit is dropped } None => { - // Channel closed, no more work log::info!( "[SCHEDULER][Worker-{}] Channel closed, no more work", self.id @@ -115,26 +103,27 @@ impl SchedulerWorker { } } } + log::info!("[SCHEDULER][Worker-{}] Exiting", self.id); Ok(()) } /// Handles a given trigger pub async fn handle_trigger(&self, trace_id: &str, trigger: Trigger) -> Result<()> { - handle_triggers(trace_id, trigger).await + let trace_id = trace_id.to_owned(); + // If there is any panic, it should not crash the scheduler worker + let handler_job = tokio::spawn(async move { handle_triggers(&trace_id, trigger).await }); + + if let Err(e) = handler_job.await { + return Err(anyhow::anyhow!("Error in handler: {}", e)); + } + + Ok(()) } } impl SchedulerJobPuller { - pub fn new( - tx: mpsc::Sender, - available_workers: Arc, - config: SchedulerConfig, - ) -> Self { - Self { - tx, - available_workers, - config, - } + pub fn new(tx: mpsc::Sender, config: SchedulerConfig) -> Self { + Self { tx, config } } /// Runs the job puller. Pulls jobs from the database and sends them to the workers. @@ -148,72 +137,74 @@ impl SchedulerJobPuller { let interval = self.config.poll_interval_secs; let mut interval = time::interval(time::Duration::from_secs(interval)); interval.tick().await; // The first tick + loop { // Check how many workers are available - let available_count = self.available_workers.available_permits(); let trace_id = config::ider::uuid(); + + log::info!("[SCHEDULER][JobPuller-{}] Pulling jobs", trace_id); + + // Pull only as many jobs as we have workers + let triggers = match self.pull().await { + Ok(triggers) => triggers, + Err(e) => { + log::error!( + "[SCHEDULER][JobPuller-{}] Error pulling triggers: {}", + trace_id, + e + ); + continue; + } + }; + log::info!( - "[SCHEDULER][JobPuller-{}] Available workers: {}", + "[SCHEDULER][JobPuller-{}] Pulled {} jobs from scheduler", trace_id, - available_count + triggers.len() ); - if available_count > 0 { - // Pull only as many jobs as we have available workers - let triggers = self.pull(available_count).await?; + // Print counts for each module + if !triggers.is_empty() { + let mut grouped_triggers: std::collections::HashMap> = + HashMap::new(); - log::info!( - "[SCHEDULER][JobPuller-{}] Pulled {} jobs from scheduler", - trace_id, - triggers.len() - ); - - // Print counts for each module - if !triggers.is_empty() { - let mut grouped_triggers: std::collections::HashMap< - TriggerModule, - Vec<&Trigger>, - > = HashMap::new(); - - // Group triggers by module - for trigger in &triggers { - grouped_triggers - .entry(trigger.module.clone()) - .or_default() - .push(trigger); - } - - // Print counts for each module - for (module, triggers) in grouped_triggers { - log::info!( - "[SCHEDULER JobPuller trace_id {trace_id}] Pulled {:?}: {} jobs", - module, - triggers.len() - ); - } + // Group triggers by module + for trigger in &triggers { + grouped_triggers + .entry(trigger.module.clone()) + .or_default() + .push(trigger); } - // Send all triggers to be processed - for trigger in triggers { - let scheduled_job = ScheduledJob { - trace_id: trace_id.clone(), - trigger, - }; - if self.tx.send(scheduled_job).await.is_err() { - // Channel closed, exit loop - log::error!( - "[SCHEDULER][JobPuller-{}] Channel closed, exiting job puller", - trace_id - ); - return Ok(()); - } + // Print counts for each module + for (module, triggers) in grouped_triggers { + log::info!( + "[SCHEDULER] [JobPuller-{}] Pulled {:?}: {} jobs", + trace_id, + module, + triggers.len() + ); + } + } + + // Send all triggers to be processed + for trigger in triggers { + let scheduled_job = ScheduledJob { + trace_id: trace_id.clone(), + trigger, + }; + + if self.tx.send(scheduled_job).await.is_err() { + log::error!( + "[SCHEDULER][JobPuller-{}] Channel closed, exiting job puller", + trace_id + ); + return Ok(()); } } - // Wait for the next tick interval.tick().await; - // Check if channel is closed if self.tx.is_closed() { break; } @@ -222,9 +213,9 @@ impl SchedulerJobPuller { Ok(()) } - pub async fn pull(&self, limit: usize) -> Result> { + pub async fn pull(&self) -> Result> { match scheduler_pull( - limit as i64, + self.config.alert_schedule_concurrency, self.config.alert_schedule_timeout, self.config.report_schedule_timeout, ) @@ -241,7 +232,7 @@ impl SchedulerJobPuller { impl Scheduler { pub fn new(config: SchedulerConfig) -> Self { - let max_workers = config.alert_schedule_concurrency; + let max_workers = config.alert_schedule_concurrency as usize; // Create a channel for work distribution with capacity for max workers let (tx, rx) = mpsc::channel(max_workers); @@ -249,17 +240,13 @@ impl Scheduler { // Share receiver between workers let rx = Arc::new(Mutex::new(rx)); - // Track available workers with a semaphore - let available_workers = Arc::new(Semaphore::new(max_workers)); - // Create workers let workers = (0..max_workers) - .map(|id| SchedulerWorker::new(id, rx.clone(), available_workers.clone())) + .map(|id| SchedulerWorker::new(id, rx.clone())) .collect(); // Create job puller - let job_puller = - SchedulerJobPuller::new(tx.clone(), available_workers.clone(), config.clone()); + let job_puller = SchedulerJobPuller::new(tx.clone(), config.clone()); Self { config, @@ -267,7 +254,6 @@ impl Scheduler { job_puller, tx, rx, - available_workers, } } diff --git a/src/service/ingestion/mod.rs b/src/service/ingestion/mod.rs index 82b2bbad2e..9da08b00aa 100644 --- a/src/service/ingestion/mod.rs +++ b/src/service/ingestion/mod.rs @@ -253,6 +253,7 @@ pub async fn evaluate_trigger(triggers: TriggerAlertData) { evaluation_took_in_secs: None, source_node: Some(LOCAL_NODE.name.clone()), query_took: None, + trigger_trace_id: None, }; match alert.send_notification(val, now, None, now).await { Err(e) => {