Automatically delete old messages

snitch-cli/src/main.rs

@@ -1,3 +1,5 @@
+use std::io::{stdin, BufRead};
+
 use chrono::Local;
 use clap::Parser;
 use eyre::{eyre, WrapErr};
@@ -17,9 +19,13 @@ struct Opt {
     #[clap(short, long, env = "SNITCH_SEVERITY", default_value = "Error")]
     severity: Severity,
 
-    /// Name of this service
+    /// Treat each line of the message as its own log message.
+    #[clap(short, long, env = "SNITCH_LINES")]
+    lines: bool,
+
+    /// Log message to snitch. If this is omitted, the message will be read from stdin.
     #[clap(env = "SNITCH_MESSAGE")]
-    message: String,
+    message: Option<String>,
 }
 
 fn main() {
@@ -27,15 +33,55 @@ fn main() {
         eprintln!("snitch error: {e:?}");
     }
 }
+
 fn run() -> eyre::Result<()> {
     let opt = Opt::parse();
     color_eyre::install()?;
 
+    if let Some(message) = &opt.message {
+        log_from(&opt, message.as_bytes())
+    } else {
+        let stdin = stdin();
+        log_from(&opt, stdin.lock())
+    }
+}
+
+/// Consume a reader and log everything to snitch
+fn log_from(opt: &Opt, r: impl BufRead) -> eyre::Result<()> {
+    if opt.lines {
+        log_lines(opt, r)
+    } else {
+        log_reader(opt, r)
+    }
+}
+
+/// Consume a reader and log everything as a single message to snitch
+fn log_reader(opt: &Opt, mut r: impl BufRead) -> eyre::Result<()> {
+    let mut s = String::new();
+    r.read_to_string(&mut s)?;
+    log_message(opt, &s)
+}
+
+/// Log each line from a reader to snitch
+fn log_lines(opt: &Opt, r: impl BufRead) -> eyre::Result<()> {
+    for line in r.lines() {
+        let line = line.wrap_err("Failed to read from stdin")?;
+        let line = line.trim();
+        if !line.is_empty() {
+            log_message(opt, line)?;
+        }
+    }
+
+    Ok(())
+}
+
+/// Log a single message to snitch
+fn log_message(opt: &Opt, message: &str) -> eyre::Result<()> {
     let message = LogMsg {
         time: Some(Local::now()),
         severity: opt.severity,
-        service: opt.service,
+        service: opt.service.clone(),
-        message: opt.message,
+        message: message.to_string(),
         hostname: probe_hostname(),
         file: None,
         line: None,

snitch-lib/src/message.rs

@@ -4,11 +4,11 @@ use chrono::{DateTime, Local};
 use eyre::{eyre, ContextCompat};
 use serde::{Deserialize, Serialize};
 
-#[derive(Clone, Debug, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Severity {
-    Fatal,
-    Error,
     Warning,
+    Error,
+    Fatal,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -32,13 +32,17 @@ pub struct LogMsg {
 }
 
 impl LogMsg {
-    pub fn new(service: String, message: String, hostname: String) -> Self {
+    pub fn new(
+        service: impl Into<String>,
+        message: impl Into<String>,
+        hostname: impl Into<String>,
+    ) -> Self {
         Self {
             time: Some(Local::now()),
             severity: Severity::Error,
-            service,
+            service: service.into(),
-            message,
+            message: message.into(),
-            hostname,
+            hostname: hostname.into(),
             file: None,
             line: None,
         }

snitch-web/Cargo.toml

@@ -17,3 +17,6 @@ eyre = "0.6.12"
 color-eyre = "0.6.3"
 redb = { version = "2.1.3", features = ["logging"] }
 rmp-serde = "1.3.0"
+
+[dev-dependencies]
+rand = "0.8.5"

snitch-web/snitch.service

@@ -0,0 +1,13 @@
+[Unit]
+Description=snitch: Simple error logging web service
+
+[Service]
+Type=simple
+Environment=SNITCH_DB=/var/lib/snitch/snitch.db
+Environment=ROCKET_TEMPLATE_DIR=/var/lib/snitch/templates
+ExecStart=snitch
+
+Restart=always
+RestartSec=100ms
+RestartMaxDelaySec=10s
+RestartSteps=10

snitch-web/src/dashboard.rs

@@ -1,33 +1,49 @@
+use std::{cmp::max, collections::BTreeMap};
+
 use rocket::{http::Status, response::content::RawHtml, State};
 use rocket_dyn_templates::{context, Template};
 use serde::Serialize;
+use snitch_lib::Severity;
 
 use crate::database::Database;
 
 #[get("/")]
 pub async fn index(db: &State<Database>) -> Result<RawHtml<Template>, Status> {
+    db.drop_old_messages().map_err(|e| {
+        log::error!("failed to query database: {e}");
+        Status::InternalServerError
+    })?;
+
     // TODO:
     // maybe only show most recent error for each service in dashboard view, i.e.:
     // select distinct on (service) time, message from (select * from record order by time desc) as foo;
-    let messages = db.get_all_log_messages().map_err(|e| {
+    let records = db.get_all_log_messages().map_err(|e| {
         log::error!("failed to query database: {e}");
         Status::InternalServerError
     })?;
 
+    #[derive(Serialize)]
+    struct ServiceRecords<'a> {
+        first_record: &'a Record,
+        last_record: &'a Record,
+        severity: Severity,
+        all_records: Vec<&'a Record>,
+    }
+
     #[derive(Serialize)]
     struct Record {
         time: String,
-        severity: String,
+        severity: Severity,
         service: String,
         message: String,
         location: String,
     }
 
-    let messages: Vec<_> = messages
+    let records: Vec<_> = records
         .into_iter()
         .map(|(_id, record)| Record {
             time: record.time.map(|time| time.to_string()).unwrap_or_default(),
-            severity: record.severity.as_str().to_string(),
+            severity: record.severity,
             service: record.service,
             message: record.message,
             location: record
@@ -38,10 +54,28 @@ pub async fn index(db: &State<Database>) -> Result<RawHtml<Template>, Status> {
         })
         .collect();
 
+    let mut records_by_service = BTreeMap::new();
+
+    for record in &records {
+        let entry = records_by_service
+            .entry(record.service.as_str())
+            .or_insert_with(|| ServiceRecords {
+                first_record: record,
+                last_record: record,
+                all_records: vec![],
+                severity: record.severity,
+            });
+
+        entry.last_record = record;
+        entry.all_records.push(record);
+        entry.severity = max(record.severity, entry.severity);
+    }
+
     Ok(RawHtml(Template::render(
         "index",
         context! {
-            messages: &messages
+            records: &records,
+            records_by_service: &records_by_service,
         },
     )))
 }

snitch-web/src/database.rs

@@ -1,9 +1,13 @@
 use std::{
     fmt::Debug,
     marker::PhantomData,
-    sync::atomic::{AtomicU64, Ordering},
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Mutex,
+    },
 };
 
+use chrono::{DateTime, Duration, Utc};
 use eyre::{eyre, WrapErr};
 use redb::ReadableTable;
 use serde::{de::DeserializeOwned, Serialize};
@@ -22,28 +26,6 @@ struct Serialized<T> {
     _type: PhantomData<T>,
 }
 
-pub(crate) fn open(opt: &Opt) -> eyre::Result<Database> {
-    let inner = redb::Database::create(&opt.db)
-        .context(eyre!("Failed to open database at {}", opt.db.display()))?;
-
-    let txn = inner.begin_write()?;
-    let next_log_id = txn
-        .open_table(table::LOG_MESSAGE)?
-        .range::<u64>(..)?
-        .next_back()
-        .transpose()
-        .ok()
-        .flatten()
-        .map(|(id, _)| id.value() + 1)
-        .unwrap_or_default();
-    txn.commit()?;
-
-    Ok(Database {
-        inner,
-        next_log_id: next_log_id.into(),
-    })
-}
-
 mod table {
     use super::Serialized;
     use redb::TableDefinition;
@@ -53,7 +35,31 @@ mod table {
         TableDefinition::new("log_message");
 }
 
+pub const MAX_MESSAGE_AGE: Duration = Duration::weeks(8);
+
 impl Database {
+    pub(crate) fn open(opt: &Opt) -> eyre::Result<Self> {
+        let inner = redb::Database::create(&opt.db)
+            .context(eyre!("Failed to open database at {}", opt.db.display()))?;
+
+        let txn = inner.begin_write()?;
+        let next_log_id = txn
+            .open_table(table::LOG_MESSAGE)?
+            .range::<u64>(..)?
+            .next_back()
+            .transpose()
+            .ok()
+            .flatten()
+            .map(|(id, _)| id.value() + 1)
+            .unwrap_or_default();
+        txn.commit()?;
+
+        Ok(Database {
+            inner,
+            next_log_id: next_log_id.into(),
+        })
+    }
+
     pub fn write_log(&self, log: &LogMsg) -> eyre::Result<u64> {
         let txn = self.inner.begin_write()?;
         let id = self.next_log_id.fetch_add(1, Ordering::SeqCst);
@@ -77,6 +83,63 @@ impl Database {
             })
             .collect()
     }
+
+    /// Drop any messages older than [MAX_MESSAGE_AGE]
+    pub fn drop_old_messages(&self) -> eyre::Result<usize> {
+        static LAST_RUN: Mutex<DateTime<Utc>> = Mutex::new(DateTime::<Utc>::MIN_UTC);
+
+        let now = Utc::now();
+
+        // Only run this once a day
+        {
+            let mut last_run = LAST_RUN.lock().unwrap();
+            if now > *last_run + Duration::days(1) {
+                *last_run = now;
+            } else {
+                return Ok(0);
+            }
+        }
+
+        log::info!("cleaning up old messages");
+
+        let txn = self.inner.begin_write()?;
+        let mut n = 0;
+        txn.open_table(table::LOG_MESSAGE)?
+            .retain_in(0.., |_id, log_msg| {
+                let keep = match log_msg.deserialize() {
+                    Ok(log_msg) => match log_msg.time {
+                        Some(time) => {
+                            let too_old = now > time + MAX_MESSAGE_AGE;
+                            if too_old {
+                                log::info!("removing old message posted at {time}");
+                            }
+                            !too_old
+                        }
+
+                        // Timestamp should always be set.
+                        // TODO: make it not be an option
+                        None => {
+                            log::warn!("removing message with no timestamp");
+                            false
+                        }
+                    },
+                    Err(_) => {
+                        log::warn!("removing message that failed to deserialize");
+                        false
+                    }
+                };
+
+                if !keep {
+                    n += 1;
+                }
+
+                keep
+            })?;
+
+        txn.commit()?;
+
+        Ok(n)
+    }
 }
 
 impl<T> Serialized<T>
@@ -100,11 +163,13 @@ impl<T> redb::Value for Serialized<T>
 where
     T: Debug + Serialize + DeserializeOwned,
 {
-    type SelfType<'a> = Self
+    type SelfType<'a>
+        = Self
     where
         Self: 'a;
 
-    type AsBytes<'a> = &'a [u8]
+    type AsBytes<'a>
+        = &'a [u8]
     where
         Self: 'a;
 
@@ -134,3 +199,40 @@ where
         redb::TypeName::new(std::any::type_name::<Serialized<T>>())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use chrono::Local;
+
+    use super::*;
+
+    #[test]
+    fn clear_old_messages() {
+        let tmp: u32 = rand::random();
+        let tmp = format!("/tmp/{tmp:08x}.db");
+
+        let opt = Opt { db: tmp.into() };
+        let db = Database::open(&opt).unwrap();
+
+        let now = Local::now();
+        let msg_with_time = |time| LogMsg {
+            time: Some(time),
+            ..LogMsg::new("service", "message", "hostname")
+        };
+
+        // not old mesages
+        db.write_log(&msg_with_time(now)).unwrap();
+        db.write_log(&msg_with_time(now - MAX_MESSAGE_AGE + Duration::days(1)))
+            .unwrap();
+
+        // old messages
+        db.write_log(&msg_with_time(now - MAX_MESSAGE_AGE - Duration::days(1)))
+            .unwrap();
+        db.write_log(&msg_with_time(now - MAX_MESSAGE_AGE - Duration::days(365)))
+            .unwrap();
+        db.write_log(&msg_with_time(now - MAX_MESSAGE_AGE - Duration::seconds(1)))
+            .unwrap();
+
+        assert_eq!(db.drop_old_messages().unwrap(), 3);
+    }
+}

snitch-web/src/main.rs

@@ -8,6 +8,7 @@ mod database;
 use std::path::PathBuf;
 
 use clap::Parser;
+use database::Database;
 use rocket_dyn_templates::Template;
 
 #[derive(Parser)]
@@ -24,7 +25,7 @@ async fn rocket() -> _ {
     let opt = Opt::parse();
     color_eyre::install().unwrap();
 
-    let db = database::open(&opt).expect("open database");
+    let db = Database::open(&opt).expect("open database");
 
     rocket::build()
         .manage(db)

snitch-web/templates/index.hbs

@@ -26,26 +26,66 @@
 			padding: 0.5em;
 			font-family: 'Ubuntu Mono', mono;
 		}
+
+		.service-error-list {
+			max-height: 20em;
+			font-size: small;
+			overflow: scroll;
+			transition: max-height ease-in-out 0.4s;
+		}
+
+		.collapsed {
+			max-height: 0;
+		}
 		</style>
+		<script>
+			function collapse(id) {
+				document.getElementById(id).classList.toggle("collapsed")
+			}
+		</script>
 	</head>
 	<body>
+	{{#if records}}
 	<table>
-	<tr>
+		<tr>
-		<th>Service</th>
+			<th>Severity</th>
-		<th>Severity</th>
+			<th>Service</th>
-		<th>Time</th>
+			<th>Count</th>
-		<th>Message</th>
+			<th>Last message at</th>
-		<th>Location</th>
+		</tr>
-	</tr>
+		{{#each records_by_service}}
-	{{#each messages}}
+		<tr onclick="collapse('{{@key}}-error-list')">
-	<tr>
+			<td>{{this.severity}}</td>
-		<td>{{this.service}}</td>
+			<td>{{@key}}</td>
-		<td>{{this.severity}}</td>
+			<td>{{len this.all_records}}</td>
-		<td>{{this.time}}</td>
+			<td>{{this.last_record.time}}</td>
-		<td>{{this.message}}</td>
+		</tr>
-		<td>{{this.location}}</td>
+		<tr><td colspan="4">
-	</tr>
+		<div
-	{{/each}}
+		    id="{{@key}}-error-list"
+		    class="service-error-list collapsed"
+		>
+			<table>
+				<tr>
+					<th>Severity</th>
+					<th>Message</th>
+					<th>Time</th>
+				</tr>
+				{{#each this.all_records}}
+				<tr>
+					<td>{{this.severity}}</td>
+					<td>{{this.message}}</td>
+					<td>{{this.time}}</td>
+				</tr>
+				{{/each}}
+			</table>
+		</div></td></tr>
+		{{/each}}
 	</table>
+	{{else}}
+	<p>
+	No entries! All is well.
+	</p>
+	{{/if}}
 	</body>
 </html>