opssignal_core/providers/
pagerduty.rs

1use crate::delivery::Provider;
2use crate::signal::{Severity, Signal, SignalError};
3use crate::template::{self, Templates};
4
5/// Events API v2 endpoint for US-hosted accounts. EU accounts use
6/// `https://events.eu.pagerduty.com/v2/enqueue`, set through `url`.
7pub const DEFAULT_EVENTS_URL: &str = "https://events.pagerduty.com/v2/enqueue";
8
9/// PagerDuty rejects longer values with a 400, which retrying cannot
10/// fix, so they are cut to fit instead.
11const MAX_SUMMARY_CHARS: usize = 1024;
12const MAX_DEDUP_KEY_CHARS: usize = 255;
13
14/// Triggers a PagerDuty alert through the Events API v2. The signal's
15/// `dedup_key` becomes PagerDuty's, so a repeat of the same signal adds
16/// to the open alert instead of paging again.
17pub struct PagerDutyProvider {
18    routing_key: String,
19    pub url: String,
20    client: reqwest::Client,
21    templates: Option<Templates>,
22}
23
24impl PagerDutyProvider {
25    pub fn new(routing_key: String, url: Option<String>) -> Self {
26        PagerDutyProvider {
27            routing_key,
28            url: url.unwrap_or_else(|| DEFAULT_EVENTS_URL.to_string()),
29            client: reqwest::Client::new(),
30            templates: None,
31        }
32    }
33
34    /// Uses `templates` for the parts they set; see [`crate::template`].
35    pub fn with_templates(mut self, templates: Option<Templates>) -> Self {
36        self.templates = templates;
37        self
38    }
39
40    /// PagerDuty has no `success` severity; a success signal is
41    /// informational.
42    fn severity_for(severity: Severity) -> &'static str {
43        match severity {
44            Severity::Info | Severity::Success => "info",
45            Severity::Warning => "warning",
46            Severity::Error => "error",
47            Severity::Critical => "critical",
48        }
49    }
50
51    fn event_for(&self, signal: &Signal) -> serde_json::Value {
52        let mut details = serde_json::Map::new();
53        details.insert("signal_id".into(), signal.id.to_string().into());
54        details.insert("event_type".into(), signal.event_type.clone().into());
55        details.insert("severity".into(), signal.severity.to_string().into());
56        let optional = [
57            ("message", &signal.message),
58            ("environment", &signal.environment),
59            ("team", &signal.team),
60            ("run_id", &signal.run_id),
61            ("correlation_id", &signal.correlation_id),
62            ("trace_id", &signal.trace_id),
63        ];
64        for (key, value) in optional {
65            if let Some(value) = value {
66                details.insert(key.into(), value.clone().into());
67            }
68        }
69        if signal.metadata.as_object().is_some_and(|m| !m.is_empty()) {
70            details.insert("metadata".into(), signal.metadata.clone());
71        }
72
73        let summary = template::render(&self.templates, "summary", signal, &[])
74            .unwrap_or_else(|| signal.title.clone());
75        let mut payload = serde_json::json!({
76            "summary": truncate(&summary, MAX_SUMMARY_CHARS),
77            "source": signal.source,
78            "severity": Self::severity_for(signal.severity),
79            "timestamp": signal.timestamp.to_rfc3339(),
80            "class": signal.event_type,
81            "custom_details": details,
82        });
83        if let Some(component) = &signal.component {
84            payload["component"] = component.clone().into();
85        }
86        if let Some(service) = &signal.service {
87            payload["group"] = service.clone().into();
88        }
89
90        let mut event = serde_json::json!({
91            "routing_key": self.routing_key,
92            "event_action": "trigger",
93            "payload": payload,
94        });
95        if let Some(key) = &signal.dedup_key {
96            event["dedup_key"] = truncate(key, MAX_DEDUP_KEY_CHARS).into();
97        }
98        if let Some(log_url) = &signal.log_url {
99            event["links"] = serde_json::json!([{ "href": log_url, "text": "Logs" }]);
100        }
101        event
102    }
103}
104
105fn truncate(s: &str, max_chars: usize) -> String {
106    match s.char_indices().nth(max_chars) {
107        Some((end, _)) => s[..end].to_string(),
108        None => s.to_string(),
109    }
110}
111
112#[async_trait::async_trait]
113impl Provider for PagerDutyProvider {
114    fn name(&self) -> &'static str {
115        "pagerduty"
116    }
117
118    /// The routing key is a credential, so it is left out.
119    fn preview(&self, signal: &Signal) -> Option<serde_json::Value> {
120        let mut event = self.event_for(signal);
121        event["routing_key"] = "<redacted>".into();
122        Some(event)
123    }
124
125    async fn send(&self, signal: &Signal) -> Result<(), SignalError> {
126        let resp = self
127            .client
128            .post(&self.url)
129            .json(&self.event_for(signal))
130            .send()
131            .await
132            .map_err(|e| SignalError::Delivery(e.to_string()))?;
133
134        if !resp.status().is_success() {
135            return Err(SignalError::Delivery(format!(
136                "pagerduty events API returned status {}",
137                resp.status()
138            )));
139        }
140        Ok(())
141    }
142}
143
144#[cfg(test)]
145mod tests {
146    use super::*;
147    use crate::signal::SignalInput;
148    use wiremock::matchers::{body_partial_json, header, method, path};
149    use wiremock::{Mock, MockServer, ResponseTemplate};
150
151    fn signal(severity: Severity) -> Signal {
152        Signal::from_input(SignalInput {
153            source: "airflow".into(),
154            service: Some("orders".into()),
155            component: Some("dbt".into()),
156            event_type: "task_failed".into(),
157            title: "dbt model failed".into(),
158            message: Some("model orders timed out".into()),
159            severity,
160            environment: Some("prod".into()),
161            log_url: Some("https://logs.example.test/run/1".into()),
162            metadata: Some(serde_json::json!({ "dag": "daily" })),
163            ..Default::default()
164        })
165    }
166
167    #[test]
168    fn severities_map_to_pagerduty_levels() {
169        assert_eq!(PagerDutyProvider::severity_for(Severity::Info), "info");
170        assert_eq!(PagerDutyProvider::severity_for(Severity::Success), "info");
171        assert_eq!(
172            PagerDutyProvider::severity_for(Severity::Warning),
173            "warning"
174        );
175        assert_eq!(PagerDutyProvider::severity_for(Severity::Error), "error");
176        assert_eq!(
177            PagerDutyProvider::severity_for(Severity::Critical),
178            "critical"
179        );
180    }
181
182    #[test]
183    fn defaults_to_the_us_events_endpoint() {
184        let provider = PagerDutyProvider::new("key".into(), None);
185        assert_eq!(provider.url, DEFAULT_EVENTS_URL);
186    }
187
188    #[test]
189    fn event_carries_signal_fields() {
190        let provider = PagerDutyProvider::new("R0UT1NG".into(), None);
191        let signal = signal(Severity::Critical);
192        let event = provider.event_for(&signal);
193
194        assert_eq!(event["routing_key"], "R0UT1NG");
195        assert_eq!(event["event_action"], "trigger");
196        assert_eq!(event["dedup_key"], signal.dedup_key.clone().unwrap());
197        assert_eq!(event["links"][0]["href"], "https://logs.example.test/run/1");
198        let payload = &event["payload"];
199        assert_eq!(payload["summary"], "dbt model failed");
200        assert_eq!(payload["source"], "airflow");
201        assert_eq!(payload["severity"], "critical");
202        assert_eq!(payload["component"], "dbt");
203        assert_eq!(payload["group"], "orders");
204        assert_eq!(payload["class"], "task_failed");
205        assert_eq!(payload["timestamp"], signal.timestamp.to_rfc3339());
206        let details = &payload["custom_details"];
207        assert_eq!(details["signal_id"], signal.id.to_string());
208        assert_eq!(details["message"], "model orders timed out");
209        assert_eq!(details["environment"], "prod");
210        assert_eq!(details["metadata"]["dag"], "daily");
211        assert!(details.get("team").is_none());
212    }
213
214    #[test]
215    fn minimal_signal_omits_optional_fields() {
216        let provider = PagerDutyProvider::new("k".into(), None);
217        let signal = Signal::from_input(SignalInput {
218            source: "ci".into(),
219            event_type: "build_failed".into(),
220            title: "main is red".into(),
221            ..Default::default()
222        });
223        let event = provider.event_for(&signal);
224        assert!(event.get("links").is_none());
225        assert!(event["payload"].get("component").is_none());
226        assert!(event["payload"].get("group").is_none());
227        assert!(event["payload"]["custom_details"].get("metadata").is_none());
228    }
229
230    #[test]
231    fn long_summary_and_dedup_key_are_cut_to_pagerduty_limits() {
232        let provider = PagerDutyProvider::new("k".into(), None);
233        let signal = Signal::from_input(SignalInput {
234            source: "ci".into(),
235            event_type: "build_failed".into(),
236            // Multi-byte characters, so a byte-based cut would panic.
237            title: "é".repeat(2000),
238            dedup_key: Some("ü".repeat(300)),
239            ..Default::default()
240        });
241        let event = provider.event_for(&signal);
242        let summary = event["payload"]["summary"].as_str().unwrap();
243        assert_eq!(summary.chars().count(), MAX_SUMMARY_CHARS);
244        let key = event["dedup_key"].as_str().unwrap();
245        assert_eq!(key.chars().count(), MAX_DEDUP_KEY_CHARS);
246    }
247
248    #[tokio::test]
249    async fn posts_the_event_to_the_events_api() {
250        let server = MockServer::start().await;
251        Mock::given(method("POST"))
252            .and(path("/v2/enqueue"))
253            .and(header("content-type", "application/json"))
254            .and(body_partial_json(serde_json::json!({
255                "routing_key": "R0UT1NG",
256                "event_action": "trigger",
257                "payload": { "summary": "dbt model failed", "severity": "error" },
258            })))
259            .respond_with(ResponseTemplate::new(202))
260            .expect(1)
261            .mount(&server)
262            .await;
263
264        PagerDutyProvider::new(
265            "R0UT1NG".into(),
266            Some(format!("{}/v2/enqueue", server.uri())),
267        )
268        .send(&signal(Severity::Error))
269        .await
270        .expect("202 is success");
271    }
272
273    #[tokio::test]
274    async fn rejected_event_is_a_delivery_error() {
275        let server = MockServer::start().await;
276        Mock::given(method("POST"))
277            .respond_with(ResponseTemplate::new(400))
278            .mount(&server)
279            .await;
280
281        let err = PagerDutyProvider::new("bad".into(), Some(server.uri()))
282            .send(&signal(Severity::Error))
283            .await
284            .expect_err("400 must fail");
285        assert!(matches!(err, SignalError::Delivery(_)));
286        assert!(err.to_string().contains("400"));
287    }
288
289    #[tokio::test]
290    async fn rate_limit_is_a_delivery_error() {
291        let server = MockServer::start().await;
292        Mock::given(method("POST"))
293            .respond_with(ResponseTemplate::new(429))
294            .mount(&server)
295            .await;
296
297        let err = PagerDutyProvider::new("k".into(), Some(server.uri()))
298            .send(&signal(Severity::Error))
299            .await
300            .expect_err("429 must fail so the retry loop backs off");
301        assert!(err.to_string().contains("429"));
302    }
303
304    #[test]
305    fn summary_template_replaces_the_title_and_is_cut_to_fit() {
306        let provider = PagerDutyProvider::new("k".into(), None).with_templates(template::for_test(
307            "pagerduty",
308            &[("summary", "[{{ environment }}] {{ service }}: {{ title }}")],
309        ));
310        let event = provider.event_for(&signal(Severity::Error));
311        assert_eq!(
312            event["payload"]["summary"],
313            "[prod] orders: dbt model failed"
314        );
315
316        let long = PagerDutyProvider::new("k".into(), None).with_templates(template::for_test(
317            "pagerduty",
318            &[("summary", "{% for i in range(2000) %}x{% endfor %}")],
319        ));
320        let event = long.event_for(&signal(Severity::Error));
321        assert_eq!(
322            event["payload"]["summary"]
323                .as_str()
324                .unwrap()
325                .chars()
326                .count(),
327            MAX_SUMMARY_CHARS
328        );
329    }
330}
opssignal_core/providers/pagerduty.rs

opssignal_core/providers/
pagerduty.rs
