Impl Ingest Server

add DELETE endpoint for stream on IngestServer
add QuriedStats struct
add stats endpoint for logstream on Ingest Server
add guard to create stream automatically during ingest on Ingest Server
fix: api path /logstream on ingest server
make get_ingestor_info func pub
add logstream api endpoint to put
fix: func dir_with_stream to use the proper stream metadata filename
fix: update stream metadata file name
make stream_json_path give the correct file path when in different server modes

Author: Eshanatnight

server/src/handlers/http.rs

@@ -32,9 +32,6 @@ pub(crate) mod query;
 pub(crate) mod rbac;
 pub(crate) mod role;
 
-// this needs to be removed from here. It is in modal->mod.rs
-// include!(concat!(env!("OUT_DIR"), "/generated.rs"));
-
 pub const MAX_EVENT_PAYLOAD_SIZE: usize = 10485760;
 pub const API_BASE_PATH: &str = "/api";
 pub const API_VERSION: &str = "v1";

server/src/handlers/http/ingest.rs

@@ -32,6 +32,7 @@ use crate::handlers::{
     STREAM_NAME_HEADER_KEY,
 };
 use crate::metadata::STREAM_INFO;
+use crate::option::{Mode, CONFIG};
 use crate::utils::header_parsing::{collect_labelled_headers, ParseHeaderError};
 
 use super::logstream::error::CreateStreamError;
@@ -140,7 +141,18 @@ pub async fn create_stream_if_not_exists(stream_name: &str) -> Result<(), PostEr
     if STREAM_INFO.stream_exists(stream_name) {
         return Ok(());
     }
-    super::logstream::create_stream(stream_name.to_string()).await?;
+
+    match &CONFIG.parseable.mode {
+        Mode::All | Mode::Query => {
+            super::logstream::create_stream(stream_name.to_string()).await?;
+        }
+        Mode::Ingest => {
+            return Err(PostError::Invalid(anyhow::anyhow!(
+                "Stream {} not found. Has it been created?",
+                stream_name
+            )));
+        }
+    }
     Ok(())
 }
 

server/src/handlers/http/logstream.rs

@@ -25,14 +25,16 @@ use serde_json::Value;
 
 use crate::alerts::Alerts;
 use crate::metadata::STREAM_INFO;
-use crate::option::CONFIG;
+use crate::option::{Mode, CONFIG};
 use crate::storage::retention::Retention;
 use crate::storage::{LogStream, StorageDir};
 use crate::{catalog, event, stats};
 use crate::{metadata, validator};
 
 use self::error::{CreateStreamError, StreamError};
 
+use super::modal::query_server::{self, IngestionStats, QueriedStats, StorageStats};
+
 pub async fn delete(req: HttpRequest) -> Result<impl Responder, StreamError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
 
@@ -111,7 +113,6 @@ pub async fn get_alert(req: HttpRequest) -> Result<impl Responder, StreamError>
 
 pub async fn put_stream(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) {
         // Error if the log stream already exists
         return Err(StreamError::Custom {
@@ -121,7 +122,11 @@ pub async fn put_stream(req: HttpRequest) -> Result<impl Responder, StreamError>
             status: StatusCode::BAD_REQUEST,
         });
     }
-    create_stream(stream_name).await?;
+    if CONFIG.parseable.mode == Mode::Query {
+        query_server::QueryServer::sync_streams_with_ingesters(&stream_name).await?;
+    }
+
+    create_stream(stream_name.clone()).await?;
 
     Ok(("log stream created", StatusCode::OK))
 }
@@ -279,30 +284,62 @@ pub async fn get_stats(req: HttpRequest) -> Result<impl Responder, StreamError>
     let stats = stats::get_current_stats(&stream_name, "json")
         .ok_or(StreamError::StreamNotFound(stream_name.clone()))?;
 
+    if CONFIG.parseable.mode == Mode::Query {
+        let stats = query_server::QueryServer::fetch_stats_from_ingesters(&stream_name).await?;
+        let stats = serde_json::to_value(stats).unwrap();
+        return Ok((web::Json(stats), StatusCode::OK));
+    }
+
     let hash_map = STREAM_INFO.read().unwrap();
     let stream_meta = &hash_map
         .get(&stream_name)
         .ok_or(StreamError::StreamNotFound(stream_name.clone()))?;
 
     let time = Utc::now();
+    let qstats = match &stream_meta.first_event_at {
+        Some(first_event_at) => {
+            let ingestion_stats = IngestionStats::new(
+                stats.events,
+                format!("{} {}", stats.ingestion, "Bytes"),
+                "json",
+            );
+            let storage_stats =
+                StorageStats::new(format!("{} {}", stats.storage, "Bytes"), "parquet");
+
+            QueriedStats::new(
+                &stream_name,
+                &stream_meta.created_at,
+                Some(first_event_at.to_owned()),
+                time,
+                ingestion_stats,
+                storage_stats,
+            )
+        }
 
-    let stats = serde_json::json!({
-        "stream": stream_name,
-        "creation_time": &stream_meta.created_at,
-        "first_event_at": Some(&stream_meta.first_event_at),
-        "time": time,
-        "ingestion": {
-            "count": stats.events,
-            "size": format!("{} {}", stats.ingestion, "Bytes"),
-            "format": "json"
-        },
-        "storage": {
-            "size": format!("{} {}", stats.storage, "Bytes"),
-            "format": "parquet"
+        // ? this case should not happen
+        None => {
+            let ingestion_stats = IngestionStats::new(
+                stats.events,
+                format!("{} {}", stats.ingestion, "Bytes"),
+                "json",
+            );
+            let storage_stats =
+                StorageStats::new(format!("{} {}", stats.storage, "Bytes"), "parquet");
+
+            QueriedStats::new(
+                &stream_name,
+                &stream_meta.created_at,
+                Some('0'.to_string()),
+                time,
+                ingestion_stats,
+                storage_stats,
+            )
         }
-    });
+    };
+
+    let out_stats = serde_json::to_value(qstats).unwrap();
 
-    Ok((web::Json(stats), StatusCode::OK))
+    Ok((web::Json(out_stats), StatusCode::OK))
 }
 
 // Check if the first_event_at is empty
@@ -345,7 +382,6 @@ pub async fn create_stream(stream_name: String) -> Result<(), CreateStreamError>
     let created_at = stream_meta.unwrap().created_at;
 
     metadata::STREAM_INFO.add_stream(stream_name.to_string(), created_at);
-
     Ok(())
 }
 

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

@@ -20,6 +20,7 @@ use crate::analytics;
 use crate::banner;
 use crate::handlers::http::logstream;
 use crate::handlers::http::middleware::RouteExt;
+use crate::handlers::http::MAX_EVENT_PAYLOAD_SIZE;
 use crate::localcache::LocalCacheManager;
 use crate::metadata;
 use crate::metrics;
@@ -63,8 +64,8 @@ impl ParseableServer for IngestServer {
         prometheus: PrometheusMetrics,
         _oidc_client: Option<crate::oidc::OpenidConfig>,
     ) -> anyhow::Result<()> {
-        // set the ingestor metadata
-        self.set_ingestor_metadata().await?;
+        // set the ingester metadata
+        self.set_ingester_metadata().await?;
 
         // get the ssl stuff
         let ssl = get_ssl_acceptor(
@@ -122,7 +123,9 @@ impl IngestServer {
         config
             .service(
                 // Base path "{url}/api/v1"
-                web::scope(&base_path()).service(Server::get_ingest_factory()),
+                web::scope(&base_path())
+                    .service(Server::get_ingest_factory())
+                    .service(Self::logstream_api()),
             )
             .service(Server::get_liveness_factory())
             .service(Server::get_readiness_factory())
@@ -151,39 +154,76 @@ impl IngestServer {
     }
 
     #[inline(always)]
-    fn get_ingestor_address(&self) -> SocketAddr {
+    fn get_ingester_address(&self) -> SocketAddr {
         // this might cause an issue down the line
         // best is to make the Cli Struct better, but thats a chore
         (CONFIG.parseable.address.clone())
             .parse::<SocketAddr>()
             .unwrap()
     }
 
-    // create the ingestor metadata and put the .ingestor.json file in the object store
-    async fn set_ingestor_metadata(&self) -> anyhow::Result<()> {
+    fn logstream_api() -> Scope {
+        web::scope("/logstream")
+            .service(
+                // GET "/logstream" ==> Get list of all Log Streams on the server
+                web::resource("")
+                    .route(web::get().to(logstream::list).authorize(Action::ListStream)),
+            )
+            .service(
+                web::scope("/{logstream}")
+                    .service(
+                        web::resource("")
+                            // PUT "/logstream/{logstream}" ==> Create log stream
+                            .route(
+                                web::put()
+                                    .to(logstream::put_stream)
+                                    .authorize_for_stream(Action::CreateStream),
+                            )
+                            // DELETE "/logstream/{logstream}" ==> Delete log stream
+                            .route(
+                                web::delete()
+                                    .to(logstream::delete)
+                                    .authorize_for_stream(Action::DeleteStream),
+                            )
+                            .app_data(web::PayloadConfig::default().limit(MAX_EVENT_PAYLOAD_SIZE)),
+                    )
+                    .service(
+                        // GET "/logstream/{logstream}/stats" ==> Get stats for given log stream
+                        web::resource("/stats").route(
+                            web::get()
+                                .to(logstream::get_stats)
+                                .authorize_for_stream(Action::GetStats),
+                        ),
+                    ),
+            )
+    }
+
+    // create the ingester metadata and put the .ingester.json file in the object store
+    async fn set_ingester_metadata(&self) -> anyhow::Result<()> {
         let store = CONFIG.storage().get_object_store();
 
         // remove ip adn go with the domain name
-        let sock = self.get_ingestor_address();
+        let sock = self.get_ingester_address();
         let path = RelativePathBuf::from(format!(
-            "ingestor.{}.{}.json",
+            "ingester.{}.{}.json",
             sock.ip(), // this might be wrong
             sock.port()
         ));
 
         if store.get_object(&path).await.is_ok() {
-            println!("Ingestor metadata already exists");
+            println!("Ingester metadata already exists");
             return Ok(());
         };
 
+        let scheme = CONFIG.parseable.get_scheme();
         let resource = IngesterMetadata::new(
             sock.port().to_string(),
             CONFIG
                 .parseable
                 .domain_address
                 .clone()
                 .unwrap_or_else(|| {
-                    Url::parse(&format!("http://{}:{}", sock.ip(), sock.port())).unwrap()
+                    Url::parse(&format!("{}://{}:{}", scheme, sock.ip(), sock.port())).unwrap()
                 })
                 .to_string(),
             DEFAULT_VERSION.to_string(),
@@ -203,7 +243,7 @@ impl IngestServer {
     }
 
     // check for querier state. Is it there, or was it there in the past
-    // this should happen before the set the ingestor metadata
+    // this should happen before the set the ingester metadata
     async fn check_querier_state(&self) -> anyhow::Result<(), ObjectStorageError> {
         // how do we check for querier state?
         // based on the work flow of the system, the querier will always need to start first