Fix logstream listing and deletion (#72)

- Delete local data when stream is deleted.
- Use `/.schema` as prefix when listing streams.
- As s3 sync is an independent thread there is 
no guarantee of ordering between when stream 
list is read and when it is deleted. However when 
a stream is deleted, its `.schema` is gone. So 
by checking for `.schema` valid streams can be listed.

Fixes #54 .

Author: trueleo

server/src/handlers/logstream.rs

@@ -16,13 +16,15 @@
  *
  */
 
+use std::fs;
+
 use actix_web::http::StatusCode;
 use actix_web::{web, HttpRequest, HttpResponse, Responder};
 
 use crate::alerts::Alerts;
 use crate::response;
 use crate::s3::S3;
-use crate::storage::ObjectStorage;
+use crate::storage::{ObjectStorage, StorageDir};
 use crate::{metadata, validator};
 
 pub async fn delete(req: HttpRequest) -> HttpResponse {
@@ -57,15 +59,21 @@ pub async fn delete(req: HttpRequest) -> HttpResponse {
         .to_http();
     }
 
+    let stream_dir = StorageDir::new(&stream_name);
+    if fs::remove_dir_all(&stream_dir.data_path).is_err() {
+        log::warn!(
+            "failed to delete local data for stream {}. Clean {} manually",
+            stream_name,
+            stream_dir.data_path.to_string_lossy()
+        )
+    }
+
     if let Err(e) = metadata::STREAM_INFO.delete_stream(&stream_name) {
-        return response::ServerResponse {
-            msg: format!(
-                "failed to delete log stream {} from metadata due to err: {}",
-                stream_name, e
-            ),
-            code: StatusCode::INTERNAL_SERVER_ERROR,
-        }
-        .to_http();
+        log::warn!(
+            "failed to delete log stream {} from metadata due to err: {}",
+            stream_name,
+            e
+        )
     }
 
     response::ServerResponse {

server/src/s3.rs

@@ -297,7 +297,7 @@ impl S3 {
             .client
             .list_objects_v2()
             .bucket(&S3_CONFIG.s3_bucket_name)
-            .delimiter('/')
+            .delimiter("/.schema")
             .send()
             .await?;
 
@@ -307,7 +307,7 @@ impl S3 {
         let logstreams: Vec<_> = common_prefixes
             .iter()
             .filter_map(CommonPrefix::prefix)
-            .filter_map(|name| name.strip_suffix('/'))
+            .filter_map(|name| name.strip_suffix("/.schema"))
             .map(String::from)
             .map(|name| LogStream { name })
             .collect();

server/src/storage.rs

@@ -76,7 +76,7 @@ pub trait ObjectStorage: Sync + 'static {
 
         // entries here means all the streams present on local disk
         for stream in streams {
-            let sync = StorageSync::new(stream.clone());
+            let sync = StorageSync::new(&stream);
 
             // if data.parquet file not present, skip this stream
             if !sync.dir.parquet_path_exists() {
@@ -114,7 +114,7 @@ pub trait ObjectStorage: Sync + 'static {
         let streams = STREAM_INFO.list_streams();
 
         for stream in streams {
-            let dir = StorageDir::new(stream.clone());
+            let dir = StorageDir::new(&stream);
 
             for file in WalkDir::new(dir.temp_dir)
                 .min_depth(1)
@@ -148,14 +148,14 @@ pub struct LogStream {
 }
 
 #[derive(Debug)]
-struct StorageDir {
+pub struct StorageDir {
     pub data_path: PathBuf,
     pub temp_dir: PathBuf,
 }
 
 impl StorageDir {
-    fn new(stream_name: String) -> Self {
-        let data_path = CONFIG.parseable.local_stream_data_path(&stream_name);
+    pub fn new(stream_name: &str) -> Self {
+        let data_path = CONFIG.parseable.local_stream_data_path(stream_name);
         let temp_dir = data_path.join("tmp");
 
         Self {
@@ -186,7 +186,7 @@ struct StorageSync {
 }
 
 impl StorageSync {
-    fn new(stream_name: String) -> Self {
+    fn new(stream_name: &str) -> Self {
         let dir = StorageDir::new(stream_name);
         let time = Utc::now();
         Self { dir, time }