first (untested) version

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "restic_alarm"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+aws-config = { version = "1.1.7", features = ["behavior-version-latest"] }
+aws-sdk-s3 = "1.21.0"
+tokio = { version = "1", features = ["full"] }
+serde = { version = "1.0", features = ["derive"] }
+toml = { version = "0.5.8" }
+eyre = "0.6.12"
+lettre = { version = "0.11.6", features = ["builder", "hostname", "sendmail-transport", "tokio1", "tokio1-native-tls"], default-features = false }

src/main.rs

@@ -0,0 +1,222 @@
+use std::collections::HashMap;
+use std::time::{SystemTime,Duration};
+use std::convert::TryFrom;
+use serde::{Deserialize, Serialize};
+use aws_config::Region;
+use aws_sdk_s3 as s3;
+use eyre::{WrapErr,OptionExt};
+
+const RESTIC_ALARM_BUCKET: &str = "restic-alarm-state";
+const RESTIC_ALARM_STATE_FILE: &str = "state.toml";
+const RESTIC_ALARM_WATCH_DIR: &str = "watch/";
+
+const S3_REGION: &str = "infracoll";
+const S3_ENDPOINT: &str = "http://garage.isomorphis.me:3900";
+
+#[derive(Serialize, Deserialize, Clone)]
+struct State {
+    last_alert: HashMap<String, u64>
+}
+
+impl State {
+    fn last_alert(&self, repo: &str) -> Option<Duration> {
+        let time = *self.last_alert.get(repo)?;
+        let time = SystemTime::UNIX_EPOCH + Duration::from_secs(time);
+        SystemTime::now().duration_since(time).ok()
+    }
+
+    fn update_last_alert(&mut self, repo: &str) {
+        let time = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap();
+        self.last_alert.insert(repo.to_owned(), time.as_secs());
+    }
+}
+
+fn default_alert_interval() -> u64 { 1 }
+fn default_alert_duration() -> u64 { 7 }
+
+#[derive(Serialize, Deserialize)]
+struct RepoConfig {
+    email: String,
+    // all durations below are measured in days
+    inactivity: u64,
+    #[serde(default = "default_alert_interval")]
+    alert_interval: u64,
+    #[serde(default = "default_alert_duration")]
+    alert_duration: u64,
+}
+
+fn is_alert_needed(cfg: &RepoConfig, inactive_for: Duration, since_last_alert: Option<Duration>) -> bool {
+    let one_day: u64 = 3600 * 24;
+    inactive_for.as_secs() >= cfg.inactivity * one_day
+    && inactive_for.as_secs() <= cfg.alert_duration * one_day
+    && match since_last_alert {
+        Some(d) => d.as_secs() >= cfg.alert_interval * one_day,
+        None => true,
+    }
+}
+
+async fn send_email(email: &str, repo: &str, inactive: Duration) -> eyre::Result<()> {
+    use lettre::{Message, AsyncSendmailTransport, AsyncTransport};
+    let email = Message::builder()
+        .from("infracoll <infracoll>".parse().unwrap())
+        .to(email.parse()?)
+        .subject(format!("restic-alarm: inactive repository {}", repo))
+        .body(format!("Alert: Repository {} has been inactive for {} days.\n",
+              repo, inactive.as_secs() / (3600*24)))
+        .unwrap();
+    let mailer = AsyncSendmailTransport::new();
+    mailer.send(email).await?;
+    Ok(())
+}
+
+use s3::error::SdkError;
+use s3::operation::get_object::GetObjectError;
+
+fn is_no_such_key_error<R>(err: &SdkError<GetObjectError,R>) -> bool {
+    match err {
+        SdkError::ServiceError(e) => {
+            match e.err() {
+                GetObjectError::NoSuchKey(_) => true,
+                _ => false,
+            }
+        },
+        _ => false,
+    }
+}
+
+async fn write_state(client: &s3::Client, state: &State) -> eyre::Result<()> {
+    let data = toml::to_vec(state)?;
+    client.put_object()
+          .bucket(RESTIC_ALARM_BUCKET)
+          .key(RESTIC_ALARM_STATE_FILE)
+          .body(s3::primitives::ByteStream::from(data))
+          .send().await?;
+    Ok(())
+}
+
+async fn read_state(client: &s3::Client) -> eyre::Result<State> {
+    let output = client
+        .get_object()
+        .bucket(RESTIC_ALARM_BUCKET)
+        .key(RESTIC_ALARM_STATE_FILE)
+        .send().await;
+    match output {
+        Ok(output) => {
+            let state_s = output.body.collect().await
+                                               .wrap_err("error reading state file")?
+                                               .into_bytes();
+            let state: State = toml::from_slice(&state_s).expect("invalid state file");
+            Ok(state)
+        },
+        Err(e) if is_no_such_key_error(&e) => {
+            let state = State { last_alert: HashMap::new() };
+            write_state(client, &state).await?;
+            Ok(state)
+        },
+        Err(err) => Err(err)?,
+    }
+}
+
+async fn to_watch(client: &s3::Client) -> eyre::Result<Vec<String>> {
+    let mut to_watch = Vec::new();
+    let mut objs_resp = client
+        .list_objects_v2()
+        .bucket(RESTIC_ALARM_BUCKET)
+        .prefix(RESTIC_ALARM_WATCH_DIR)
+        .into_paginator().send();
+    while let Some(res) = objs_resp.next().await {
+        let output = res?;
+        for obj in output.contents() {
+            let key = obj.key().ok_or_eyre("object with no key")?;
+            to_watch.push(key.strip_prefix(RESTIC_ALARM_WATCH_DIR).unwrap().to_owned())
+        }
+    }
+    Ok(to_watch)
+}
+
+async fn read_repo_config(client: &s3::Client, repo: &str) -> eyre::Result<RepoConfig> {
+    let repo_config_path = RESTIC_ALARM_WATCH_DIR.to_owned() + repo;
+    let output = client
+        .get_object()
+        .bucket(RESTIC_ALARM_BUCKET)
+        .key(&repo_config_path)
+        .send().await?;
+    let config_s = output.body.collect().await
+        .wrap_err_with(|| format!("reading repo config file {}", &repo_config_path))?
+        .into_bytes();
+    let config = toml::from_slice(&config_s)
+        .wrap_err_with(|| format!("parsing repo config file {}", &repo_config_path))?;
+    Ok(config)
+}
+
+async fn repo_last_snapshot(client: &s3::Client, repo: &str) -> eyre::Result<Option<Duration>> {
+    let mut objs = client
+        .list_objects_v2()
+        .bucket(repo)
+        .prefix("snapshots/")
+        .into_paginator().send();
+    let mut modtimes = Vec::new();
+    while let Some(res) = objs.next().await {
+        let output = res?;
+        for obj in output.contents() {
+            if let Some(last_mod) = obj.last_modified() {
+                modtimes.push(SystemTime::try_from(last_mod.clone()).unwrap());
+            }
+        }
+    }
+    modtimes.sort();
+    match modtimes.last() {
+        None => Ok(None),
+        Some(time) =>
+            Ok(SystemTime::now().duration_since(*time).ok())
+    }
+}
+
+// this function can fail for reasons that depend on the user-provided 'repo' config
+// (e.g. if it fails to parse).
+// So the error must not be propagated to the toplevel, which would abort the
+// alert for remaining repositories; it should instead just be reported/logged.
+async fn check_repo(client: &s3::Client, state: &State, repo: &str) -> eyre::Result<State> {
+    let config = read_repo_config(client, repo).await?;
+    let mut state = state.to_owned();
+    match repo_last_snapshot(client, repo).await? {
+        Some(last_snapshot) => {
+            let is_alert = is_alert_needed(&config, last_snapshot, state.last_alert(repo));
+            if is_alert {
+                println!("Sending alert to {} about bucket {}", config.email, repo);
+                send_email(&config.email, repo, last_snapshot).await?;
+            }
+            state.update_last_alert(repo)
+        },
+        None => (),
+    }
+    Ok(state)
+}
+
+#[::tokio::main]
+async fn main() -> eyre::Result<()> {
+    let sdk_config = aws_config::load_from_env().await;
+    let config = aws_sdk_s3::config::Builder::from(&sdk_config)
+        .region(Region::new(S3_REGION))
+        .endpoint_url(S3_ENDPOINT)
+        .force_path_style(true)
+        .build();
+    let client = aws_sdk_s3::Client::from_conf(config);
+
+    let repos = to_watch(&client).await?;
+    let mut state = read_state(&client).await?;
+    for repo in repos {
+        match check_repo(&client, &state, &repo).await {
+            Ok(new_state) => {
+                state = new_state;
+                write_state(&client, &state).await?;
+            },
+            Err(err) => {
+                // is this the best way to log the error?
+                eprintln!("Error: {:?}", err)
+            }
+        }
+    }
+
+    Ok(())
+}