feat: separate out ingest and query functionality of the server (#634)

Add P_MODE with options `ingest`, `query` and `all`. Default mode is `all`. 
There are still more changes required for both modes to work well. Will be
added in next subsequent PRs.

Fixes #617

Author: Eshanatnight

server/src/handlers/http.rs

@@ -35,7 +35,7 @@ use rustls_pemfile::{certs, pkcs8_private_keys};
 use crate::option::CONFIG;
 use crate::rbac::role::Action;
 
-use self::middleware::{DisAllowRootUser, RouteExt};
+use self::middleware::{DisAllowRootUser, ModeFilter, RouteExt};
 
 mod about;
 mod health_check;
@@ -76,6 +76,7 @@ pub async fn run_http(
             .wrap(actix_web::middleware::Logger::default())
             .wrap(actix_web::middleware::Compress::default())
             .wrap(cross_origin_config())
+            .wrap(ModeFilter)
     };
 
     let ssl_acceptor = match (
@@ -272,13 +273,16 @@ pub fn configure_routes(
     );
 
     let role_api = web::scope("/role")
+        // GET Role List
         .service(resource("").route(web::get().to(role::list).authorize(Action::ListRole)))
         .service(
+            // PUT and GET Default Role
             resource("/default")
                 .route(web::put().to(role::put_default).authorize(Action::PutRole))
                 .route(web::get().to(role::get_default).authorize(Action::GetRole)),
         )
         .service(
+            // PUT, GET, DELETE Roles
             resource("/{name}")
                 .route(web::put().to(role::put).authorize(Action::PutRole))
                 .route(web::delete().to(role::delete).authorize(Action::DeleteRole))
@@ -299,6 +303,7 @@ pub fn configure_routes(
     cfg.service(
         // Base path "{url}/api/v1"
         web::scope(&base_path())
+            // .wrap(PathFilter)
             // POST "/query" ==> Get results of the SQL query passed in request body
             .service(
                 web::resource("/query")

server/src/handlers/http/middleware.rs

@@ -27,9 +27,12 @@ use actix_web::{
 };
 use futures_util::future::LocalBoxFuture;
 
-use crate::handlers::{
-    AUTHORIZATION_KEY, KINESIS_COMMON_ATTRIBUTES_KEY, LOG_SOURCE_KEY, LOG_SOURCE_KINESIS,
-    STREAM_NAME_HEADER_KEY,
+use crate::{
+    handlers::{
+        AUTHORIZATION_KEY, KINESIS_COMMON_ATTRIBUTES_KEY, LOG_SOURCE_KEY, LOG_SOURCE_KINESIS,
+        STREAM_NAME_HEADER_KEY,
+    },
+    option::Mode,
 };
 use crate::{
     option::CONFIG,
@@ -252,3 +255,98 @@ where
         })
     }
 }
+
+/// ModeFilterMiddleware factory
+pub struct ModeFilter;
+
+/// PathFilterMiddleware needs to implement Service trait
+impl<S, B> Transform<S, ServiceRequest> for ModeFilter
+where
+    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
+    S::Future: 'static,
+    B: 'static,
+{
+    type Response = ServiceResponse<B>;
+    type Error = Error;
+    type InitError = ();
+    type Transform = ModeFilterMiddleware<S>;
+    type Future = Ready<Result<Self::Transform, Self::InitError>>;
+
+    fn new_transform(&self, service: S) -> Self::Future {
+        ready(Ok(ModeFilterMiddleware { service }))
+    }
+}
+
+/// Actual middleware service
+pub struct ModeFilterMiddleware<S> {
+    service: S,
+}
+
+/// Impl the service trait for the middleware service
+impl<S, B> Service<ServiceRequest> for ModeFilterMiddleware<S>
+where
+    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
+    S::Future: 'static,
+    B: 'static,
+{
+    type Response = ServiceResponse<B>;
+    type Error = Error;
+    type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    // impl poll_ready
+    actix_web::dev::forward_ready!(service);
+
+    fn call(&self, req: ServiceRequest) -> Self::Future {
+        let path = req.path();
+        let mode = &CONFIG.parseable.mode;
+
+        // change error messages based on mode
+        match mode {
+            Mode::Query => {
+                let cond = path.split('/').any(|x| x == "ingest");
+                if cond {
+                    Box::pin(async {
+                        Err(actix_web::error::ErrorUnauthorized(
+                            "Ingest API cannot be accessed in Query Mode",
+                        ))
+                    })
+                } else {
+                    let fut = self.service.call(req);
+
+                    Box::pin(async move {
+                        let res = fut.await?;
+                        Ok(res)
+                    })
+                }
+            }
+
+            Mode::Ingest => {
+                let accessable_endpoints = ["ingest", "logstream", "liveness", "readiness"];
+                let cond = path.split('/').any(|x| accessable_endpoints.contains(&x));
+                if !cond {
+                    Box::pin(async {
+                        Err(actix_web::error::ErrorUnauthorized(
+                            "Only Ingestion API can be accessed in Ingest Mode",
+                        ))
+                    })
+                } else {
+                    let fut = self.service.call(req);
+
+                    Box::pin(async move {
+                        let res = fut.await?;
+                        Ok(res)
+                    })
+                }
+            }
+
+            Mode::All => {
+                let fut = self.service.call(req);
+
+                Box::pin(async move {
+                    let res = fut.await?;
+                    Ok(res)
+                })
+            }
+        }
+    }
+}

server/src/option.rs

@@ -251,6 +251,9 @@ pub struct Server {
 
     /// Parquet compression algorithm
     pub parquet_compression: Compression,
+
+    /// Mode of operation
+    pub mode: Mode,
 }
 
 impl FromArgMatches for Server {
@@ -354,6 +357,17 @@ impl FromArgMatches for Server {
             _ => None,
         };
 
+        self.mode = match m
+            .get_one::<String>(Self::MODE)
+            .expect("Mode not set")
+            .as_str()
+        {
+            "query" => Mode::Query,
+            "ingest" => Mode::Ingest,
+            "all" => Mode::All,
+            _ => unreachable!(),
+        };
+
         Ok(())
     }
 }
@@ -382,6 +396,7 @@ impl Server {
     pub const QUERY_MEM_POOL_SIZE: &'static str = "query-mempool-size";
     pub const ROW_GROUP_SIZE: &'static str = "row-group-size";
     pub const PARQUET_COMPRESSION_ALGO: &'static str = "compression-algo";
+    pub const MODE: &'static str = "mode";
     pub const DEFAULT_USERNAME: &'static str = "admin";
     pub const DEFAULT_PASSWORD: &'static str = "admin";
 
@@ -578,6 +593,18 @@ impl Server {
                     .default_value("16384")
                     .value_parser(value_parser!(usize))
                     .help("Number of rows in a row group"),
+            ).arg(
+                Arg::new(Self::MODE)
+                    .long(Self::MODE)
+                    .env("P_MODE")
+                    .value_name("STRING")
+                    .required(false)
+                    .default_value("all")
+                    .value_parser([
+                        "query",
+                        "ingest",
+                        "all"])
+                    .help("Mode of operation"),
             )
             .arg(
                 Arg::new(Self::PARQUET_COMPRESSION_ALGO)
@@ -604,6 +631,14 @@ impl Server {
     }
 }
 
+#[derive(Debug, Default, Eq, PartialEq)]
+pub enum Mode {
+    Query,
+    Ingest,
+    #[default]
+    All,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 #[allow(non_camel_case_types, clippy::upper_case_acronyms)]
 pub enum Compression {