forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
4.1 KiB
1//! Healthchecks.io style pings for scheduled backups. `GET <url>/start` when a run begins, then
2//! `GET <url>` on success or `POST <url>/fail` with the error text. The URL is admin-configured and
3//! trusted, so it is not SSRF-guarded. Pings never fail a backup: an error is logged and dropped.
4
5use std::time::Duration;
6
7const PING_TIMEOUT: Duration = Duration::from_secs(10);
8/// Longest error text sent with a failure ping.
9const MAX_BODY_CHARS: usize = 1000;
10
11/// Pings `<url>/start`.
12pub async fn start(url: &str) {
13 send(url, "/start", None).await;
14}
15
16/// Pings `<url>` after a good run.
17pub async fn success(url: &str) {
18 send(url, "", None).await;
19}
20
21/// Pings `<url>/fail` with the reason.
22pub async fn failure(url: &str, reason: &str) {
23 let body: String = reason.chars().take(MAX_BODY_CHARS).collect();
24 send(url, "/fail", Some(body)).await;
25}
26
27async fn send(url: &str, suffix: &str, body: Option<String>) {
28 let target = format!("{}{suffix}", url.trim_end_matches('/'));
29 let client = match reqwest::Client::builder().timeout(PING_TIMEOUT).build() {
30 Ok(client) => client,
31 Err(e) => {
32 tracing::warn!(error = %e, "healthcheck client could not be built");
33 return;
34 }
35 };
36 let request = match body {
37 Some(text) => client.post(&target).body(text),
38 None => client.get(&target),
39 };
40 match request.send().await {
41 Ok(res) if res.status().is_success() => {}
42 Ok(res) => tracing::warn!(status = %res.status(), "healthcheck ping was refused"),
43 Err(e) => tracing::warn!(error = %e, "healthcheck ping failed"),
44 }
45}
46
47#[cfg(test)]
48mod tests {
49 use super::*;
50 use std::sync::{Arc, Mutex};
51 use tokio::io::{AsyncReadExt, AsyncWriteExt};
52 use tokio::net::TcpListener;
53
54 /// Serves `count` requests and returns the request lines and bodies seen.
55 async fn serve(count: usize) -> (String, Arc<Mutex<Vec<(String, String)>>>) {
56 let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
57 let addr = listener.local_addr().expect("addr");
58 let seen = Arc::new(Mutex::new(Vec::new()));
59 let log = Arc::clone(&seen);
60 tokio::spawn(async move {
61 for _ in 0..count {
62 let (mut sock, _) = listener.accept().await.expect("accept");
63 let mut buf = vec![0u8; 8192];
64 let n = sock.read(&mut buf).await.expect("read");
65 let text = String::from_utf8_lossy(&buf[..n]).into_owned();
66 let line = text.lines().next().unwrap_or_default().to_owned();
67 let body = text.split("\r\n\r\n").nth(1).unwrap_or_default().to_owned();
68 log.lock().expect("lock").push((line, body));
69 sock.write_all(
70 b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
71 )
72 .await
73 .expect("write");
74 }
75 });
76 (format!("http://{addr}/ping/abc/"), seen)
77 }
78
79 #[tokio::test]
80 async fn start_success_and_failure_hit_the_expected_paths() {
81 let (url, seen) = serve(3).await;
82 start(&url).await;
83 success(&url).await;
84 failure(&url, "backup exploded").await;
85 // Give the server task a moment to record the last request.
86 for _ in 0..50 {
87 if seen.lock().expect("lock").len() == 3 {
88 break;
89 }
90 tokio::time::sleep(Duration::from_millis(10)).await;
91 }
92 let seen = seen.lock().expect("lock").clone();
93 assert_eq!(seen.len(), 3);
94 assert_eq!(seen[0].0, "GET /ping/abc/start HTTP/1.1");
95 assert_eq!(seen[1].0, "GET /ping/abc HTTP/1.1");
96 assert_eq!(seen[2].0, "POST /ping/abc/fail HTTP/1.1");
97 assert_eq!(seen[2].1, "backup exploded");
98 }
99
100 #[tokio::test]
101 async fn an_unreachable_endpoint_does_not_panic() {
102 // Nothing listens on this port: the ping logs and returns.
103 let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
104 let addr = listener.local_addr().expect("addr");
105 drop(listener);
106 start(&format!("http://{addr}")).await;
107 }
108}