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

fix: remove semaphore in alert manager (#6295)

Browse Source 
- Fixes #6289 
- Also, Removes use of semaphore inside the scheduler. Currently it only
uses the mpsc channel and waits if the mpsc channel is full. Apart from
this, it uses db locks in postgres and mysql when pulling
scheduled_jobs.
This commit is contained in:
Subhra264
2025-03-18 23:04:33 +05:30
committed by GitHub
parent 88eb0ded69
commit bcd5bf30ce
8 changed files with 258 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/utils/auth.rs
+14 -14
View File
@@ -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!(
"{}:{}",
src/config/src/meta/self_reporting/usage.rs
+1
View File
@@ -73,6 +73,7 @@ pub struct TriggerData {
pub evaluation_took_in_secs: Option<f64>,
pub source_node: Option<String>,
pub query_took: Option<i64>,
pub trigger_trace_id: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
src/infra/src/scheduler/mysql.rs
+125 -7
View File
@@ -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<Vec<Trigger>> {
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(",")
src/infra/src/scheduler/postgres.rs
+24 -4
View File
@@ -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<Trigger> = match sqlx::query_as::<_, Trigger>(query)
.bind(TriggerStatus::Processing)
.bind(now)
src/service/alerts/scheduler/handlers.rs
+4
View File
@@ -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
src/service/alerts/scheduler/mod.rs
+2 -83
View File
@@ -13,92 +13,11 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
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
src/service/alerts/scheduler/worker.rs
+87 -101
View File
@@ -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<Mutex<mpsc::Receiver<ScheduledJob>>>,
pub available_workers: Arc<Semaphore>,
}
/// Scheduler job puller that fetches jobs from the database
#[derive(Clone)]
pub struct SchedulerJobPuller {
pub tx: mpsc::Sender<ScheduledJob>,
pub available_workers: Arc<Semaphore>,
pub config: SchedulerConfig,
}
@@ -61,52 +59,42 @@ pub struct Scheduler {
pub job_puller: SchedulerJobPuller,
pub tx: mpsc::Sender<ScheduledJob>,
pub rx: Arc<Mutex<mpsc::Receiver<ScheduledJob>>>,
pub available_workers: Arc<Semaphore>,
}
impl SchedulerWorker {
pub fn new(
id: usize,
rx: Arc<Mutex<mpsc::Receiver<ScheduledJob>>>,
available_workers: Arc<Semaphore>,
) -> Self {
Self {
id,
rx,
available_workers,
}
pub fn new(id: usize, rx: Arc<Mutex<mpsc::Receiver<ScheduledJob>>>) -> 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<ScheduledJob>,
available_workers: Arc<Semaphore>,
config: SchedulerConfig,
) -> Self {
Self {
tx,
available_workers,
config,
}
pub fn new(tx: mpsc::Sender<ScheduledJob>, 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<TriggerModule, Vec<&Trigger>> =
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<Vec<Trigger>> {
pub async fn pull(&self) -> Result<Vec<Trigger>> {
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,
}
}
src/service/ingestion/mod.rs
+1
View File
@@ -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) => {