From 702f71ed2ef2f97d784fbca4fcc1c7e8954a97c3 Mon Sep 17 00:00:00 2001
From: Nikhil Sinha <nikhil.sinha@cloudsurfex.com>
Date: Fri, 2 Aug 2024 15:53:24 +0530
Subject: [PATCH] fix: ingestion with POST /lostream/{logstream} API

- works well in standalone server but fails in distributed withe error - method not allowed
added the API POST /lostream/{logstream} in the scope of logstream apis for ingestor server
- added a check to return error if stream does not exist
---
 server/src/handlers/http/ingest.rs              | 3 +++
 server/src/handlers/http/modal/ingest_server.rs | 9 +++++++++
 2 files changed, 12 insertions(+)

diff --git a/server/src/handlers/http/ingest.rs b/server/src/handlers/http/ingest.rs
index 0a1e9bdd8..93b40f35d 100644
--- a/server/src/handlers/http/ingest.rs
+++ b/server/src/handlers/http/ingest.rs
@@ -182,6 +182,9 @@ pub async fn post_event(req: HttpRequest, body: Bytes) -> Result<HttpResponse, P
             stream_name
         )));
     }
+    if !STREAM_INFO.stream_exists(&stream_name) {
+        return Err(PostError::StreamNotFound(stream_name));
+    }
     flatten_and_push_logs(req, body, stream_name).await?;
     Ok(HttpResponse::Ok().finish())
 }
diff --git a/server/src/handlers/http/modal/ingest_server.rs b/server/src/handlers/http/modal/ingest_server.rs
index b5f31a4a5..bb71b0983 100644
--- a/server/src/handlers/http/modal/ingest_server.rs
+++ b/server/src/handlers/http/modal/ingest_server.rs
@@ -18,6 +18,7 @@
 use crate::analytics;
 use crate::banner;
 use crate::handlers::airplane;
+use crate::handlers::http::ingest;
 use crate::handlers::http::logstream;
 use crate::handlers::http::middleware::RouteExt;
 use crate::localcache::LocalCacheManager;
@@ -167,15 +168,23 @@ impl IngestServer {
             web::scope("/{logstream}")
                 .service(
                     web::resource("")
+                        // DELETE "/logstream/{logstream}" ==> Delete a log stream
                         .route(
                             web::delete()
                                 .to(logstream::delete)
                                 .authorize_for_stream(Action::DeleteStream),
                         )
+                        // PUT "/logstream/{logstream}" ==> Create a new log stream
                         .route(
                             web::put()
                                 .to(logstream::put_stream)
                                 .authorize_for_stream(Action::CreateStream),
+                        )
+                        // POST "/logstream/{logstream}" ==> Post logs to given log stream
+                        .route(
+                            web::post()
+                                .to(ingest::post_event)
+                                .authorize_for_stream(Action::Ingest),
                         ),
                 )
                 .service(