opssignal_core/providers/
slack.rs

1use crate::delivery::Provider;
2use crate::signal::{Severity, Signal, SignalError};
3use crate::template::{self, Templates};
4
5pub struct SlackProvider {
6    pub webhook_url: String,
7    client: reqwest::Client,
8    templates: Option<Templates>,
9}
10
11impl SlackProvider {
12    pub fn new(webhook_url: String) -> Self {
13        SlackProvider {
14            webhook_url,
15            client: reqwest::Client::new(),
16            templates: None,
17        }
18    }
19
20    /// Uses `templates` for the parts they set; see [`crate::template`].
21    pub fn with_templates(mut self, templates: Option<Templates>) -> Self {
22        self.templates = templates;
23        self
24    }
25
26    fn emoji_for(severity: Severity) -> &'static str {
27        match severity {
28            Severity::Info => ":information_source:",
29            Severity::Success => ":white_check_mark:",
30            Severity::Warning => ":warning:",
31            Severity::Error => ":x:",
32            Severity::Critical => ":rotating_light:",
33        }
34    }
35}
36
37#[async_trait::async_trait]
38impl Provider for SlackProvider {
39    fn name(&self) -> &'static str {
40        "slack"
41    }
42
43    fn preview(&self, signal: &Signal) -> Option<serde_json::Value> {
44        Some(self.body_for(signal))
45    }
46
47    async fn send(&self, signal: &Signal) -> Result<(), SignalError> {
48        let body = self.body_for(signal);
49
50        let resp = self
51            .client
52            .post(&self.webhook_url)
53            .json(&body)
54            .send()
55            .await
56            .map_err(|e| SignalError::Delivery(e.to_string()))?;
57
58        if !resp.status().is_success() {
59            return Err(SignalError::Delivery(format!(
60                "slack webhook returned status {}",
61                resp.status()
62            )));
63        }
64        Ok(())
65    }
66}
67
68impl SlackProvider {
69    fn body_for(&self, signal: &Signal) -> serde_json::Value {
70        let emoji = Self::emoji_for(signal.severity);
71        let text = template::render(&self.templates, "text", signal, &[("emoji", emoji.into())])
72            .unwrap_or_else(|| {
73                format!(
74                    "{emoji} *{}*\n{}",
75                    signal.title,
76                    signal.message.clone().unwrap_or_default()
77                )
78            });
79        serde_json::json!({ "text": text })
80    }
81}
82
83#[cfg(test)]
84mod tests {
85    use super::*;
86    use crate::signal::SignalInput;
87    use wiremock::matchers::{body_json, header, method, path};
88    use wiremock::{Mock, MockServer, ResponseTemplate};
89
90    fn signal(severity: Severity, message: Option<&str>) -> Signal {
91        Signal::from_input(SignalInput {
92            source: "airflow".into(),
93            event_type: "task_failed".into(),
94            title: "dbt model failed".into(),
95            message: message.map(str::to_string),
96            severity,
97            ..Default::default()
98        })
99    }
100
101    #[test]
102    fn every_severity_has_a_distinct_emoji() {
103        let all = [
104            Severity::Info,
105            Severity::Success,
106            Severity::Warning,
107            Severity::Error,
108            Severity::Critical,
109        ];
110        let mut emojis: Vec<&str> = all.iter().map(|s| SlackProvider::emoji_for(*s)).collect();
111        emojis.sort_unstable();
112        emojis.dedup();
113        assert_eq!(emojis.len(), all.len());
114        assert_eq!(
115            SlackProvider::emoji_for(Severity::Critical),
116            ":rotating_light:"
117        );
118    }
119
120    #[tokio::test]
121    async fn posts_slack_formatted_text_to_the_webhook() {
122        let server = MockServer::start().await;
123        Mock::given(method("POST"))
124            .and(path("/services/T/B/X"))
125            .and(header("content-type", "application/json"))
126            .and(body_json(serde_json::json!({
127                "text": ":x: *dbt model failed*\nmodel orders timed out"
128            })))
129            .respond_with(ResponseTemplate::new(200))
130            .expect(1)
131            .mount(&server)
132            .await;
133
134        let provider = SlackProvider::new(format!("{}/services/T/B/X", server.uri()));
135        provider
136            .send(&signal(Severity::Error, Some("model orders timed out")))
137            .await
138            .expect("delivered");
139    }
140
141    #[tokio::test]
142    async fn missing_message_sends_title_only() {
143        let server = MockServer::start().await;
144        Mock::given(method("POST"))
145            .and(body_json(serde_json::json!({
146                "text": ":information_source: *dbt model failed*\n"
147            })))
148            .respond_with(ResponseTemplate::new(200))
149            .expect(1)
150            .mount(&server)
151            .await;
152
153        SlackProvider::new(server.uri())
154            .send(&signal(Severity::Info, None))
155            .await
156            .expect("delivered");
157    }
158
159    #[tokio::test]
160    async fn non_success_status_is_a_delivery_error() {
161        let server = MockServer::start().await;
162        Mock::given(method("POST"))
163            .respond_with(ResponseTemplate::new(500))
164            .mount(&server)
165            .await;
166
167        let err = SlackProvider::new(server.uri())
168            .send(&signal(Severity::Error, None))
169            .await
170            .expect_err("500 must fail");
171        assert!(matches!(err, SignalError::Delivery(_)));
172        assert!(err.to_string().contains("500"));
173    }
174
175    #[tokio::test]
176    async fn unreachable_webhook_is_a_delivery_error() {
177        // Bind then drop a listener so the port is known to be closed.
178        let port = std::net::TcpListener::bind("127.0.0.1:0")
179            .expect("bind")
180            .local_addr()
181            .expect("addr")
182            .port();
183
184        let err = SlackProvider::new(format!("http://127.0.0.1:{port}/hook"))
185            .send(&signal(Severity::Error, None))
186            .await
187            .expect_err("connection refused must fail");
188        assert!(matches!(err, SignalError::Delivery(_)));
189    }
190
191    #[tokio::test]
192    async fn text_template_replaces_the_default() {
193        let server = MockServer::start().await;
194        Mock::given(method("POST"))
195            .and(body_json(serde_json::json!({
196                "text": ":x: airflow/task_failed: dbt model failed (model orders timed out)"
197            })))
198            .respond_with(ResponseTemplate::new(200))
199            .expect(1)
200            .mount(&server)
201            .await;
202
203        SlackProvider::new(server.uri())
204            .with_templates(template::for_test(
205                "slack",
206                &[(
207                    "text",
208                    "{{ emoji }} {{ source }}/{{ event_type }}: {{ title }}{% if message %} ({{ message }}){% endif %}",
209                )],
210            ))
211            .send(&signal(Severity::Error, Some("model orders timed out")))
212            .await
213            .expect("delivered");
214    }
215
216    #[tokio::test]
217    async fn failing_template_sends_the_default() {
218        let server = MockServer::start().await;
219        Mock::given(method("POST"))
220            .and(body_json(serde_json::json!({
221                "text": ":x: *dbt model failed*\n"
222            })))
223            .respond_with(ResponseTemplate::new(200))
224            .expect(1)
225            .mount(&server)
226            .await;
227
228        SlackProvider::new(server.uri())
229            .with_templates(template::for_test("slack", &[("text", "{{ title + 1 }}")]))
230            .send(&signal(Severity::Error, None))
231            .await
232            .expect("delivered");
233    }
234}
opssignal_core/providers/slack.rs

opssignal_core/providers/
slack.rs
