Return 404 in get_stats if no metrics are present for the logstream. Deleting a logstream also deletes its corresponding stats. (#412)

Author: davidrhp

server/src/handlers/http/logstream.rs

@@ -46,6 +46,9 @@ pub async fn delete(req: HttpRequest) -> Result<impl Responder, StreamError> {
     objectstore.delete_stream(&stream_name).await?;
     metadata::STREAM_INFO.delete_stream(&stream_name);
     event::STREAM_WRITERS.delete_stream(&stream_name);
+    stats::delete_stats(&stream_name, "json").unwrap_or_else(|e| {
+        log::warn!("failed to delete stats for stream {}: {:?}", stream_name, e)
+    });
 
     let stream_dir = StorageDir::new(&stream_name);
     if fs::remove_dir_all(&stream_dir.data_path).is_err() {
@@ -363,3 +366,30 @@ pub mod error {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::handlers::http::logstream::error::StreamError;
+    use crate::handlers::http::logstream::get_stats;
+    use actix_web::test::TestRequest;
+    use anyhow::bail;
+
+    #[actix_web::test]
+    #[should_panic]
+    async fn get_stats_panics_without_logstream() {
+        let req = TestRequest::default().to_http_request();
+        let _ = get_stats(req).await;
+    }
+
+    #[actix_web::test]
+    async fn get_stats_stream_not_found_error_for_unknown_logstream() -> anyhow::Result<()> {
+        let req = TestRequest::default()
+            .param("logstream", "test")
+            .to_http_request();
+
+        match get_stats(req).await {
+            Err(StreamError::StreamNotFound(_)) => Ok(()),
+            _ => bail!("expected StreamNotFound error"),
+        }
+    }
+}

server/src/stats.rs

@@ -27,25 +27,62 @@ pub struct Stats {
 }
 
 pub fn get_current_stats(stream_name: &str, format: &'static str) -> Option<Stats> {
+    let event_labels = event_labels(stream_name, format);
+    let storage_size_labels = storage_size_labels(stream_name);
+
     let events_ingested = EVENTS_INGESTED
-        .get_metric_with_label_values(&[stream_name, format])
+        .get_metric_with_label_values(&event_labels)
         .ok()?
         .get();
     let ingestion_size = EVENTS_INGESTED_SIZE
-        .get_metric_with_label_values(&[stream_name, format])
+        .get_metric_with_label_values(&event_labels)
         .ok()?
         .get();
     let storage_size = STORAGE_SIZE
-        .get_metric_with_label_values(&["data", stream_name, "parquet"])
+        .get_metric_with_label_values(&storage_size_labels)
         .ok()?
         .get();
     // this should be valid for all cases given that gauge must never go negative
     let ingestion_size = ingestion_size as u64;
     let storage_size = storage_size as u64;
 
-    Some(Stats {
-        events: events_ingested,
-        ingestion: ingestion_size,
-        storage: storage_size,
-    })
+    // If all metrics are 0, it is presumed that 'stream_name' does not exist and therefore has no stats
+    if events_ingested == 0 && ingestion_size == 0 && storage_size == 0 {
+        None
+    } else {
+        Some(Stats {
+            events: events_ingested,
+            ingestion: ingestion_size,
+            storage: storage_size,
+        })
+    }
+}
+
+pub fn delete_stats(stream_name: &str, format: &'static str) -> prometheus::Result<()> {
+    let event_labels = event_labels(stream_name, format);
+    let storage_size_labels = storage_size_labels(stream_name);
+
+    EVENTS_INGESTED.remove_label_values(&event_labels)?;
+    EVENTS_INGESTED_SIZE.remove_label_values(&event_labels)?;
+    STORAGE_SIZE.remove_label_values(&storage_size_labels)?;
+
+    Ok(())
+}
+
+fn event_labels<'a>(stream_name: &'a str, format: &'static str) -> [&'a str; 2] {
+    [stream_name, format]
+}
+
+fn storage_size_labels(stream_name: &str) -> [&str; 3] {
+    ["data", stream_name, "parquet"]
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::stats::get_current_stats;
+
+    #[test]
+    fn get_current_stats_returns_none_without_data() {
+        assert!(get_current_stats("test", "json").is_none())
+    }
 }