1. removed method created for syncing internal stream

2. modified sync_streams_with_ingestors that can be used for all streams

Author: nikhilsinhaparseable

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

@@ -23,19 +23,18 @@ 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::http::header::{self, HeaderMap};
 use actix_web::{HttpRequest, Responder};
 use bytes::Bytes;
 use chrono::Utc;
-use http::StatusCode;
+use http::{header as http_header, StatusCode};
 use itertools::Itertools;
 use relative_path::RelativePathBuf;
 use serde::de::Error;
@@ -97,57 +96,15 @@ pub async fn sync_cache_with_ingestors(
 
 // forward the request to all ingestors to keep them in sync
 pub async fn sync_streams_with_ingestors(
-    req: HttpRequest,
+    headers: HeaderMap,
     body: Bytes,
     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 mut reqwest_headers = http_header::HeaderMap::new();
 
-    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)
-            .headers(req.headers().into())
-            .header(header::AUTHORIZATION, &ingestor.token)
-            .body(body.clone())
-            .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
-            );
-        }
+    for (key, value) in headers.iter() {
+        reqwest_headers.insert(key.clone(), value.clone());
     }
-
-    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)
@@ -167,9 +124,9 @@ pub async fn sync_internal_streams_with_ingestors(stream_name: &str) -> Result<(
         );
         let res = client
             .put(url)
+            .headers(reqwest_headers.clone())
             .header(header::AUTHORIZATION, &ingestor.token)
-            .header(header::CONTENT_TYPE, "application/json")
-            .header(STREAM_TYPE_KEY, StreamType::Internal.to_string())
+            .body(body.clone())
             .send()
             .await
             .map_err(|err| {
@@ -185,7 +142,7 @@ pub async fn sync_internal_streams_with_ingestors(stream_name: &str) -> Result<(
             log::error!(
                 "failed to forward upsert stream request to ingestor: {}\nResponse Returned: {:?}",
                 ingestor.domain_name,
-                res.text().await
+                res
             );
         }
     }

server/src/handlers/http/logstream.rs

@@ -20,8 +20,8 @@ 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_internal_streams_with_ingestors, sync_streams_with_ingestors, INTERNAL_STREAM_NAME,
+    fetch_daily_stats_from_ingestors, fetch_stats_from_ingestors, sync_streams_with_ingestors,
+    INTERNAL_STREAM_NAME,
 };
 use super::ingest::create_stream_if_not_exists;
 use crate::alerts::Alerts;
@@ -44,16 +44,19 @@ use crate::{
 };
 
 use crate::{metadata, validator};
+use actix_web::http::header::{self, HeaderMap};
 use actix_web::http::StatusCode;
 use actix_web::{web, HttpRequest, Responder};
 use arrow_schema::{Field, Schema};
 use bytes::Bytes;
 use chrono::Utc;
+use http::{HeaderName, HeaderValue};
 use itertools::Itertools;
 use serde_json::Value;
 use std::collections::HashMap;
 use std::fs;
 use std::num::NonZeroU32;
+use std::str::FromStr;
 use std::sync::Arc;
 
 pub async fn delete(req: HttpRequest) -> Result<impl Responder, StreamError> {
@@ -179,12 +182,9 @@ pub async fn put_stream(req: HttpRequest, body: Bytes) -> Result<impl Responder,
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
 
     if CONFIG.parseable.mode == Mode::Query {
-        create_update_stream(&req, &body, &stream_name).await?;
-        sync_streams_with_ingestors(req, body, &stream_name).await?;
+        let headers = create_update_stream(&req, &body, &stream_name).await?;
+        sync_streams_with_ingestors(headers, 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?;
     }
 
@@ -314,7 +314,7 @@ async fn create_update_stream(
     req: &HttpRequest,
     body: &Bytes,
     stream_name: &str,
-) -> Result<(), StreamError> {
+) -> Result<HeaderMap, StreamError> {
     let (
         time_partition,
         time_partition_limit,
@@ -355,17 +355,16 @@ async fn create_update_stream(
             let time_partition_days = validate_time_partition_limit(&time_partition_limit)?;
             update_time_partition_limit_in_stream(stream_name.to_string(), time_partition_days)
                 .await?;
-            return Ok(());
+            return Ok(req.headers().clone());
         }
 
         if !custom_partition.is_empty() {
             validate_custom_partition(&custom_partition)?;
             update_custom_partition_in_stream(stream_name.to_string(), &custom_partition).await?;
-            return Ok(());
         } else {
             update_custom_partition_in_stream(stream_name.to_string(), "").await?;
-            return Ok(());
         }
+        return Ok(req.headers().clone());
     }
     let mut time_partition_in_days = "";
     if !time_partition_limit.is_empty() {
@@ -398,7 +397,7 @@ async fn create_update_stream(
     )
     .await?;
 
-    Ok(())
+    Ok(req.headers().clone())
 }
 pub async fn put_alert(
     req: HttpRequest,
@@ -1066,7 +1065,16 @@ pub async fn create_internal_stream_if_not_exists() -> Result<(), StreamError> {
         .await
         .is_ok()
     {
-        sync_internal_streams_with_ingestors(INTERNAL_STREAM_NAME).await?;
+        let mut header_map = HeaderMap::new();
+        header_map.insert(
+            HeaderName::from_str(STREAM_TYPE_KEY).unwrap(),
+            HeaderValue::from_str(&StreamType::Internal.to_string()).unwrap(),
+        );
+        header_map.insert(
+            header::CONTENT_TYPE,
+            HeaderValue::from_static("application/json"),
+        );
+        sync_streams_with_ingestors(header_map, Bytes::new(), INTERNAL_STREAM_NAME).await?;
     }
     Ok(())
 }