fix: add delete stream API in distributed mode (#766)

with this PR, when delete stream is called, querier 
deletes the stream folder from the storage then calls 
delete stream API for each ingestor. Finally, ingestor 
deletes the stream from its local map

Author: nikhilsinhaparseable

server/src/handlers/http/cluster/mod.rs

@@ -96,8 +96,8 @@ pub async fn sync_streams_with_ingestors(
                 stream_name
             );
 
-            // roll back the stream creation
-            send_stream_rollback_request(&url, ingestor.clone()).await?;
+            // delete the stream
+            send_stream_delete_request(&url, ingestor.clone()).await?;
         }
 
         // this might be a bit too much
@@ -188,15 +188,13 @@ async fn send_stream_sync_request(
 }
 
 /// send a rollback request to all ingestors
-#[allow(dead_code)]
-async fn send_stream_rollback_request(
+pub async fn send_stream_delete_request(
     url: &str,
     ingestor: IngestorMetadata,
 ) -> Result<(), StreamError> {
     if !utils::check_liveness(&ingestor.domain_name).await {
         return Ok(());
     }
-
     let client = reqwest::Client::new();
     let resp = client
         .delete(url)
@@ -207,7 +205,7 @@ async fn send_stream_rollback_request(
         .map_err(|err| {
             // log the error and return a custom error
             log::error!(
-                "Fatal: failed to rollback stream creation: {}\n Error: {:?}",
+                "Fatal: failed to delete stream: {}\n Error: {:?}",
                 ingestor.domain_name,
                 err
             );
@@ -218,18 +216,10 @@ async fn send_stream_rollback_request(
     // this could be a bit too much, but we need to be sure it covers all cases
     if !resp.status().is_success() {
         log::error!(
-            "failed to rollback stream creation: {}\nResponse Returned: {:?}",
+            "failed to delete stream: {}\nResponse Returned: {:?}",
             ingestor.domain_name,
             resp
         );
-        return Err(StreamError::Custom {
-            msg: format!(
-                "failed to rollback stream creation: {}\nResponse Returned: {:?}",
-                ingestor.domain_name,
-                resp.text().await.unwrap_or_default()
-            ),
-            status: StatusCode::INTERNAL_SERVER_ERROR,
-        });
     }
 
     Ok(())

server/src/handlers/http/logstream.rs

@@ -26,6 +26,7 @@ use crate::storage::{retention::Retention, LogStream, StorageDir, StreamInfo};
 use crate::{catalog, event, stats};
 use crate::{metadata, validator};
 
+use super::base_path_without_preceding_slash;
 use super::cluster::fetch_stats_from_ingestors;
 use super::cluster::utils::{merge_quried_stats, IngestionStats, QueriedStats, StorageStats};
 use actix_web::http::StatusCode;
@@ -40,28 +41,49 @@ use std::sync::Arc;
 
 pub async fn delete(req: HttpRequest) -> Result<impl Responder, StreamError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
-
     if !metadata::STREAM_INFO.stream_exists(&stream_name) {
         return Err(StreamError::StreamNotFound(stream_name));
     }
+    match CONFIG.parseable.mode {
+        Mode::Query | Mode::All => {
+            let objectstore = CONFIG.storage().get_object_store();
+
+            objectstore.delete_stream(&stream_name).await?;
+            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()
+                )
+            }
+
+            let ingestor_metadata = super::cluster::get_ingestor_info().await.map_err(|err| {
+                log::error!("Fatal: failed to get ingestor info: {:?}", err);
+                StreamError::from(err)
+            })?;
+
+            for ingestor in ingestor_metadata {
+                let url = format!(
+                    "{}{}/logstream/{}",
+                    ingestor.domain_name,
+                    base_path_without_preceding_slash(),
+                    stream_name
+                );
+
+                // delete the stream
+                super::cluster::send_stream_delete_request(&url, ingestor.clone()).await?;
+            }
+        }
+        _ => {}
+    }
 
-    let objectstore = CONFIG.storage().get_object_store();
-    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() {
-        log::warn!(
-            "failed to delete local data for stream {}. Clean {} manually",
-            stream_name,
-            stream_dir.data_path.to_string_lossy()
-        )
-    }
-
     Ok((format!("log stream {stream_name} deleted"), StatusCode::OK))
 }
 

server/src/handlers/http/modal/ingest_server.rs

@@ -157,6 +157,13 @@ impl IngestServer {
     fn logstream_api() -> Scope {
         web::scope("/logstream").service(
             web::scope("/{logstream}")
+                .service(
+                    web::resource("").route(
+                        web::delete()
+                            .to(logstream::delete)
+                            .authorize_for_stream(Action::DeleteStream),
+                    ),
+                )
                 .service(
                     // GET "/logstream/{logstream}/stats" ==> Get stats for given log stream
                     web::resource("/stats").route(