feat(incidents): incident template, key and resolved message, stage 1d

Add alert_incidents.episode_opened_at, set on create and on reopen.
Every incident send carries the incident kind, the key
{incident_id}:{episode_opened_at} and the event, with the candidates
default_templates.incident then the alert's template. Resolving an
incident writes the status and the resolve rows in one transaction and
sends one resolved message per destination. A partial alert delivery
now stores the episode key it minted, so the retry reuses it.
This commit is contained in:
Loakesh Indiran
2026-11-02 23:36:32 +05:30
parent b27a1b39a7
commit 86e88174f8
13 changed files with 395 additions and 51 deletions
@@ -2492,6 +2492,7 @@ mod tests {
acknowledged_by: None,
acknowledged_at: None,
created_at: now,
episode_opened_at: now,
updated_at: now,
muted_by_downtime_id: None,
rca_run_id: None,
@@ -3435,6 +3435,7 @@ mod tests {
acknowledged_by: None,
acknowledged_at: None,
created_at: now,
episode_opened_at: now,
updated_at: now,
muted_by_downtime_id: None,
rca_run_id: rca_run_id.map(|run_id| sid(org_id, run_id)),
+1 -1
View File
@@ -130,7 +130,7 @@ pub type RwBTreeMap<K, V> = tokio::sync::RwLock<BTreeMap<K, V>>;
// 130: add suppress_repeats to alerts.
// 131: org domain mapping migrations
// 132: create oncall_chat_messages; add oncall_teams slack columns.
pub const DB_SCHEMA_VERSION: u64 = 134;
pub const DB_SCHEMA_VERSION: u64 = 135;
pub const DB_SCHEMA_KEY: &str = "/db_schema_version/";
// global version variables
-41
View File
@@ -3418,47 +3418,6 @@ async fn send_to_destination_type(
}
}
/// Render `ctx` with the alert's template, else the destination's, else the org default, and send
/// it.
#[cfg(feature = "enterprise")]
pub(crate) async fn send_templated_to_destination(
org_id: &str,
sns_subject: &str,
dest_name: &str,
alert_template: Option<&Template>,
ctx: &mut NotificationContext,
) -> Result<String, anyhow::Error> {
let (dest, dest_template) = destinations::get_with_template(org_id, dest_name).await?;
let Module::Alert {
destination_type, ..
} = dest.module
else {
anyhow::bail!("not an alert destination");
};
let explicit = choose_template(alert_template, dest_template.as_ref()).cloned();
let effective =
crate::alerts::notifications::org_default::resolve_effective_template(org_id, explicit)
.await;
if let DestinationType::SlackApp(target) = &destination_type {
return crate::alerts::slack_publish::send_templated_post(
&crate::alerts::slack_publish::QueueSink,
org_id,
target,
effective.template(),
ctx,
)
.await;
}
send_to_destination_type(
org_id,
sns_subject,
&destination_type,
effective.template(),
ctx,
)
.await
}
/// Send a pre-built message string to a single destination type.
///
/// Used by synthetics and downtime notifications, which build their own
+250 -9
View File
@@ -439,8 +439,8 @@ async fn send_incident_notifications_inner(
return None;
}
// Load incident to get severity, title and service_name.
let (severity, title, service_name) =
// Load incident to get severity, title, service_name and the episode it is in.
let (severity, title, service_name, episode_opened_at) =
match infra::table::alert_incidents::get(org_id, incident_id).await {
Ok(Some(model)) => {
let gv: std::collections::HashMap<String, String> =
@@ -451,10 +451,12 @@ async fn send_incident_notifications_inner(
.cloned()
.unwrap_or_default();
let title_str = model.title.unwrap_or_default();
(model.severity, title_str, svc)
(model.severity, title_str, svc, model.episode_opened_at)
}
_ => ("P3".to_string(), String::new(), String::new()),
_ => ("P3".to_string(), String::new(), String::new(), 0),
};
let incident_template = incident_override(org_id).await;
let ledger_event = incident_ledger_event(event, chrono::Utc::now().timestamp_micros());
let cfg = config::get_config();
let incident_url = format!(
@@ -476,17 +478,27 @@ async fn send_incident_notifications_inner(
&service_name,
&incident_url,
);
mark_incident_send(&mut send.ctx, incident_id, episode_opened_at, &ledger_event);
let candidates: Vec<config::meta::destinations::Template> = incident_template
.iter()
.chain(send.alert_template)
.cloned()
.collect();
for dest_name in &send.dest_names {
match crate::alerts::alert::send_templated_to_destination(
let sent = match crate::alerts::notifications::notify::notify(
org_id,
&subject,
dest_name,
send.alert_template,
&candidates,
Some(&subject),
&mut send.ctx,
)
.await
{
Ok(resp) => success_parts.push(format!("{dest_name}: {resp}")),
Ok(_) => Ok(()),
Err(e) => skip_is_not_an_error(e),
};
match sent {
Ok(()) => success_parts.push(format!("{dest_name}: sent")),
Err(e) => {
log::error!(
"[incidents] Failed to notify {dest_name} for incident {org_id}/{incident_id}: {e}"
@@ -512,6 +524,50 @@ async fn send_incident_notifications_inner(
}
}
/// The org's `default_templates.incident`, loaded once per batch as precedence level 1a.
#[cfg(feature = "enterprise")]
async fn incident_override(org_id: &str) -> Option<config::meta::destinations::Template> {
let defaults = crate::alerts::notifications::org_default::get_default_templates(org_id).await;
let name = defaults.get(config::meta::alerts::notification_kind::NotificationKind::Incident)?;
crate::db::alerts::templates::get(org_id, name).await.ok()
}
/// `opened` for a new incident, else `updated:{incident_event}:{event_time_us}`.
fn incident_ledger_event(event: &str, event_time_us: i64) -> String {
match event {
"new_incident_created" => "opened".to_string(),
"resolved" => "resolved".to_string(),
other => format!("updated:{other}:{event_time_us}"),
}
}
/// Kind, owner, episode key and ledger event of one incident send.
#[cfg(feature = "enterprise")]
fn mark_incident_send(
ctx: &mut crate::alerts::notifications::NotificationContext,
incident_id: &str,
episode_opened_at: i64,
ledger_event: &str,
) {
ctx.kind = config::meta::alerts::notification_kind::NotificationKind::Incident;
ctx.object_id = incident_id.to_string();
ctx.episode_id = Some(format!("{incident_id}:{episode_opened_at}"));
ctx.event = ledger_event.to_string();
ctx.cluster_name = config::get_cluster_name();
}
/// A skip the ledger handles, or one already delivered, is not a failed notification.
#[cfg(feature = "enterprise")]
fn skip_is_not_an_error(
e: crate::alerts::notifications::notify::NotifyError,
) -> Result<(), crate::alerts::notifications::notify::NotifyError> {
use crate::alerts::notifications::notify::{NotifyError, SkipOutcome};
match e {
NotifyError::Skipped(reason) if reason.outcome() != SkipOutcome::Unsuccessful => Ok(()),
other => Err(other),
}
}
/// Point `ctx` at the incident: its link, and the `incident_*` template variables.
#[cfg(feature = "enterprise")]
fn apply_incident_to_context(
@@ -3191,7 +3247,11 @@ pub async fn update_status_as(
) && infra::table::alert_incidents::get(org_id, incident_id)
.await?
.is_some_and(|m| m.muted_by_downtime_id.is_some());
let updated = infra::table::alert_incidents::update_status(org_id, incident_id, status).await?;
let updated = if status == "resolved" {
resolve_status(org_id, incident_id, quiet || muted_before).await?
} else {
infra::table::alert_incidents::update_status(org_id, incident_id, status).await?
};
let muted = status_change_is_muted(
quiet || muted_before,
@@ -3248,6 +3308,186 @@ pub async fn update_status_as(
model_to_incident(updated).await
}
/// Write the resolve and close the episode in one transaction; a muted resolve sends nothing.
#[cfg(feature = "enterprise")]
async fn resolve_status(
org_id: &str,
incident_id: &str,
muted: bool,
) -> Result<infra::table::entity::alert_incidents::Model, anyhow::Error> {
use config::meta::alerts::notification_kind::NotificationKind;
use infra::table::notification_deliveries::{CLOSED_BY_PRODUCER, CLOSED_BY_RESOLVE};
use crate::alerts::notifications::delivery;
let Some(before) = infra::table::alert_incidents::get(org_id, incident_id).await? else {
anyhow::bail!("Incident not found");
};
let episode_id = format!("{incident_id}:{}", before.episode_opened_at);
let (current, recorded) = if muted {
(Vec::new(), Vec::new())
} else {
incident_resolve_rows(org_id, incident_id, &before, &episode_id).await
};
let written = std::sync::Arc::new(tokio::sync::Mutex::new(None));
let slot = written.clone();
let (org, id) = (org_id.to_string(), incident_id.to_string());
let state_change: delivery::StateChange = Box::new(move |txn| {
Box::pin(async move {
let model =
infra::table::alert_incidents::update_status_with(txn, &org, &id, "resolved")
.await?;
*slot.lock().await = Some(model);
Ok(())
})
});
let results = delivery::resolve_episode(
org_id,
NotificationKind::Incident,
&episode_id,
incident_id,
current,
recorded,
if muted {
CLOSED_BY_PRODUCER
} else {
CLOSED_BY_RESOLVE
},
!muted,
state_change,
)
.await?;
for failed in results.iter().filter_map(|r| r.as_ref().err()) {
log::warn!("[incidents] resolved notification for {org_id}/{incident_id}: {failed}");
}
written
.lock()
.await
.take()
.ok_or_else(|| anyhow::anyhow!("Incident not found"))
}
#[cfg(not(feature = "enterprise"))]
async fn resolve_status(
org_id: &str,
incident_id: &str,
_muted: bool,
) -> Result<infra::table::entity::alert_incidents::Model, anyhow::Error> {
Ok(infra::table::alert_incidents::update_status(org_id, incident_id, "resolved").await?)
}
/// The resolve rows: today's alert destination merge, then destinations only the episode recorded.
#[cfg(feature = "enterprise")]
async fn incident_resolve_rows(
org_id: &str,
incident_id: &str,
incident: &infra::table::entity::alert_incidents::Model,
episode_id: &str,
) -> (
Vec<infra::table::notification_deliveries::FanoutDestination>,
Vec<infra::table::notification_deliveries::FanoutDestination>,
) {
use config::meta::alerts::notification_kind::NotificationKind;
use crate::alerts::notifications::delivery;
let first_alert = match infra::table::alert_incidents::get_incident_alerts(incident_id).await {
Ok(alerts) => match alerts.first() {
Some(ia) => crate::db::alerts::alert::get_alert_from_cache(org_id, &ia.alert_id)
.await
.map(|(_, alert)| alert),
None => None,
},
Err(_) => None,
};
let now = chrono::Utc::now().timestamp_micros();
let (mut ctx, destinations, alert_template) = match &first_alert {
Some(alert) => {
let ctx = crate::alerts::alert::build_send_context(
alert,
&[],
now,
None,
now,
Some(config::meta::alerts::level::AlertLevel::Ok),
None,
None,
None,
true,
)
.await;
let destinations =
collect_incident_destinations(org_id, incident_id, &alert.destinations).await;
let template = match alert.template.as_ref().filter(|n| !n.is_empty()) {
Some(name) => crate::db::alerts::templates::get(org_id, name).await.ok(),
None => None,
};
(ctx, destinations, template)
}
None => (
subject_notification_context(org_id, incident.title.as_deref().unwrap_or(""), now)
.await,
Vec::new(),
None,
),
};
let gv: std::collections::HashMap<String, String> =
serde_json::from_value(incident.group_values.clone()).unwrap_or_default();
let service = gv
.get("service_name")
.or_else(|| gv.get("service"))
.cloned()
.unwrap_or_default();
let incident_url = format!(
"{}/web/alerts/incidents/{incident_id}?org_identifier={org_id}",
config::get_config().common.web_url
);
let title = incident.title.clone().unwrap_or_default();
apply_incident_to_context(
&mut ctx,
incident_id,
"resolved",
&title,
&incident.severity,
&service,
&incident_url,
);
mark_incident_send(
&mut ctx,
incident_id,
incident.episode_opened_at,
"resolved",
);
ctx.alert_status = crate::alerts::notifications::STATUS_RESOLVED.to_string();
ctx.alert_description = "All linked alerts recovered".to_string();
let candidates: Vec<config::meta::destinations::Template> = incident_override(org_id)
.await
.into_iter()
.chain(alert_template)
.collect();
let subject = format!("[resolved] {title}");
let mut current = Vec::new();
for dest in &destinations {
match delivery::fanout_destination(org_id, dest, &candidates, Some(&subject), &ctx).await {
Ok(d) => current.push(d),
Err(e) => log::warn!("[incidents] no resolve row for {dest} on {incident_id}: {e}"),
}
}
let mut recorded = Vec::new();
for dest in delivery::recorded_destinations(org_id, NotificationKind::Incident, episode_id)
.await
.into_iter()
.filter(|d| !destinations.contains(d))
{
if let Ok(d) =
delivery::fanout_destination(org_id, &dest, &candidates, Some(&subject), &ctx).await
{
recorded.push(d);
}
}
(current, recorded)
}
/// Runs the one rule when the status change is the one ack's (AK-2, D20). [strong] (reads)
pub async fn authorize_status(
org_id: &str,
@@ -3663,6 +3903,7 @@ mod tests {
acknowledged_by: None,
acknowledged_at: None,
created_at: 1,
episode_opened_at: 1,
updated_at: 1,
muted_by_downtime_id: muted_by.map(str::to_string),
rca_run_id: None,
@@ -229,6 +229,9 @@ pub async fn send(
ctx: &mut NotificationContext,
) -> Result<DeliveryRow, NotifyError> {
apply_destination_metadata(&resolved.destination_type, ctx);
if let Some(sent) = send_outside_ledger(org_id, resolved, ctx).await {
return sent;
}
super::delivery::send_on_ledger(org_id, resolved, subject, ctx).await
}
+1
View File
@@ -662,6 +662,7 @@ pub(crate) mod tests {
acknowledged_by: None,
acknowledged_at: None,
created_at: now,
episode_opened_at: now,
updated_at: now,
muted_by_downtime_id: None,
rca_run_id: None,
+13
View File
@@ -2122,6 +2122,11 @@ pub(crate) fn anomaly_success_response(
.to_string()
}
/// A partial delivery opened the episode for the destinations that accepted it.
fn partial_delivery_stores_key(succeeded: &[String]) -> bool {
!succeeded.is_empty()
}
/// The model executed: anomalies found (delivered or not), none found, or muted by a downtime.
#[cfg(feature = "enterprise")]
fn detection_ran(outcome: &RunOutcome) -> bool {
@@ -4466,6 +4471,8 @@ async fn handle_alert_triggers(
// destinations already received.
trigger_data.notified_destinations =
merge_ledger(&trigger_data.notified_destinations, &succeeded);
// Store the key the delivered destinations saw, or the retry mints another.
episode_delivered |= partial_delivery_stores_key(&succeeded);
log::warn!(
"[SCHEDULER trace_id {scheduler_trace_id}] Alert {}/{}: {} of {} \
destinations failed, retrying only those; ledger now {:?}",
@@ -9821,4 +9828,10 @@ mod tests {
"{written:?}"
);
}
#[test]
fn a_partial_delivery_stores_the_minted_episode_key() {
assert!(partial_delivery_stores_key(&["slack".to_string()]));
assert!(!partial_delivery_stores_key(&[]));
}
}
+7
View File
@@ -83,6 +83,7 @@ fn new_incident(
acknowledged_by: Set(None),
acknowledged_at: Set(None),
created_at: Set(now),
episode_opened_at: Set(now),
updated_at: Set(now),
muted_by_downtime_id: Set(muted_by_downtime_id),
rca_run_id: Set(None),
@@ -347,6 +348,7 @@ pub async fn update_status_with<C: ConnectionTrait>(
if status == "acknowledged" && incident.status != "open" {
return Ok(incident);
}
let was_resolved = incident.status == "resolved";
let mut active: alert_incidents::ActiveModel = incident.into();
active.status = Set(status.to_string());
active.updated_at = Set(now);
@@ -356,6 +358,10 @@ pub async fn update_status_with<C: ConnectionTrait>(
// A manual reopen must run as a live incident, not stay muted forever.
active.muted_by_downtime_id = Set(None);
} else if status == "open" {
// A reopen starts a new notification episode, so its key differs from the resolved one.
if was_resolved {
active.episode_opened_at = Set(now);
}
active.resolved_at = Set(None);
active.acknowledged_by = Set(None);
active.acknowledged_at = Set(None);
@@ -636,6 +642,7 @@ pub async fn list_with<C: ConnectionTrait>(
.column(alert_incidents::Column::AssignedTo)
.column(alert_incidents::Column::CreatedAt)
.column(alert_incidents::Column::UpdatedAt)
.column(alert_incidents::Column::EpisodeOpenedAt)
.column(alert_incidents::Column::MutedByDowntimeId)
.column(alert_incidents::Column::RcaRunId)
.column(alert_incidents::Column::RcaState)
@@ -34,6 +34,8 @@ pub struct Model {
pub created_at: i64,
pub updated_at: i64,
/// Start of the current notification episode: created_at, then each reopen; keys the ledger.
pub episode_opened_at: i64,
/// Set while a downtime mutes this incident; nothing about it notifies.
pub muted_by_downtime_id: Option<String>,
@@ -89,6 +91,7 @@ mod tests {
acknowledged_by: None,
acknowledged_at: None,
created_at: 1000,
episode_opened_at: 1000,
updated_at: 2000,
muted_by_downtime_id: None,
rca_run_id: None,
@@ -129,6 +132,7 @@ mod tests {
acknowledged_by: Some("user@example.com".to_string()),
acknowledged_at: Some(1500),
created_at: 1000,
episode_opened_at: 1000,
updated_at: 1500,
muted_by_downtime_id: Some("2f9K".to_string()),
rca_run_id: None,
@@ -0,0 +1,110 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// 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/>.
//! `episode_opened_at`: start of an incident's notification episode, filled from created_at.
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Explicit guard because `add_column_if_not_exists` is not idempotent on SQLite.
if !manager.has_column(INCIDENTS, COLUMN).await? {
manager.alter_table(add_column_stmt()).await?;
}
manager
.get_connection()
.execute_unprepared(BACKFILL)
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
if !manager.has_column(INCIDENTS, COLUMN).await? {
return Ok(());
}
manager
.alter_table(
Table::alter()
.table(Alias::new(INCIDENTS))
.drop_column(Alias::new(COLUMN))
.to_owned(),
)
.await
}
}
const INCIDENTS: &str = "alert_incidents";
const COLUMN: &str = "episode_opened_at";
const BACKFILL: &str =
"UPDATE alert_incidents SET episode_opened_at = created_at WHERE episode_opened_at = 0";
fn add_column_stmt() -> TableAlterStatement {
Table::alter()
.table(Alias::new(INCIDENTS))
.add_column(
ColumnDef::new(Alias::new(COLUMN))
.big_integer()
.not_null()
.default(0),
)
.to_owned()
}
#[cfg(test)]
mod tests {
use sea_orm::{ConnectionTrait, Database, Statement};
use sea_orm_migration::MigrationName;
use super::*;
#[tokio::test]
async fn up_backfills_from_created_at_and_reruns() {
let db = Database::connect("sqlite::memory:").await.unwrap();
db.execute_unprepared(
"CREATE TABLE alert_incidents (id TEXT PRIMARY KEY, created_at BIGINT NOT NULL)",
)
.await
.unwrap();
db.execute_unprepared("INSERT INTO alert_incidents VALUES ('i1', 42)")
.await
.unwrap();
let manager = SchemaManager::new(&db);
Migration.up(&manager).await.unwrap();
Migration.up(&manager).await.unwrap();
let row = db
.query_one(Statement::from_string(
db.get_database_backend(),
"SELECT episode_opened_at FROM alert_incidents WHERE id = 'i1'",
))
.await
.unwrap()
.unwrap();
assert_eq!(row.try_get::<i64>("", "episode_opened_at").unwrap(), 42);
Migration.down(&manager).await.unwrap();
assert!(!manager.has_column(INCIDENTS, COLUMN).await.unwrap());
}
#[test]
fn migration_name() {
assert_eq!(
Migration.name(),
"m20261110_000003_add_incident_episode_opened_at"
);
}
}
+3
View File
@@ -250,6 +250,7 @@ mod m20261030_000002_create_org_domain_ownership_req_table;
mod m20261101_000001_create_oncall_slack;
mod m20261110_000001_create_notification_deliveries;
mod m20261110_000002_add_anomaly_alert_episode;
mod m20261110_000003_add_incident_episode_opened_at;
/// Shared body of the two `folder_id` migrations above; not a migration itself.
mod workflow_folder_id;
@@ -781,6 +782,7 @@ impl MigratorTrait for Migrator {
Box::new(m20261101_000001_create_oncall_slack::Migration),
Box::new(m20261110_000001_create_notification_deliveries::Migration),
Box::new(m20261110_000002_add_anomaly_alert_episode::Migration),
Box::new(m20261110_000003_add_incident_episode_opened_at::Migration),
]
}
}
@@ -883,6 +885,7 @@ mod tests {
(132, "m20261101_000001_create_oncall_slack"),
(133, "m20261110_000001_create_notification_deliveries"),
(134, "m20261110_000002_add_anomaly_alert_episode"),
(135, "m20261110_000003_add_incident_episode_opened_at"),
];
// Two branches that each claim the next version merge cleanly, so a reused number must fail
+1
View File
@@ -1447,6 +1447,7 @@ mod tests {
acknowledged_by: Set(None),
acknowledged_at: Set(None),
created_at: Set(T0),
episode_opened_at: Set(T0),
updated_at: Set(T0),
muted_by_downtime_id: Set(None),
rca_run_id: Set(None),