review comments incorporated

query server creates internal stream and sync with ingestors on server start
no ingestion can happen from ingestor for internal stream

Author: nikhilsinhaparseable

server/src/handlers.rs

@@ -34,6 +34,7 @@ const STATIC_SCHEMA_FLAG: &str = "x-p-static-schema-flag";
 const AUTHORIZATION_KEY: &str = "authorization";
 const SEPARATOR: char = '^';
 const UPDATE_STREAM_KEY: &str = "x-p-update-stream";
+const STREAM_TYPE_KEY: &str = "x-p-stream-type";
 const OIDC_SCOPE: &str = "openid profile email";
 const COOKIE_AGE_DAYS: usize = 7;
 const SESSION_COOKIE_NAME: &str = "session";

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

@@ -23,13 +23,14 @@ use crate::handlers::http::cluster::utils::{
 };
 use crate::handlers::http::ingest::{ingest_internal_stream, PostError};
 use crate::handlers::http::logstream::error::StreamError;
+use crate::handlers::STREAM_TYPE_KEY;
 use crate::option::CONFIG;
 
 use crate::metrics::prom_utils::Metrics;
 use crate::stats::Stats;
 use crate::storage::object_storage::ingestor_metadata_path;
-use crate::storage::PARSEABLE_ROOT_DIRECTORY;
 use crate::storage::{ObjectStorageError, STREAM_ROOT_DIRECTORY};
+use crate::storage::{StreamType, PARSEABLE_ROOT_DIRECTORY};
 use actix_web::http::header;
 use actix_web::{HttpRequest, Responder};
 use bytes::Bytes;
@@ -145,6 +146,53 @@ pub async fn sync_streams_with_ingestors(
     Ok(())
 }
 
+/// sync internal streams with all ingestors
+pub async fn sync_internal_streams_with_ingestors(stream_name: &str) -> Result<(), StreamError> {
+    let ingestor_infos = get_ingestor_info().await.map_err(|err| {
+        log::error!("Fatal: failed to get ingestor info: {:?}", err);
+        StreamError::Anyhow(err)
+    })?;
+
+    let client = reqwest::Client::new();
+    for ingestor in ingestor_infos.iter() {
+        if !utils::check_liveness(&ingestor.domain_name).await {
+            log::warn!("Ingestor {} is not live", ingestor.domain_name);
+            continue;
+        }
+        let url = format!(
+            "{}{}/logstream/{}",
+            ingestor.domain_name,
+            base_path_without_preceding_slash(),
+            stream_name
+        );
+        let res = client
+            .put(url)
+            .header(header::AUTHORIZATION, &ingestor.token)
+            .header(header::CONTENT_TYPE, "application/json")
+            .header(STREAM_TYPE_KEY, StreamType::Internal.to_string())
+            .send()
+            .await
+            .map_err(|err| {
+                log::error!(
+                    "Fatal: failed to forward upsert stream request to ingestor: {}\n Error: {:?}",
+                    ingestor.domain_name,
+                    err
+                );
+                StreamError::Network(err)
+            })?;
+
+        if !res.status().is_success() {
+            log::error!(
+                "failed to forward upsert stream request to ingestor: {}\nResponse Returned: {:?}",
+                ingestor.domain_name,
+                res.text().await
+            );
+        }
+    }
+
+    Ok(())
+}
+
 pub async fn fetch_daily_stats_from_ingestors(
     stream_name: &str,
     date: &str,
@@ -572,7 +620,6 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
 
 pub fn init_cluster_metrics_schedular() -> Result<(), PostError> {
     log::info!("Setting up schedular for cluster metrics ingestion");
-
     let mut scheduler = AsyncScheduler::new();
     scheduler
         .every(CLUSTER_METRICS_INTERVAL_SECONDS)
@@ -583,11 +630,9 @@ pub fn init_cluster_metrics_schedular() -> Result<(), PostError> {
                     if !metrics.is_empty() {
                         log::info!("Cluster metrics fetched successfully from all ingestors");
                         if let Ok(metrics_bytes) = serde_json::to_vec(&metrics) {
-                            let stream_name = INTERNAL_STREAM_NAME;
-
                             if matches!(
                                 ingest_internal_stream(
-                                    stream_name.to_string(),
+                                    INTERNAL_STREAM_NAME.to_string(),
                                     bytes::Bytes::from(metrics_bytes),
                                 )
                                 .await,

server/src/handlers/http/ingest.rs

@@ -16,8 +16,7 @@
  *
  */
 
-use super::cluster::INTERNAL_STREAM_NAME;
-use super::logstream::error::CreateStreamError;
+use super::logstream::error::{CreateStreamError, StreamError};
 use super::users::dashboards::DashboardError;
 use super::users::filters::FiltersError;
 use super::{kinesis, otel};
@@ -58,14 +57,13 @@ pub async fn ingest(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostE
     {
         let stream_name = stream_name.to_str().unwrap().to_owned();
         let internal_stream_names = STREAM_INFO.list_internal_streams();
-
-        if internal_stream_names.contains(&stream_name) || stream_name == INTERNAL_STREAM_NAME {
+        if internal_stream_names.contains(&stream_name) {
             return Err(PostError::Invalid(anyhow::anyhow!(
                 "The stream {} is reserved for internal use and cannot be ingested into",
                 stream_name
             )));
         }
-        create_stream_if_not_exists(&stream_name, StreamType::UserDefined).await?;
+        create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
 
         flatten_and_push_logs(req, body, stream_name).await?;
         Ok(HttpResponse::Ok().finish())
@@ -75,7 +73,6 @@ pub async fn ingest(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostE
 }
 
 pub async fn ingest_internal_stream(stream_name: String, body: Bytes) -> Result<(), PostError> {
-    create_stream_if_not_exists(&stream_name, StreamType::Internal).await?;
     let size: usize = body.len();
     let parsed_timestamp = Utc::now().naive_utc();
     let (rb, is_first) = {
@@ -119,7 +116,7 @@ pub async fn ingest_otel_logs(req: HttpRequest, body: Bytes) -> Result<HttpRespo
         .find(|&(key, _)| key == STREAM_NAME_HEADER_KEY)
     {
         let stream_name = stream_name.to_str().unwrap().to_owned();
-        create_stream_if_not_exists(&stream_name, StreamType::UserDefined).await?;
+        create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
 
         //flatten logs
         if let Some((_, log_source)) = req.headers().iter().find(|&(key, _)| key == LOG_SOURCE_KEY)
@@ -180,7 +177,7 @@ async fn flatten_and_push_logs(
 pub async fn post_event(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
     let internal_stream_names = STREAM_INFO.list_internal_streams();
-    if internal_stream_names.contains(&stream_name) || stream_name == INTERNAL_STREAM_NAME {
+    if internal_stream_names.contains(&stream_name) {
         return Err(PostError::Invalid(anyhow::anyhow!(
             "Stream {} is an internal stream and cannot be ingested into",
             stream_name
@@ -419,7 +416,7 @@ fn into_event_batch(
 // Check if the stream exists and create a new stream if doesn't exist
 pub async fn create_stream_if_not_exists(
     stream_name: &str,
-    stream_type: StreamType,
+    stream_type: &str,
 ) -> Result<(), PostError> {
     if STREAM_INFO.stream_exists(stream_name) {
         return Ok(());
@@ -494,6 +491,8 @@ pub enum PostError {
     DashboardError(#[from] DashboardError),
     #[error("Error: {0}")]
     CacheError(#[from] CacheError),
+    #[error("Error: {0}")]
+    StreamError(#[from] StreamError),
 }
 
 impl actix_web::ResponseError for PostError {
@@ -515,6 +514,7 @@ impl actix_web::ResponseError for PostError {
             PostError::DashboardError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             PostError::FiltersError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             PostError::CacheError(_) => StatusCode::INTERNAL_SERVER_ERROR,
+            PostError::StreamError(_) => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }
 

server/src/handlers/http/logstream.rs

@@ -20,12 +20,14 @@ use self::error::{CreateStreamError, StreamError};
 use super::base_path_without_preceding_slash;
 use super::cluster::utils::{merge_quried_stats, IngestionStats, QueriedStats, StorageStats};
 use super::cluster::{
-    fetch_daily_stats_from_ingestors, fetch_stats_from_ingestors, sync_streams_with_ingestors,
+    fetch_daily_stats_from_ingestors, fetch_stats_from_ingestors,
+    sync_internal_streams_with_ingestors, sync_streams_with_ingestors, INTERNAL_STREAM_NAME,
 };
+use super::ingest::create_stream_if_not_exists;
 use crate::alerts::Alerts;
 use crate::handlers::{
-    CUSTOM_PARTITION_KEY, STATIC_SCHEMA_FLAG, TIME_PARTITION_KEY, TIME_PARTITION_LIMIT_KEY,
-    UPDATE_STREAM_KEY,
+    CUSTOM_PARTITION_KEY, STATIC_SCHEMA_FLAG, STREAM_TYPE_KEY, TIME_PARTITION_KEY,
+    TIME_PARTITION_LIMIT_KEY, UPDATE_STREAM_KEY,
 };
 use crate::hottier::{HotTierManager, StreamHotTier};
 use crate::metadata::STREAM_INFO;
@@ -180,6 +182,9 @@ pub async fn put_stream(req: HttpRequest, body: Bytes) -> Result<impl Responder,
         create_update_stream(&req, &body, &stream_name).await?;
         sync_streams_with_ingestors(req, body, &stream_name).await?;
     } else {
+        if STREAM_INFO.stream_exists(&stream_name) {
+            return Ok(("Log stream already exists", StatusCode::OK));
+        }
         create_update_stream(&req, &body, &stream_name).await?;
     }
 
@@ -188,12 +193,13 @@ pub async fn put_stream(req: HttpRequest, body: Bytes) -> Result<impl Responder,
 
 fn fetch_headers_from_put_stream_request(
     req: &HttpRequest,
-) -> (String, String, String, String, String) {
+) -> (String, String, String, String, String, String) {
     let mut time_partition = String::default();
     let mut time_partition_limit = String::default();
     let mut custom_partition = String::default();
     let mut static_schema_flag = String::default();
     let mut update_stream = String::default();
+    let mut stream_type = StreamType::UserDefined.to_string();
     req.headers().iter().for_each(|(key, value)| {
         if key == TIME_PARTITION_KEY {
             time_partition = value.to_str().unwrap().to_string();
@@ -210,6 +216,9 @@ fn fetch_headers_from_put_stream_request(
         if key == UPDATE_STREAM_KEY {
             update_stream = value.to_str().unwrap().to_string();
         }
+        if key == STREAM_TYPE_KEY {
+            stream_type = value.to_str().unwrap().to_string();
+        }
     });
 
     (
@@ -218,6 +227,7 @@ fn fetch_headers_from_put_stream_request(
         custom_partition,
         static_schema_flag,
         update_stream,
+        stream_type,
     )
 }
 
@@ -305,8 +315,14 @@ async fn create_update_stream(
     body: &Bytes,
     stream_name: &str,
 ) -> Result<(), StreamError> {
-    let (time_partition, time_partition_limit, custom_partition, static_schema_flag, update_stream) =
-        fetch_headers_from_put_stream_request(req);
+    let (
+        time_partition,
+        time_partition_limit,
+        custom_partition,
+        static_schema_flag,
+        update_stream,
+        stream_type,
+    ) = fetch_headers_from_put_stream_request(req);
 
     if metadata::STREAM_INFO.stream_exists(stream_name) && update_stream != "true" {
         return Err(StreamError::Custom {
@@ -378,7 +394,7 @@ async fn create_update_stream(
         &custom_partition,
         &static_schema_flag,
         schema,
-        StreamType::UserDefined,
+        &stream_type,
     )
     .await?;
 
@@ -647,7 +663,9 @@ 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()))?;
 
-    let ingestor_stats = if CONFIG.parseable.mode == Mode::Query {
+    let ingestor_stats = if CONFIG.parseable.mode == Mode::Query
+        && STREAM_INFO.stream_type(&stream_name).unwrap() == StreamType::UserDefined.to_string()
+    {
         Some(fetch_stats_from_ingestors(&stream_name).await?)
     } else {
         None
@@ -836,11 +854,11 @@ pub async fn create_stream(
     custom_partition: &str,
     static_schema_flag: &str,
     schema: Arc<Schema>,
-    stream_type: StreamType,
+    stream_type: &str,
 ) -> Result<(), CreateStreamError> {
     // fail to proceed if invalid stream name
-    if stream_type != StreamType::Internal {
-        validator::stream_name(&stream_name, &stream_type)?;
+    if stream_type != StreamType::Internal.to_string() {
+        validator::stream_name(&stream_name, stream_type)?;
     }
     // Proceed to create log stream if it doesn't exist
     let storage = CONFIG.storage().get_object_store();
@@ -876,6 +894,7 @@ pub async fn create_stream(
                 custom_partition.to_string(),
                 static_schema_flag.to_string(),
                 static_schema,
+                stream_type,
             );
         }
         Err(err) => {
@@ -1041,6 +1060,16 @@ pub async fn delete_stream_hot_tier(req: HttpRequest) -> Result<impl Responder,
         StatusCode::OK,
     ))
 }
+
+pub async fn create_internal_stream_if_not_exists() -> Result<(), StreamError> {
+    if create_stream_if_not_exists(INTERNAL_STREAM_NAME, &StreamType::Internal.to_string())
+        .await
+        .is_ok()
+    {
+        sync_internal_streams_with_ingestors(INTERNAL_STREAM_NAME).await?;
+    }
+    Ok(())
+}
 #[allow(unused)]
 fn classify_json_error(kind: serde_json::error::Category) -> StatusCode {
     match kind {

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

@@ -18,6 +18,7 @@
 
 use crate::handlers::airplane;
 use crate::handlers::http::cluster::{self, init_cluster_metrics_schedular};
+use crate::handlers::http::logstream::create_internal_stream_if_not_exists;
 use crate::handlers::http::middleware::RouteExt;
 use crate::handlers::http::{base_path, cross_origin_config, API_BASE_PATH, API_VERSION};
 use crate::hottier::HotTierManager;
@@ -174,6 +175,9 @@ impl QueryServer {
 
         migration::run_migration(&CONFIG).await?;
 
+        //create internal stream at server start
+        create_internal_stream_if_not_exists().await?;
+
         FILTERS.load().await?;
         DASHBOARDS.load().await?;
         // track all parquet files already in the data directory