fix: ingestor-querier sync

Below changes are done in this PR -

1. removed querier endpoint and token from parseable.json
2. added migration steps to update parseable.json for latest release
3. removed ingestor to querier sync for ingestion with new stream creation
4. updated logic to list stream from storage
5. updated logic in stream migration at server start
6. updated logic in places where querier/ingestors need to check if stream is created in S3
before confirming the existence of the stream

Author: nikhilsinhaparseable

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

@@ -30,7 +30,6 @@ use crate::metrics::prom_utils::Metrics;
 use crate::rbac::role::model::DefaultPrivilege;
 use crate::rbac::user::User;
 use crate::stats::Stats;
-use crate::storage::get_staging_metadata;
 use crate::storage::object_storage::ingestor_metadata_path;
 use crate::storage::{ObjectStorageError, STREAM_ROOT_DIRECTORY};
 use crate::storage::{ObjectStoreFormat, PARSEABLE_ROOT_DIRECTORY};
@@ -65,7 +64,6 @@ pub async fn sync_streams_with_ingestors(
     headers: HeaderMap,
     body: Bytes,
     stream_name: &str,
-    skip_ingestor: Option<String>,
 ) -> Result<(), StreamError> {
     let mut reqwest_headers = http_header::HeaderMap::new();
 
@@ -79,15 +77,7 @@ pub async fn sync_streams_with_ingestors(
 
     let client = reqwest::Client::new();
 
-    let final_ingestor_infos = match skip_ingestor {
-        None => ingestor_infos,
-        Some(skip_ingestor) => ingestor_infos
-            .into_iter()
-            .filter(|ingestor| ingestor.domain_name != to_url_string(skip_ingestor.clone()))
-            .collect::<Vec<IngestorMetadata>>(),
-    };
-
-    for ingestor in final_ingestor_infos {
+    for ingestor in ingestor_infos {
         if !utils::check_liveness(&ingestor.domain_name).await {
             log::warn!("Ingestor {} is not live", ingestor.domain_name);
             continue;
@@ -852,62 +842,3 @@ pub fn init_cluster_metrics_schedular() -> Result<(), PostError> {
 
     Ok(())
 }
-
-pub async fn forward_create_stream_request(stream_name: &str) -> Result<(), StreamError> {
-    let client = reqwest::Client::new();
-
-    let staging_metadata = get_staging_metadata().unwrap().ok_or_else(|| {
-        StreamError::Anyhow(anyhow::anyhow!("Failed to retrieve staging metadata"))
-    })?;
-    let querier_endpoint = to_url_string(staging_metadata.querier_endpoint.unwrap());
-    let token = staging_metadata.querier_auth_token.unwrap();
-
-    if !check_liveness(&querier_endpoint).await {
-        log::warn!("Querier {} is not live", querier_endpoint);
-        return Err(StreamError::Anyhow(anyhow::anyhow!("Querier is not live")));
-    }
-
-    let url = format!(
-        "{}{}/logstream/{}?skip_ingestors={}",
-        querier_endpoint,
-        base_path_without_preceding_slash(),
-        stream_name,
-        CONFIG.parseable.ingestor_endpoint,
-    );
-
-    let response = client
-        .put(&url)
-        .header(header::AUTHORIZATION, &token)
-        .send()
-        .await
-        .map_err(|err| {
-            log::error!(
-                "Fatal: failed to forward create stream request to querier: {}\n Error: {:?}",
-                &url,
-                err
-            );
-            StreamError::Network(err)
-        })?;
-
-    let status = response.status();
-
-    if !status.is_success() {
-        let response_text = response.text().await.map_err(|err| {
-            log::error!("Failed to read response text from querier: {}", &url);
-            StreamError::Network(err)
-        })?;
-
-        log::error!(
-            "Failed to forward create stream request to querier: {}\nResponse Returned: {:?}",
-            &url,
-            response_text
-        );
-
-        return Err(StreamError::Anyhow(anyhow::anyhow!(
-            "Request failed with status: {}",
-            status,
-        )));
-    }
-
-    Ok(())
-}

server/src/handlers/http/ingest.rs

@@ -26,11 +26,11 @@ use crate::event::{
     error::EventError,
     format::{self, EventFormat},
 };
-use crate::handlers::http::cluster::forward_create_stream_request;
+use crate::handlers::http::modal::utils::logstream_utils::create_stream_and_schema_from_storage;
 use crate::handlers::{LOG_SOURCE_KEY, LOG_SOURCE_OTEL, STREAM_NAME_HEADER_KEY};
 use crate::localcache::CacheError;
 use crate::metadata::error::stream_info::MetadataError;
-use crate::metadata::{self, STREAM_INFO};
+use crate::metadata::STREAM_INFO;
 use crate::option::{Mode, CONFIG};
 use crate::storage::{LogStream, ObjectStorageError, StreamType};
 use crate::utils::header_parsing::ParseHeaderError;
@@ -153,7 +153,7 @@ pub async fn post_event(req: HttpRequest, body: Bytes) -> Result<HttpResponse, P
         )));
     }
     if !STREAM_INFO.stream_exists(&stream_name) {
-        return Err(PostError::StreamNotFound(stream_name));
+        create_stream_and_schema_from_storage(&stream_name).await?;
     }
 
     flatten_and_push_logs(req, body, stream_name).await?;
@@ -190,49 +190,29 @@ pub async fn create_stream_if_not_exists(
         stream_exists = true;
         return Ok(stream_exists);
     }
-    match &CONFIG.parseable.mode {
-        Mode::All | Mode::Query => {
-            super::logstream::create_stream(
-                stream_name.to_string(),
-                "",
-                "",
-                "",
-                "",
-                Arc::new(Schema::empty()),
-                stream_type,
-            )
-            .await?;
-        }
-        Mode::Ingest => {
-            // here the ingest server has not found the stream
-            // so it should check if the stream exists in storage
-            let store = CONFIG.storage().get_object_store();
-            let streams = store.list_streams().await?;
-            if !streams.contains(&LogStream {
-                name: stream_name.to_owned(),
-            }) {
-                match forward_create_stream_request(stream_name).await {
-                    Ok(()) => log::info!("Stream {} created", stream_name),
-                    Err(e) => {
-                        return Err(PostError::Invalid(anyhow::anyhow!(
-                            "Unable to create stream: {} using query server. Error: {}",
-                            stream_name,
-                            e.to_string(),
-                        )))
-                    }
-                };
-            }
-            metadata::STREAM_INFO
-                .upsert_stream_info(
-                    &*store,
-                    LogStream {
-                        name: stream_name.to_owned(),
-                    },
-                )
-                .await
-                .map_err(|_| PostError::StreamNotFound(stream_name.to_owned()))?;
+
+    if CONFIG.parseable.mode != Mode::All {
+        let store = CONFIG.storage().get_object_store();
+        let streams = store.list_streams().await?;
+        if streams.contains(&LogStream {
+            name: stream_name.to_owned(),
+        }) {
+            create_stream_and_schema_from_storage(stream_name).await?;
+            return Ok(stream_exists);
         }
     }
+
+    super::logstream::create_stream(
+        stream_name.to_string(),
+        "",
+        "",
+        "",
+        "",
+        Arc::new(Schema::empty()),
+        stream_type,
+    )
+    .await?;
+
     Ok(stream_exists)
 }
 

server/src/handlers/http/logstream.rs

@@ -20,17 +20,19 @@ use self::error::{CreateStreamError, StreamError};
 use super::cluster::utils::{merge_quried_stats, IngestionStats, QueriedStats, StorageStats};
 use super::cluster::{sync_streams_with_ingestors, INTERNAL_STREAM_NAME};
 use super::ingest::create_stream_if_not_exists;
-use super::modal::utils::logstream_utils::create_update_stream;
+use super::modal::utils::logstream_utils::{
+    create_stream_and_schema_from_storage, create_update_stream,
+};
 use crate::alerts::Alerts;
 use crate::event::format::update_data_type_to_datetime;
 use crate::handlers::STREAM_TYPE_KEY;
 use crate::hottier::{HotTierManager, StreamHotTier, CURRENT_HOT_TIER_VERSION};
 use crate::metadata::STREAM_INFO;
 use crate::metrics::{EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE};
-use crate::option::CONFIG;
+use crate::option::{Mode, CONFIG};
 use crate::stats::{event_labels_date, storage_size_labels_date, Stats};
 use crate::storage::StreamType;
-use crate::storage::{retention::Retention, LogStream, StorageDir, StreamInfo};
+use crate::storage::{retention::Retention, StorageDir, StreamInfo};
 use crate::{catalog, event, stats};
 
 use crate::{metadata, validator};
@@ -82,11 +84,12 @@ pub async fn delete(req: HttpRequest) -> Result<impl Responder, StreamError> {
 }
 
 pub async fn list(_: HttpRequest) -> impl Responder {
-    let res: Vec<LogStream> = STREAM_INFO
+    let res = CONFIG
+        .storage()
+        .get_object_store()
         .list_streams()
-        .into_iter()
-        .map(|stream| LogStream { name: stream })
-        .collect();
+        .await
+        .unwrap();
 
     web::Json(res)
 }
@@ -180,7 +183,11 @@ pub async fn put_alert(
     validator::alert(&alerts)?;
 
     if !STREAM_INFO.stream_initialized(&stream_name)? {
-        return Err(StreamError::UninitializedLogstream);
+        if CONFIG.parseable.mode == Mode::Query {
+            create_stream_and_schema_from_storage(&stream_name).await?;
+        } else {
+            return Err(StreamError::UninitializedLogstream);
+        }
     }
 
     let schema = STREAM_INFO.schema(&stream_name)?;
@@ -218,7 +225,11 @@ pub async fn put_alert(
 pub async fn get_retention(req: HttpRequest) -> Result<impl Responder, StreamError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
     if !STREAM_INFO.stream_exists(&stream_name) {
-        return Err(StreamError::StreamNotFound(stream_name.to_string()));
+        if CONFIG.parseable.mode == Mode::Query {
+            create_stream_and_schema_from_storage(&stream_name).await?;
+        } else {
+            return Err(StreamError::StreamNotFound(stream_name));
+        }
     }
     let retention = STREAM_INFO.get_retention(&stream_name);
 
@@ -328,7 +339,11 @@ pub async fn get_stats(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) {
-        return Err(StreamError::StreamNotFound(stream_name));
+        if CONFIG.parseable.mode == Mode::Query {
+            create_stream_and_schema_from_storage(&stream_name).await?;
+        } else {
+            return Err(StreamError::StreamNotFound(stream_name));
+        }
     }
 
     let query_string = req.query_string();
@@ -497,7 +512,11 @@ pub async fn create_stream(
 pub async fn get_stream_info(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) {
-        return Err(StreamError::StreamNotFound(stream_name));
+        if CONFIG.parseable.mode == Mode::Query {
+            create_stream_and_schema_from_storage(&stream_name).await?;
+        } else {
+            return Err(StreamError::StreamNotFound(stream_name));
+        }
     }
 
     let store = CONFIG.storage().get_object_store();
@@ -540,7 +559,11 @@ pub async fn put_stream_hot_tier(
 ) -> Result<impl Responder, StreamError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
     if !metadata::STREAM_INFO.stream_exists(&stream_name) {
-        return Err(StreamError::StreamNotFound(stream_name));
+        if CONFIG.parseable.mode == Mode::Query {
+            create_stream_and_schema_from_storage(&stream_name).await?;
+        } else {
+            return Err(StreamError::StreamNotFound(stream_name));
+        }
     }
 
     if STREAM_INFO.stream_type(&stream_name).unwrap() == Some(StreamType::Internal.to_string()) {
@@ -590,7 +613,11 @@ pub async fn get_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, Str
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
 
     if !metadata::STREAM_INFO.stream_exists(&stream_name) {
-        return Err(StreamError::StreamNotFound(stream_name));
+        if CONFIG.parseable.mode == Mode::Query {
+            create_stream_and_schema_from_storage(&stream_name).await?;
+        } else {
+            return Err(StreamError::StreamNotFound(stream_name));
+        }
     }
 
     if CONFIG.parseable.hot_tier_storage_path.is_none() {
@@ -615,7 +642,11 @@ pub async fn delete_stream_hot_tier(req: HttpRequest) -> Result<impl Responder,
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
 
     if !metadata::STREAM_INFO.stream_exists(&stream_name) {
-        return Err(StreamError::StreamNotFound(stream_name));
+        if CONFIG.parseable.mode == Mode::Query {
+            create_stream_and_schema_from_storage(&stream_name).await?;
+        } else {
+            return Err(StreamError::StreamNotFound(stream_name));
+        }
     }
 
     if CONFIG.parseable.hot_tier_storage_path.is_none() {
@@ -654,7 +685,7 @@ pub async fn create_internal_stream_if_not_exists() -> Result<(), StreamError> {
             header::CONTENT_TYPE,
             HeaderValue::from_static("application/json"),
         );
-        sync_streams_with_ingestors(header_map, Bytes::new(), INTERNAL_STREAM_NAME, None).await?;
+        sync_streams_with_ingestors(header_map, Bytes::new(), INTERNAL_STREAM_NAME).await?;
     }
     Ok(())
 }

server/src/handlers/http/modal/ingest/ingester_logstream.rs

@@ -7,7 +7,10 @@ use crate::{
     catalog::remove_manifest_from_snapshot,
     event,
     handlers::http::{
-        logstream::error::StreamError, modal::utils::logstream_utils::create_update_stream,
+        logstream::error::StreamError,
+        modal::utils::logstream_utils::{
+            create_stream_and_schema_from_storage, create_update_stream,
+        },
     },
     metadata::{self, STREAM_INFO},
     option::CONFIG,
@@ -22,8 +25,7 @@ pub async fn retention_cleanup(
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
     let storage = CONFIG.storage().get_object_store();
     if !metadata::STREAM_INFO.stream_exists(&stream_name) {
-        log::error!("Stream {} not found", stream_name.clone());
-        return Err(StreamError::StreamNotFound(stream_name.clone()));
+        create_stream_and_schema_from_storage(&stream_name).await?;
     }
     let date_list: Vec<String> = serde_json::from_slice(&body).unwrap();
     let res = remove_manifest_from_snapshot(storage.clone(), &stream_name, date_list).await;
@@ -40,7 +42,7 @@ pub async fn retention_cleanup(
 pub async fn delete(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) {
-        return Err(StreamError::StreamNotFound(stream_name));
+        create_stream_and_schema_from_storage(&stream_name).await?;
     }
 
     metadata::STREAM_INFO.delete_stream(&stream_name);