Function evaluate_alert

nodedb/src/event/alert/executor.rs:94–142

Evaluate a single alert rule: execute aggregate query, check condition per group, feed through hysteresis, dispatch notifications.

(
    state: &Arc<SharedState>,
    alert: &AlertDef,
    hysteresis: &super::hysteresis::HysteresisManager,
    now_ms: u64,
)

/// Evaluate a single alert rule: execute aggregate query, check condition per group,
/// feed through hysteresis, dispatch notifications.
async fn evaluate_alert(
    state: &Arc<SharedState>,
    alert: &AlertDef,
    hysteresis: &super::hysteresis::HysteresisManager,
    now_ms: u64,
) -> crate::Result<()> {
    let tenant_id = TenantId::new(alert.tenant_id);
    let window_start = now_ms.saturating_sub(alert.window_ms);

    debug!(alert = alert.name, %window_start, %now_ms, "evaluating alert");

    // Dispatch aggregate scan directly to Data Plane via SPSC bridge.
    let results =
        execute_aggregate_scan(state, tenant_id, alert, window_start as i64, now_ms as i64).await?;

    // Process each group's result through hysteresis.
    for (group_key, agg_value) in &results {
        let condition_met = alert
            .condition
            .op
            .evaluate(*agg_value, alert.condition.threshold);

        let transition = hysteresis.evaluate(
            alert.tenant_id,
            &alert.name,
            group_key,
            condition_met,
            *agg_value,
            alert.fire_after,
            alert.recover_after,
            now_ms,
        );

        if transition != HysteresisTransition::NoChange {
            info!(
                alert = alert.name,
                group = group_key,
                value = agg_value,
                threshold = alert.condition.threshold,
                transition = ?transition,
                "alert state transition"
            );

            dispatch_notifications(state, alert, group_key, *agg_value, transition, now_ms).await;
        }
    }

    Ok(())
}

/// Dispatch an aggregate scan to the Data Plane via the SPSC bridge.
///