feat(oncall): stage 2c, on-call sends through the shared send path

The on-call engine reaches notify() through a registered sender slot. Pages,
escalations, manual escalates and handoffs are Firing with the response id
as the platform key; coverage gap, verdict and test pages are
Informational, so platforms skip them. Email and SNS destinations now get
pages (D9), and a generic webhook keeps today's page JSON (D1). Nothing is
sent on acknowledge or resolve (strict D8, D4). oncall_policies gains a
template column (m20261210_000001) with validation and replication, and a
startup line names the policies whose email or SNS destinations now get
pages.
This commit is contained in:
Loakesh Indiran
2026-11-03 14:24:05 +05:30
parent b24f9e4425
commit bcc37562a0
19 changed files with 827 additions and 11 deletions
@@ -183,6 +183,10 @@ pub struct SetPolicyRequest {
/// Absent leaves it unchanged, so editing rungs cannot silently un-configure L0.
#[serde(default)]
pub l0: Option<config::meta::oncall::L0Policy>,
/// Template override for on-call messages; absent leaves it, null or "" clears it.
#[serde(default, deserialize_with = "double_option")]
#[schema(value_type = Option<String>)]
pub template: Option<Option<String>>,
}
#[derive(Debug, Default, Deserialize)]
@@ -1670,6 +1674,28 @@ pub async fn set_policy(
other => internal_error("set_policy node versions", &other),
};
}
if let Some(template) = body.template.clone() {
let template = template
.map(|t| t.trim().to_string())
.filter(|t| !t.is_empty());
if let Some(name) = template.as_deref()
&& let Err(e) = openobserve_core::alerts::templates::check_binding(
&org_id,
name,
config::meta::alerts::notification_kind::NotificationKind::OnCall,
)
.await
{
return MetaHttpResponse::bad_request(e.as_str());
}
if let Err(e) = o2_enterprise::enterprise::oncall::service::set_policy_template(
&org_id, &team_id, template,
)
.await
{
return to_response(e);
}
}
match o2_enterprise::enterprise::oncall::service::set_policy(
&org_id,
&team_id,
@@ -6838,6 +6864,7 @@ mod tests {
}],
destinations: None,
l0: None,
template: None,
};
let response = set_policy(
Path(("acme".to_string(), "t_dm8".to_string())),
+2 -1
View File
@@ -134,7 +134,8 @@ pub type RwBTreeMap<K, V> = tokio::sync::RwLock<BTreeMap<K, V>>;
// 134: rename user templates under the reserved o2_default_ prefix.
// 135: add template to anomaly_detection_config.
// 136: add template to synthetics.
pub const DB_SCHEMA_VERSION: u64 = 136;
// 137: add template to oncall_policies.
pub const DB_SCHEMA_VERSION: u64 = 137;
pub const DB_SCHEMA_KEY: &str = "/db_schema_version/";
// global version variables
+5
View File
@@ -404,6 +404,9 @@ pub struct EscalationPolicy {
/// to behave like a team that never opened the screen.
#[serde(default = "super::agent::L0Policy::defaults")]
pub l0: super::agent::L0Policy,
/// Template override for this team's on-call messages; `None` uses each destination's.
#[serde(default)]
pub template: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -548,6 +551,7 @@ impl EscalationPolicy {
channels: vec![],
},
],
template: None,
}
}
@@ -606,6 +610,7 @@ impl EscalationPolicy {
channels: vec![],
},
],
template: None,
}
}
@@ -27,7 +27,7 @@ use config::{
DEFAULT_ORG,
meta::{
alerts::{
content_spec::{ChartSpec, ContentField, ContentSpec, RowsSpec},
content_spec::{ChartSpec, ContentField, ContentLink, ContentSpec, RowsSpec},
notification_kind::NotificationKind,
},
destinations::{Template, TemplateKind, TemplateType},
@@ -82,6 +82,7 @@ pub fn compiled_content_for(kind: NotificationKind) -> ContentSpec {
NotificationKind::Incident => compiled_incident_content(),
NotificationKind::Anomaly => compiled_anomaly_content(),
NotificationKind::Synthetics => compiled_synthetics_content(),
NotificationKind::OnCall => compiled_on_call_content(),
_ => compiled_default_content(),
}
}
@@ -89,9 +90,10 @@ pub fn compiled_content_for(kind: NotificationKind) -> ContentSpec {
/// The reserved name that `compiled_content_for(kind)` is seeded under.
pub fn compiled_name_for(kind: NotificationKind) -> &'static str {
match kind {
NotificationKind::Incident | NotificationKind::Anomaly | NotificationKind::Synthetics => {
kind.compiled_default_name()
}
NotificationKind::Incident
| NotificationKind::Anomaly
| NotificationKind::Synthetics
| NotificationKind::OnCall => kind.compiled_default_name(),
_ => DEFAULT_CONTENT_TEMPLATE_NAME,
}
}
@@ -141,6 +143,38 @@ pub fn compiled_synthetics_content() -> ContentSpec {
}
}
/// Today's page header as the title; the Acknowledge and Runbook links drop out when unset.
pub fn compiled_on_call_content() -> ContentSpec {
ContentSpec {
title: "{oncall_priority} · {alert_name}".into(),
title_overrides: Default::default(),
body: "**{oncall_team}** — {oncall_reason}\n\n{alert_description}".into(),
fields: vec![
field("Stage", "{oncall_stage}"),
field("Recipient", "{oncall_recipient}"),
],
rows: RowsSpec {
enabled: false,
max: 5,
columns: None,
format: None,
},
links: vec![
ContentLink {
label: "Acknowledge".into(),
url: "{oncall_ack_url}".into(),
show_when: None,
},
ContentLink {
label: "Runbook".into(),
url: "{runbook_url}".into(),
show_when: None,
},
],
chart: ChartSpec { enabled: false },
}
}
pub fn compiled_incident_content() -> ContentSpec {
ContentSpec {
title: "Incident {incident_severity}: {incident_title}".into(),
@@ -192,6 +226,7 @@ pub(crate) async fn ensure_default_content_template(
NotificationKind::Incident,
NotificationKind::Anomaly,
NotificationKind::Synthetics,
NotificationKind::OnCall,
] {
ensure_compiled_row(
compiled_name_for(kind),
+69 -1
View File
@@ -69,6 +69,17 @@ pub const SYNTHETICS_VARIABLES: [&str; 16] = [
"check_title",
];
pub const ONCALL_VARIABLES: [&str; 8] = [
"oncall_team",
"oncall_reason",
"oncall_stage",
"oncall_priority",
"oncall_recipient",
"oncall_response_id",
"oncall_ack_url",
"runbook_url",
];
pub const DOWNTIME_VARIABLES: [&str; 8] = [
"downtime_name",
"downtime_event",
@@ -117,6 +128,7 @@ pub fn variables_for(kind: NotificationKind) -> Vec<String> {
NotificationKind::Incident => &INCIDENT_VARIABLES,
NotificationKind::Synthetics => &SYNTHETICS_VARIABLES,
NotificationKind::Downtime => &DOWNTIME_VARIABLES,
NotificationKind::OnCall => &ONCALL_VARIABLES,
_ => &[],
};
COMMON_VARIABLES
@@ -135,12 +147,13 @@ pub struct VariableGroup {
}
/// The kinds whose producer sends through templates in this release; the UI hides the others.
pub const SHIPPED_KINDS: [NotificationKind; 5] = [
pub const SHIPPED_KINDS: [NotificationKind; 6] = [
NotificationKind::Alert,
NotificationKind::Incident,
NotificationKind::Anomaly,
NotificationKind::Synthetics,
NotificationKind::Downtime,
NotificationKind::OnCall,
];
/// `common`, then one group per shipped kind with its own variables.
@@ -161,6 +174,7 @@ pub fn variable_groups() -> Vec<VariableGroup> {
NotificationKind::Incident => &INCIDENT_VARIABLES,
NotificationKind::Synthetics => &SYNTHETICS_VARIABLES,
NotificationKind::Downtime => &DOWNTIME_VARIABLES,
NotificationKind::OnCall => &ONCALL_VARIABLES,
_ => &[],
};
groups.push(group(kind.as_str(), vars));
@@ -174,6 +188,7 @@ pub fn example_context(kind: NotificationKind, resolved: bool) -> NotificationCo
NotificationKind::Anomaly => anomaly_example(resolved),
NotificationKind::Incident => incident_example(resolved),
NotificationKind::Synthetics => synthetics_example(resolved),
NotificationKind::OnCall => on_call_example(resolved),
_ => {
let level = if resolved {
AlertLevel::Ok
@@ -328,6 +343,44 @@ fn synthetics_example(resolved: bool) -> NotificationContext {
ctx
}
fn on_call_example(resolved: bool) -> NotificationContext {
let url = "https://o2.acme.com/web/oncall/responses/resp_31?org_identifier=acme";
let level = if resolved {
AlertLevel::Ok
} else {
AlertLevel::Critical
};
let mut ctx = base(NotificationKind::OnCall, resolved, level);
ctx.alert_name = "High error rate on web-logs".into();
ctx.alert_url = url.into();
ctx.episode_id = Some("resp_31".into());
ctx.event = event_of(resolved);
ctx.alert_trigger_time_str = "2026-11-02 10:05:00".into();
ctx.alert_description = "error rate 12.4% over the 5% threshold on checkout".into();
let (stage, reason, ack) = if resolved {
("resolved", "the incident was resolved", "")
} else {
(
"1",
"paged by policy checkout-escalation",
"https://o2.acme.com/api/acme/oncall/ack/resp_31",
)
};
ctx.context_attributes = [
("oncall_team", "checkout"),
("oncall_reason", reason),
("oncall_stage", stage),
("oncall_priority", "P2"),
("oncall_recipient", "maria@acme.com"),
("oncall_response_id", "resp_31"),
("oncall_ack_url", ack),
("runbook_url", "https://wiki.acme.com/runbooks/checkout"),
]
.map(|(k, v)| (k.to_string(), v.to_string()))
.to_vec();
ctx
}
fn incident_example(resolved: bool) -> NotificationContext {
let url = "https://o2.acme.com/web/alerts/incidents/inc_7f3a?org_identifier=acme";
let level = if resolved {
@@ -408,6 +461,7 @@ mod tests {
const EXAMPLE_ANOMALY: &str = include_str!("render/testdata/examples/anomaly.json");
const EXAMPLE_INCIDENT: &str = include_str!("render/testdata/examples/incident.json");
const EXAMPLE_SYNTHETICS: &str = include_str!("render/testdata/examples/synthetics.json");
const EXAMPLE_ON_CALL: &str = include_str!("render/testdata/examples/on_call.json");
fn rendered(
kind: NotificationKind,
@@ -466,6 +520,19 @@ mod tests {
);
}
#[test]
fn golden_example_on_call_keeps_todays_header_and_drops_an_unset_ack_link() {
let want: serde_json::Value = serde_json::from_str(EXAMPLE_ON_CALL).unwrap();
assert_eq!(example(NotificationKind::OnCall), want);
let firing = want["slack_firing"].to_string();
assert!(
firing.contains("P2 · High error rate on web-logs"),
"{firing}"
);
assert!(firing.contains("Acknowledge"));
assert!(!want["slack_resolved"].to_string().contains("Acknowledge"));
}
#[test]
fn anomaly_recovery_hides_empty_fields() {
let ctx = example_context(NotificationKind::Anomaly, true);
@@ -500,6 +567,7 @@ mod tests {
(NotificationKind::Anomaly, "anomaly.json"),
(NotificationKind::Incident, "incident.json"),
(NotificationKind::Synthetics, "synthetics.json"),
(NotificationKind::OnCall, "on_call.json"),
] {
let body = serde_json::to_string_pretty(&example(kind)).unwrap() + "\n";
std::fs::write(format!("{dir}/{file}"), body).unwrap();
+2
View File
@@ -28,6 +28,8 @@ pub mod format;
pub mod jira;
pub(crate) mod modifiers;
pub mod notify;
#[cfg(feature = "enterprise")]
pub mod oncall;
pub mod org_default;
pub mod platform;
pub mod preview;
+269
View File
@@ -0,0 +1,269 @@
// 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/>.
//! The on-call sender the enterprise engine reaches through its registered slot.
use config::meta::alerts::{
level::AlertLevel,
notification_kind::{NotificationEvent, NotificationKind},
};
use o2_enterprise::enterprise::oncall::sender::{OnCallNotification, OnCallSendError, OnCallStage};
use super::{
LegacyPayload, NotificationContext, STATUS_FIRING,
notify::{NotifyError, SkipReason, notify},
};
/// Sends one on-call message to one destination through the shared path.
pub async fn notify_destination(n: OnCallNotification) -> Result<(), OnCallSendError> {
let mut ctx = on_call_context(&n);
let explicit = super::org_default::load_override(&n.org_id, n.template.as_deref()).await;
let result = notify(&n.org_id, &n.destination, &explicit, None, &mut ctx).await;
send_result(result.map(|_| ()))
}
/// Upgrade notice: one line per policy whose email or SNS destinations now get pages.
pub async fn log_mail_destination_policies() {
use config::meta::destinations::{DestinationType, Module};
let policies = match infra::table::oncall_policies::list_all().await {
Ok(p) => p,
Err(e) => {
log::warn!("[ONCALL] could not list policies for the upgrade notice: {e}");
return;
}
};
for policy in policies {
let mut mail = Vec::new();
for name in &policy.destinations {
if let Ok(Some(dest)) = infra::table::destinations::get(&policy.org_id, name).await
&& let Module::Alert {
destination_type: DestinationType::Email(_) | DestinationType::Sns(_),
..
} = dest.module
{
mail.push(name.as_str());
}
}
if !mail.is_empty() {
log::info!(
"[ONCALL] policy of team {} in org {} names email or SNS destinations ({}); they receive pages from this version",
policy.team_id,
policy.org_id,
mail.join(", ")
);
}
}
}
/// The shared path's context for one on-call message (Backend section 6 field mapping).
pub fn on_call_context(n: &OnCallNotification) -> NotificationContext {
let level = match n.priority.as_str() {
"P1" => Some(AlertLevel::Critical),
"P2" => Some(AlertLevel::Warning),
_ => None,
};
let title = if n.stage == OnCallStage::Test && !n.title.starts_with("[TEST] ") {
format!("[TEST] {}", n.title)
} else {
n.title.clone()
};
let now = config::utils::time::now_micros();
let legacy = n.legacy_payload.clone();
let attribute = |k: &str, v: String| (k.to_string(), v);
NotificationContext {
org_name: n.org_id.clone(),
alert_name: title,
alert_level: level.map(|l| l.to_string()).unwrap_or_default(),
alert_status: STATUS_FIRING.to_string(),
episode_id: (!n.key.is_empty()).then(|| n.key.clone()),
alert_url: n.url.clone(),
alert_trigger_time: now,
alert_trigger_time_str: crate::anomaly_detection::fmt_local_us(now),
alert_description: n.text.clone(),
context_attributes: vec![
attribute("oncall_team", n.team.clone()),
attribute("oncall_reason", n.reason.clone()),
attribute("oncall_stage", n.stage.as_str().to_string()),
attribute("oncall_priority", n.priority.clone()),
attribute("oncall_recipient", n.recipient.clone().unwrap_or_default()),
attribute(
"oncall_response_id",
n.response_id.clone().unwrap_or_default(),
),
attribute("oncall_ack_url", n.ack_url.clone().unwrap_or_default()),
attribute("runbook_url", n.runbook_url.clone().unwrap_or_default()),
],
level,
kind: NotificationKind::OnCall,
event: n.event,
cluster_name: config::get_cluster_name(),
legacy_payload: Some(LegacyPayload::new(move || legacy.clone())),
render_hints: super::context::RenderHints {
fallback_subject: Some(n.title.clone()),
escape_email_attributes: true,
},
..Default::default()
}
}
/// A Retryable error is retried by the engine; a platform skip is neither delivered nor failed.
fn send_result(result: Result<(), NotifyError>) -> Result<(), OnCallSendError> {
match result {
Ok(()) | Err(NotifyError::Skipped(SkipReason::PlatformNotForKind)) => Ok(()),
Err(NotifyError::Skipped(reason)) => Err(OnCallSendError::Skipped(reason.as_str().into())),
Err(e) if e.is_retryable() => Err(OnCallSendError::Retryable(e.to_string())),
Err(e) => Err(OnCallSendError::Permanent(e.to_string())),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::alerts::notifications::{
notify::{Route, route},
org_default::{EffectiveTemplate, TemplateSource, compiled_fallback_for},
};
fn page(stage: OnCallStage) -> OnCallNotification {
OnCallNotification {
org_id: "acme".into(),
destination: "slack-ops".into(),
stage,
event: stage.event(),
key: if stage == OnCallStage::Test {
String::new()
} else {
"resp_31".into()
},
title: "High error rate".into(),
text: "error rate 12%".into(),
reason: "paged by policy".into(),
priority: "P1".into(),
team: "checkout".into(),
recipient: Some("maria@acme.com".into()),
url: "https://o2/r".into(),
ack_url: Some("https://o2/ack".into()),
response_id: Some("resp_31".into()),
legacy_payload: serde_json::json!({"text": "legacy"}),
..Default::default()
}
}
fn resolved_for(destination_type: &str) -> super::super::notify::Resolved {
let dest_type = config::meta::destinations::DestinationType::Http(
config::meta::destinations::Endpoint {
destination_type: Some(destination_type.to_string()).filter(|t| !t.is_empty()),
url: "https://hooks.example/x".into(),
..Default::default()
},
);
super::super::notify::Resolved {
destination: Default::default(),
channel: super::super::derive_channel_format(&dest_type),
destination_type: dest_type,
effective: EffectiveTemplate {
template: compiled_fallback_for("acme", NotificationKind::OnCall),
source: TemplateSource::Compiled,
skipped: vec![],
},
}
}
#[test]
fn a_page_maps_to_the_on_call_context() {
let ctx = on_call_context(&page(OnCallStage::Paged));
assert_eq!(ctx.kind, NotificationKind::OnCall);
assert_eq!(ctx.event, NotificationEvent::Firing);
assert_eq!(ctx.episode_id.as_deref(), Some("resp_31"));
assert_eq!(ctx.level, Some(AlertLevel::Critical));
assert_eq!(ctx.alert_status, STATUS_FIRING);
assert!(
ctx.context_attributes
.iter()
.any(|(k, v)| k == "oncall_ack_url" && v == "https://o2/ack")
);
assert_eq!(
ctx.legacy_payload.unwrap().build(),
serde_json::json!({"text": "legacy"})
);
}
#[test]
fn a_test_page_keeps_its_marker_and_a_p3_has_no_level() {
let mut n = page(OnCallStage::Test);
n.priority = "P3".into();
let ctx = on_call_context(&n);
assert_eq!(ctx.alert_name, "[TEST] High error rate");
assert_eq!(ctx.level, None);
assert_eq!(ctx.episode_id, None);
}
#[test]
fn the_event_matrix_on_platform_and_other_destinations() {
let paged = on_call_context(&page(OnCallStage::Paged));
assert_eq!(
route(&resolved_for("pagerduty"), &paged).unwrap(),
Route::Render
);
assert_eq!(
route(&resolved_for("slack"), &paged).unwrap(),
Route::Render
);
assert!(matches!(
route(&resolved_for(""), &paged).unwrap(),
Route::Legacy(_)
));
for stage in [
OnCallStage::CoverageGap,
OnCallStage::Verdict,
OnCallStage::Test,
] {
let ctx = on_call_context(&page(stage));
assert!(
matches!(
route(&resolved_for("opsgenie"), &ctx),
Err(NotifyError::Skipped(SkipReason::PlatformNotForKind))
),
"{stage:?}"
);
assert_eq!(route(&resolved_for("slack"), &ctx).unwrap(), Route::Render);
}
}
#[test]
fn send_results_map_to_what_the_engine_retries() {
assert_eq!(send_result(Ok(())), Ok(()));
assert_eq!(
send_result(Err(NotifyError::Skipped(SkipReason::PlatformNotForKind))),
Ok(())
);
assert!(matches!(
send_result(Err(NotifyError::Skipped(SkipReason::NoOrgMemberRecipient))),
Err(OnCallSendError::Skipped(_))
));
assert!(matches!(
send_result(Err(NotifyError::Rejected {
status: 503,
body: String::new()
})),
Err(OnCallSendError::Retryable(_))
));
assert!(matches!(
send_result(Err(NotifyError::DestinationNotFound("x".into()))),
Err(OnCallSendError::Permanent(_))
));
}
}
+2 -1
View File
@@ -440,7 +440,8 @@ mod tests {
"incident",
"anomaly",
"synthetics",
"downtime"
"downtime",
"on_call"
]
);
let incident = &resp.variable_groups[2];
@@ -0,0 +1,208 @@
{
"slack_firing": {
"attachments": [
{
"color": "#E01E5A",
"blocks": [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*<https://o2.acme.com/web/oncall/responses/resp_31?org_identifier=acme|P2 · High error rate on web-logs>*"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*checkout* — paged by policy checkout-escalation\n\nerror rate 12.4% over the 5% threshold on checkout"
}
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": "*Stage*\n1"
},
{
"type": "mrkdwn",
"text": "*Recipient*\nmaria@acme.com"
}
]
}
]
},
{
"color": "#E01E5A",
"blocks": [
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {
"type": "plain_text",
"text": "Acknowledge"
},
"url": "https://o2.acme.com/api/acme/oncall/ack/resp_31"
},
{
"type": "button",
"text": {
"type": "plain_text",
"text": "Runbook"
},
"url": "https://wiki.acme.com/runbooks/checkout"
},
{
"type": "button",
"text": {
"type": "plain_text",
"text": "View in OpenObserve"
},
"url": "https://o2.acme.com/web/oncall/responses/resp_31?org_identifier=acme",
"style": "primary"
}
]
}
]
}
]
},
"slack_resolved": {
"attachments": [
{
"color": "#2EB67D",
"blocks": [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*<https://o2.acme.com/web/oncall/responses/resp_31?org_identifier=acme|P2 · High error rate on web-logs>*"
}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*checkout* — the incident was resolved\n\nerror rate 12.4% over the 5% threshold on checkout"
}
},
{
"type": "section",
"fields": [
{
"type": "mrkdwn",
"text": "*Stage*\nresolved"
},
{
"type": "mrkdwn",
"text": "*Recipient*\nmaria@acme.com"
}
]
}
]
},
{
"color": "#2EB67D",
"blocks": [
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {
"type": "plain_text",
"text": "Runbook"
},
"url": "https://wiki.acme.com/runbooks/checkout"
},
{
"type": "button",
"text": {
"type": "plain_text",
"text": "View in OpenObserve"
},
"url": "https://o2.acme.com/web/oncall/responses/resp_31?org_identifier=acme",
"style": "primary"
}
]
}
]
}
]
},
"webhook_firing": {
"version": 1,
"title": "P2 · High error rate on web-logs",
"body_markdown": "**checkout** — paged by policy checkout\\-escalation\n\nerror rate 12\\.4\\% over the 5\\% threshold on checkout",
"fields": {
"Stage": "1",
"Recipient": "maria@acme.com"
},
"rows": [],
"links": [
{
"label": "Acknowledge",
"url": "https://o2.acme.com/api/acme/oncall/ack/resp_31"
},
{
"label": "Runbook",
"url": "https://wiki.acme.com/runbooks/checkout"
},
{
"label": "",
"url": "https://o2.acme.com/web/oncall/responses/resp_31?org_identifier=acme"
}
],
"severity": "critical",
"chart_url": null,
"context": {
"org_name": "acme",
"stream_type": "logs",
"stream_name": "web-logs",
"alert_name": "High error rate on web-logs",
"alert_type": null,
"alert_period": null,
"alert_operator": null,
"alert_threshold": null,
"alert_count": null,
"alert_agg_value": null,
"alert_level": "critical",
"alert_status": "firing",
"episode_id": "resp_31",
"recovery_reason": null,
"alert_priority": null,
"alert_tags": null,
"alert_threshold_crit": null,
"alert_threshold_warn": null,
"alert_warning_threshold": null,
"alert_start_time": "N/A",
"alert_end_time": "N/A",
"alert_url": "https://o2.acme.com/web/oncall/responses/resp_31?org_identifier=acme",
"alert_trigger_time_str": "2026-11-02 10:05:00",
"alert_time": "2026-11-02 10:05:00",
"alert_description": "error rate 12.4% over the 5% threshold on checkout",
"kind": "on_call",
"notification_kind": "on_call",
"notification_event": "firing",
"cluster_name": "o2-acme",
"alert_start_time_millis": "N/A",
"alert_end_time_millis": "N/A",
"alert_trigger_time": "0",
"alert_trigger_time_millis": "0",
"alert_trigger_time_seconds": "0",
"alert_promql_operator": null,
"alert_promql_value": null,
"oncall_team": "checkout",
"oncall_reason": "paged by policy checkout-escalation",
"oncall_stage": "1",
"oncall_priority": "P2",
"oncall_recipient": "maria@acme.com",
"oncall_response_id": "resp_31",
"oncall_ack_url": "https://o2.acme.com/api/acme/oncall/ack/resp_31",
"runbook_url": "https://wiki.acme.com/runbooks/checkout"
}
}
}
@@ -116,6 +116,10 @@ pub fn resolve_content(
unknown_variables.append(&mut unk);
let (url, mut unk) = substitute_raw(&link.url, ctx);
unknown_variables.append(&mut unk);
// A card for another kind drops a link whose URL is empty, such as a page with no runbook.
if ctx.kind != NotificationKind::Alert && url.trim().is_empty() {
continue;
}
links.push((label, url));
}
// Appended alert link: empty label — renderers supply the localized
+20 -2
View File
@@ -291,6 +291,14 @@ pub(crate) async fn bindings_for(
by.push(format!("synthetics check '{check}'"));
}
}
NotificationKind::OnCall => {
for (_, team) in
infra::table::oncall_policies::teams_using_template(binding_scope(org_id), name)
.await?
{
by.push(format!("on-call policy of team '{team}'"));
}
}
NotificationKind::Alert => {
let (_, alerts) = db::alerts::templates::cached_references(org_id, name).await;
by.extend(alerts.into_iter().map(|a| format!("alert '{a}'")));
@@ -395,6 +403,12 @@ pub async fn used_by(org_id: &str, name: &str) -> Result<UsedBy, TemplateError>
.into_iter()
.map(|(_, check)| format!("synthetics check '{check}'"))
.collect();
let on_call_names: Vec<String> =
infra::table::oncall_policies::teams_using_template(binding_scope(org_id), name)
.await?
.into_iter()
.map(|(_, team)| format!("on-call policy of team '{team}'"))
.collect();
let mut org_default: Vec<String> = Vec::new();
for (_, kinds) in pointer_orgs(org_id, name).await? {
for kind in kinds {
@@ -409,9 +423,13 @@ pub async fn used_by(org_id: &str, name: &str) -> Result<UsedBy, TemplateError>
alerts: alerts.len(),
anomaly: anomaly_names.len(),
synthetics: synthetics_names.len(),
on_call: 0,
on_call: on_call_names.len(),
org_default,
binding_names: anomaly_names.into_iter().chain(synthetics_names).collect(),
binding_names: anomaly_names
.into_iter()
.chain(synthetics_names)
.chain(on_call_names)
.collect(),
})
}
@@ -10,6 +10,7 @@ pub struct Model {
pub rungs: String,
pub destinations: String,
pub l0_json: String,
pub template: Option<String>,
pub created_at: i64,
pub updated_at: i64,
}
@@ -0,0 +1,98 @@
// 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/>.
//! `template`: an escalation policy's template override by name; NULL uses the destination's.
use sea_orm_migration::prelude::*;
const TABLE: &str = "oncall_policies";
const COLUMN: &str = "template";
#[derive(DeriveMigrationName)]
pub struct Migration;
#[derive(DeriveIden, Clone)]
enum OncallPolicies {
Table,
Template,
}
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
if manager.has_column(TABLE, COLUMN).await? {
return Ok(());
}
manager.alter_table(add_column_stmt()).await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
if !manager.has_column(TABLE, COLUMN).await? {
return Ok(());
}
manager.alter_table(drop_column_stmt()).await?;
Ok(())
}
}
fn add_column_stmt() -> TableAlterStatement {
Table::alter()
.table(OncallPolicies::Table)
.add_column(
ColumnDef::new(OncallPolicies::Template)
.string_len(256)
.null(),
)
.to_owned()
}
fn drop_column_stmt() -> TableAlterStatement {
Table::alter()
.table(OncallPolicies::Table)
.drop_column(OncallPolicies::Template)
.to_owned()
}
#[cfg(test)]
mod tests {
use collapse::*;
use super::*;
#[test]
fn postgres() {
collapsed_eq!(
&add_column_stmt().to_string(PostgresQueryBuilder),
r#"ALTER TABLE "oncall_policies" ADD COLUMN "template" varchar(256) NULL"#
);
}
#[test]
fn sqlite() {
collapsed_eq!(
&add_column_stmt().to_string(SqliteQueryBuilder),
r#"ALTER TABLE "oncall_policies" ADD COLUMN "template" varchar(256) NULL"#
);
}
#[test]
fn down_drops_the_same_column() {
collapsed_eq!(
&drop_column_stmt().to_string(PostgresQueryBuilder),
r#"ALTER TABLE "oncall_policies" DROP COLUMN "template""#
);
}
}
+3
View File
@@ -252,6 +252,7 @@ mod m20261120_000001_add_template_applies_to;
mod m20261120_000002_rename_reserved_template_collisions;
mod m20261120_000003_add_anomaly_template;
mod m20261201_000001_add_synthetics_template;
mod m20261210_000001_add_oncall_policy_template;
/// Shared body of the two `folder_id` migrations above; not a migration itself.
mod workflow_folder_id;
@@ -775,6 +776,7 @@ impl MigratorTrait for Migrator {
Box::new(m20261120_000002_rename_reserved_template_collisions::Migration),
Box::new(m20261120_000003_add_anomaly_template::Migration),
Box::new(m20261201_000001_add_synthetics_template::Migration),
Box::new(m20261210_000001_add_oncall_policy_template::Migration),
]
}
}
@@ -879,6 +881,7 @@ mod tests {
(134, "m20261120_000002_rename_reserved_template_collisions"),
(135, "m20261120_000003_add_anomaly_template"),
(136, "m20261201_000001_add_synthetics_template"),
(137, "m20261210_000001_add_oncall_policy_template"),
];
// Two branches that each claim the next version merge cleanly, so a reused number must fail
+62 -2
View File
@@ -22,10 +22,15 @@ use config::{
meta::oncall::{EscalationPolicy, PriorityRung},
utils::time::now_micros,
};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set};
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,
};
use super::{entity::oncall_policies, ttl_cache::TtlCache};
use crate::{db::get_orm_client_rw, errors};
use crate::{
db::{get_orm_client_ro, get_orm_client_rw},
errors,
};
/// The team's escalation policy, for the paging path (`06` §3).
///
@@ -99,6 +104,7 @@ fn to_policy(m: oncall_policies::Model) -> EscalationPolicy {
rungs,
destinations,
l0,
template: m.template,
},
Err(e) => {
log::error!(
@@ -190,6 +196,7 @@ async fn insert(policy: &EscalationPolicy) -> Result<EscalationPolicy, errors::E
rungs: Set(serde_json::to_string(&policy.rungs)?),
destinations: Set(serde_json::to_string(&policy.destinations)?),
l0_json: Set(serde_json::to_string(&policy.l0)?),
template: Set(policy.template.clone()),
created_at: Set(now),
updated_at: Set(now),
};
@@ -240,6 +247,58 @@ pub async fn update_rungs(
Ok(Some(updated))
}
/// Sets or clears the team's template override; false when the team has no policy row.
pub async fn set_template(
org_id: &str,
team_id: &str,
template: Option<String>,
) -> Result<bool, errors::Error> {
let client = get_orm_client_rw().await;
let res = oncall_policies::Entity::update_many()
.col_expr(
oncall_policies::Column::Template,
sea_orm::sea_query::Expr::value(template),
)
.col_expr(
oncall_policies::Column::UpdatedAt,
sea_orm::sea_query::Expr::value(now_micros()),
)
.filter(oncall_policies::Column::OrgId.eq(org_id))
.filter(oncall_policies::Column::TeamId.eq(team_id))
.exec(client)
.await?;
invalidate_and_publish(org_id, team_id).await;
Ok(res.rows_affected > 0)
}
/// `(org, team_id)` of each policy whose template override is `template`; `None` is every org.
pub async fn teams_using_template(
org_id: Option<&str>,
template: &str,
) -> Result<Vec<(String, String)>, errors::Error> {
let client = get_orm_client_ro().await;
let mut query = oncall_policies::Entity::find()
.select_only()
.column(oncall_policies::Column::OrgId)
.column(oncall_policies::Column::TeamId)
.filter(oncall_policies::Column::Template.eq(template));
if let Some(org) = org_id {
query = query.filter(oncall_policies::Column::OrgId.eq(org));
}
Ok(query.into_tuple::<(String, String)>().all(client).await?)
}
/// Every policy of every org; read once at boot for the upgrade notice.
pub async fn list_all() -> Result<Vec<EscalationPolicy>, errors::Error> {
let client = get_orm_client_ro().await;
Ok(oncall_policies::Entity::find()
.all(client)
.await?
.into_iter()
.map(to_policy)
.collect())
}
pub async fn list(org_id: &str) -> Result<Vec<EscalationPolicy>, errors::Error> {
let client = get_orm_client_rw().await;
Ok(oncall_policies::Entity::find()
@@ -282,6 +341,7 @@ mod tests {
// The column defaults, 04 §3's own: one pass, then record that nobody answered.
created_at: 10,
updated_at: 20,
template: None,
}
}
@@ -359,6 +359,7 @@ pub async fn put_policy(policy: &EscalationPolicy) -> Result<(), errors::Error>
active.rungs = Set(rungs);
active.destinations = Set(destinations);
active.l0_json = Set(l0_json);
active.template = Set(policy.template.clone());
active.updated_at = Set(now);
active.update(client).await?;
}
@@ -370,6 +371,7 @@ pub async fn put_policy(policy: &EscalationPolicy) -> Result<(), errors::Error>
rungs: Set(rungs),
destinations: Set(destinations),
l0_json: Set(l0_json),
template: Set(policy.template.clone()),
created_at: Set(now),
updated_at: Set(now),
}
+12
View File
@@ -759,6 +759,18 @@ pub async fn init() -> Result<(), anyhow::Error> {
o2_enterprise::enterprise::oncall::token::mark_spent(tag, expires_at);
}));
}
// On-call destination sends go through the shared notification path.
#[cfg(feature = "enterprise")]
{
o2_enterprise::enterprise::oncall::sender::register_oncall_sender(|n| {
Box::pin(openobserve_core::alerts::notifications::oncall::notify_destination(n))
});
if LOCAL_NODE.is_scheduler() {
tokio::spawn(
openobserve_core::alerts::notifications::oncall::log_mail_destination_policies(),
);
}
}
// Every node: the validator, which reads the token cache, runs wherever HTTP is served.
#[cfg(feature = "enterprise")]
if openobserve_core::fleet::enabled() {
+1
View File
@@ -266,6 +266,7 @@ async fn warn_about(
// No alert behind this one, so no runbook to follow.
runbook_url: None,
investigation: vec![],
event: String::new(),
},
recipient: recipient.clone(),
channel: config::meta::oncall::Channel::Email,
+1
View File
@@ -1643,6 +1643,7 @@ mod tests {
l0_json: Set("{}".to_string()),
created_at: Set(now),
updated_at: Set(now),
template: Set(None),
})
.await;
insert_active(oncall_ownership_rules::ActiveModel {