Ingestor querier sync (#1003)

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

---------

Co-authored-by: Devdutt Shenoi <[email protected]>

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,13 +26,13 @@ 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::storage::{ObjectStorageError, StreamType};
 use crate::utils::header_parsing::ParseHeaderError;
 use actix_web::{http::header::ContentType, HttpRequest, HttpResponse};
 use arrow_array::RecordBatch;
@@ -153,7 +153,17 @@ 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));
+        // For distributed deployments, if the stream not found in memory map,
+        //check if it exists in the storage
+        //create stream and schema from storage
+        if CONFIG.parseable.mode != Mode::All {
+            match create_stream_and_schema_from_storage(&stream_name).await {
+                Ok(true) => {}
+                Ok(false) | Err(_) => return Err(PostError::StreamNotFound(stream_name.clone())),
+            }
+        } else {
+            return Err(PostError::StreamNotFound(stream_name.clone()));
+        }
     }
 
     flatten_and_push_logs(req, body, stream_name).await?;
@@ -190,49 +200,25 @@ 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()))?;
-        }
+
+    // For distributed deployments, if the stream not found in memory map,
+    //check if it exists in the storage
+    //create stream and schema from storage
+    if CONFIG.parseable.mode != Mode::All {
+        return Ok(create_stream_and_schema_from_storage(stream_name).await?);
     }
+
+    super::logstream::create_stream(
+        stream_name.to_string(),
+        "",
+        "",
+        "",
+        "",
+        Arc::new(Schema::empty()),
+        stream_type,
+    )
+    .await?;
+
     Ok(stream_exists)
 }