update logic of API endpoints after rebase on main

Author: Eshanatnight

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

@@ -23,13 +23,15 @@ use crate::handlers::http::cluster::utils::{
 };
 use crate::handlers::http::ingest::PostError;
 use crate::handlers::http::logstream::error::StreamError;
+use crate::handlers::{STATIC_SCHEMA_FLAG, TIME_PARTITION_KEY};
 use crate::option::CONFIG;
 
 use crate::metrics::prom_utils::Metrics;
 use crate::storage::ObjectStorageError;
 use crate::storage::PARSEABLE_ROOT_DIRECTORY;
 use actix_web::http::header;
 use actix_web::{HttpRequest, Responder};
+use bytes::Bytes;
 use http::StatusCode;
 use itertools::Itertools;
 use relative_path::RelativePathBuf;
@@ -43,7 +45,12 @@ use super::base_path_without_preceding_slash;
 use super::modal::IngesterMetadata;
 
 // forward the request to all ingesters to keep them in sync
-pub async fn sync_streams_with_ingesters(stream_name: &str) -> Result<(), StreamError> {
+pub async fn sync_streams_with_ingesters(
+    stream_name: &str,
+    time_partition: &str,
+    static_schema: &str,
+    schema: Bytes,
+) -> Result<(), StreamError> {
     let ingester_infos = get_ingester_info().await.map_err(|err| {
         log::error!("Fatal: failed to get ingester info: {:?}", err);
         StreamError::Anyhow(err)
@@ -58,7 +65,7 @@ pub async fn sync_streams_with_ingesters(stream_name: &str) -> Result<(), Stream
             stream_name
         );
 
-        match send_stream_sync_request(&url, ingester.clone()).await {
+        match send_stream_sync_request(&url, ingester.clone(), time_partition, static_schema, schema.clone()).await {
             Ok(_) => continue,
             Err(_) => {
                 errored = true;
@@ -144,6 +151,9 @@ pub async fn fetch_stats_from_ingesters(
 async fn send_stream_sync_request(
     url: &str,
     ingester: IngesterMetadata,
+    time_partition: &str,
+    static_schema: &str,
+    schema: Bytes,
 ) -> Result<(), StreamError> {
     if !utils::check_liveness(&ingester.domain_name).await {
         return Ok(());
@@ -153,7 +163,10 @@ async fn send_stream_sync_request(
     let res = client
         .put(url)
         .header(header::CONTENT_TYPE, "application/json")
+        .header(TIME_PARTITION_KEY, time_partition)
+        .header(STATIC_SCHEMA_FLAG, static_schema)
         .header(header::AUTHORIZATION, ingester.token)
+        .body(schema)
         .send()
         .await
         .map_err(|err| {

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

@@ -28,8 +28,6 @@ use url::Url;
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct QueriedStats {
     pub stream: String,
-    pub creation_time: String,
-    pub first_event_at: Option<String>,
     pub time: DateTime<Utc>,
     pub ingestion: IngestionStats,
     pub storage: StorageStats,
@@ -38,16 +36,12 @@ pub struct QueriedStats {
 impl QueriedStats {
     pub fn new(
         stream: &str,
-        creation_time: &str,
-        first_event_at: Option<String>,
         time: DateTime<Utc>,
         ingestion: IngestionStats,
         storage: StorageStats,
     ) -> Self {
         Self {
             stream: stream.to_string(),
-            creation_time: creation_time.to_string(),
-            first_event_at,
             time,
             ingestion,
             storage,
@@ -119,26 +113,26 @@ impl StorageStats {
 
 pub fn merge_quried_stats(stats: Vec<QueriedStats>) -> QueriedStats {
     // get the actual creation time
-    let min_creation_time = stats
-        .iter()
-        .map(|x| x.creation_time.parse::<DateTime<Utc>>().unwrap())
-        .min()
-        .unwrap(); // should never be None
+    // let min_creation_time = stats
+    //     .iter()
+    //     .map(|x| x.creation_time.parse::<DateTime<Utc>>().unwrap())
+    //     .min()
+    //     .unwrap(); // should never be None
 
     // get the stream name
     let stream_name = stats[0].stream.clone();
 
     // get the first event at
-    let min_first_event_at = stats
-        .iter()
-        .map(|x| match x.first_event_at.as_ref() {
+    // let min_first_event_at = stats
+    //     .iter()
+    //     .map(|x| match x.first_event_at.as_ref() {
             // we can directly unwrap here because
             // we are sure that the first_event_at is a valid date
-            Some(fea) => fea.parse::<DateTime<Utc>>().unwrap(),
-            None => Utc::now(), // current time ie the max time
-        })
-        .min()
-        .unwrap(); // should never be None
+    //         Some(fea) => fea.parse::<DateTime<Utc>>().unwrap(),
+    //         None => Utc::now(), // current time ie the max time
+    //     })
+    //     .min()
+    //     .unwrap(); // should never be None
 
     let min_time = stats.iter().map(|x| x.time).min().unwrap_or_else(Utc::now);
 
@@ -179,8 +173,6 @@ pub fn merge_quried_stats(stats: Vec<QueriedStats>) -> QueriedStats {
 
     QueriedStats::new(
         &stream_name,
-        &min_creation_time.to_string(),
-        Some(min_first_event_at.to_string()),
         min_time,
         cumulative_ingestion,
         cumulative_storage,

server/src/handlers/http/ingest.rs

@@ -156,8 +156,13 @@ pub async fn create_stream_if_not_exists(stream_name: &str) -> Result<(), PostEr
     }
     match &CONFIG.parseable.mode {
         Mode::All | Mode::Query => {
-            super::logstream::create_stream(stream_name.to_string(), "", "", Arc::new(Schema::empty()))
-        .await?;
+            super::logstream::create_stream(
+                stream_name.to_string(),
+                "",
+                "",
+                Arc::new(Schema::empty()),
+            )
+            .await?;
         }
         Mode::Ingest => {
             return Err(PostError::Invalid(anyhow::anyhow!(

server/src/handlers/http/logstream.rs

@@ -146,24 +146,18 @@ pub async fn put_stream(req: HttpRequest, body: Bytes) -> Result<impl Responder,
         });
     }
 
-
-    if !body.is_empty() {
-        if static_schema_flag == "true" {
-            let body_str = std::str::from_utf8(&body).unwrap();
-            let static_schema: StaticSchema = serde_json::from_str(body_str).unwrap();
-            let parsed_schema = convert_static_schema_to_arrow_schema(static_schema);
-            if let Ok(parsed_schema) = parsed_schema {
-                schema = parsed_schema;
-            } else {
-                return Err(StreamError::Custom {
-                    msg: format!(
-                        "unable to commit static schema, logstream {stream_name} not created"
-                    ),
-                    status: StatusCode::BAD_REQUEST,
-                });
-            }
+    if !body.is_empty() && static_schema_flag == "true" {
+        let static_schema: StaticSchema = serde_json::from_slice(&body).unwrap();
+        let parsed_schema = convert_static_schema_to_arrow_schema(static_schema);
+        if let Ok(parsed_schema) = parsed_schema {
+            schema = parsed_schema;
+        } else {
+            return Err(StreamError::Custom {
+                msg: format!("unable to commit static schema, logstream {stream_name} not created"),
+                status: StatusCode::BAD_REQUEST,
+            });
         }
-    } else if static_schema_flag == "true" {
+    } else if body.is_empty() && static_schema_flag == "true" {
         return Err(StreamError::Custom {
                     msg: format!(
                         "please provide schema in the request body for static schema logstream {stream_name}"
@@ -172,9 +166,9 @@ pub async fn put_stream(req: HttpRequest, body: Bytes) -> Result<impl Responder,
                 });
     }
 
-    // ! broken
+
     if CONFIG.parseable.mode == Mode::Query {
-        sync_streams_with_ingesters(&stream_name).await?;
+        sync_streams_with_ingesters(&stream_name, time_partition, static_schema_flag, body).await?;
     }
     create_stream(stream_name, time_partition, static_schema_flag, schema).await?;
 
@@ -333,23 +327,8 @@ pub async fn get_stats(req: HttpRequest) -> Result<impl Responder, StreamError>
 
     let time = Utc::now();
 
-    /* // ! broken / update
-    let stats = serde_json::json!({
-        "stream": stream_name,
-        "time": time,
-        "ingestion": {
-            "count": stats.events,
-            "size": format!("{} {}", stats.ingestion, "Bytes"),
-            "format": "json"
-        },
-        "storage": {
-            "size": format!("{} {}", stats.storage, "Bytes"),
-            "format": "parquet"
-        }
-    });
-    */
     let stats = match &stream_meta.first_event_at {
-        Some(first_event_at) => {
+        Some(_) => {
             let ingestion_stats = IngestionStats::new(
                 stats.events,
                 format!("{} {}", stats.ingestion, "Bytes"),
@@ -358,14 +337,7 @@ pub async fn get_stats(req: HttpRequest) -> Result<impl Responder, StreamError>
             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,
-            )
+            QueriedStats::new(&stream_name, time, ingestion_stats, storage_stats)
         }
 
         None => {
@@ -377,14 +349,7 @@ pub async fn get_stats(req: HttpRequest) -> Result<impl Responder, StreamError>
             let storage_stats =
                 StorageStats::new(format!("{} {}", stats.storage, "Bytes"), "parquet");
 
-            QueriedStats::new(
-                &stream_name,
-                &stream_meta.created_at,
-                None,
-                time,
-                ingestion_stats,
-                storage_stats,
-            )
+            QueriedStats::new(&stream_name, time, ingestion_stats, storage_stats)
         }
     };
     let stats = if let Some(mut ingester_stats) = ingester_stats {
@@ -505,6 +470,8 @@ pub async fn get_stream_info(req: HttpRequest) -> Result<impl Responder, StreamE
         static_schema_flag: stream_meta.static_schema_flag.clone(),
     };
 
+    // get the other info from
+
     Ok((web::Json(stream_info), StatusCode::OK))
 }
 

server/src/storage/staging.rs

@@ -170,13 +170,13 @@ impl StorageDir {
             .port();
         let filename = filename.rsplit_once('.').unwrap();
         let filename = format!("{}.{}.{}", filename.0, port, filename.1);
-/*
-        let file_stem = path.file_stem().unwrap().to_str().unwrap();
-        let random_string =
-            rand::distributions::Alphanumeric.sample_string(&mut rand::thread_rng(), 20);
-        let (_, filename) = file_stem.split_once('.').unwrap();
-        let filename_with_random_number = format!("{}.{}.{}", filename, random_number, "arrows");
-*/
+        /*
+                let file_stem = path.file_stem().unwrap().to_str().unwrap();
+                let random_string =
+                    rand::distributions::Alphanumeric.sample_string(&mut rand::thread_rng(), 20);
+                let (_, filename) = file_stem.split_once('.').unwrap();
+                let filename_with_random_number = format!("{}.{}.{}", filename, random_number, "arrows");
+        */
 
         let mut parquet_path = path.to_owned();
         parquet_path.set_file_name(filename);