associate create_internal_stream_if_not_exists with Parseable

Author: Devdutt Shenoi

src/handlers/http/logstream.rs

@@ -18,10 +18,8 @@
 
 use self::error::StreamError;
 use super::cluster::utils::{merge_quried_stats, IngestionStats, QueriedStats, StorageStats};
-use super::cluster::{sync_streams_with_ingestors, INTERNAL_STREAM_NAME};
 use super::query::update_schema_when_distributed;
-use crate::event::format::{override_data_type, LogSource};
-use crate::handlers::STREAM_TYPE_KEY;
+use crate::event::format::override_data_type;
 use crate::hottier::{HotTierManager, StreamHotTier, CURRENT_HOT_TIER_VERSION};
 use crate::metadata::SchemaVersion;
 use crate::metrics::{EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE};
@@ -35,18 +33,15 @@ use crate::storage::{StreamInfo, StreamType};
 use crate::utils::actix::extract_session_key_from_req;
 use crate::{stats, validator, LOCK_EXPECT};
 
-use actix_web::http::header::{self, HeaderMap};
 use actix_web::http::StatusCode;
 use actix_web::web::{Json, Path};
 use actix_web::{web, HttpRequest, Responder};
 use arrow_json::reader::infer_json_schema_from_iterator;
 use bytes::Bytes;
 use chrono::Utc;
-use http::{HeaderName, HeaderValue};
 use itertools::Itertools;
 use serde_json::{json, Value};
 use std::fs;
-use std::str::FromStr;
 use std::sync::Arc;
 use tracing::warn;
 
@@ -534,27 +529,6 @@ pub async fn delete_stream_hot_tier(
     ))
 }
 
-pub async fn create_internal_stream_if_not_exists() -> Result<(), StreamError> {
-    if let Ok(stream_exists) = PARSEABLE
-        .create_stream_if_not_exists(INTERNAL_STREAM_NAME, StreamType::Internal, LogSource::Pmeta)
-        .await
-    {
-        if stream_exists {
-            return Ok(());
-        }
-        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(())
-}
 #[allow(unused)]
 fn classify_json_error(kind: serde_json::error::Category) -> StatusCode {
     match kind {

src/handlers/http/modal/query_server.rs

@@ -21,7 +21,6 @@ use crate::correlation::CORRELATIONS;
 use crate::handlers::airplane;
 use crate::handlers::http::base_path;
 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::{DisAllowRootUser, RouteExt};
 use crate::handlers::http::{logstream, MAX_EVENT_PAYLOAD_SIZE};
 use crate::handlers::http::{rbac, role};
@@ -100,7 +99,7 @@ impl ParseableServer for QueryServer {
         migration::run_migration(&PARSEABLE).await?;
 
         //create internal stream at server start
-        create_internal_stream_if_not_exists().await?;
+        PARSEABLE.create_internal_stream_if_not_exists().await?;
 
         if let Err(e) = CORRELATIONS.load().await {
             error!("{e}");

src/handlers/mod.rs

@@ -28,7 +28,7 @@ const CUSTOM_PARTITION_KEY: &str = "x-p-custom-partition";
 const STATIC_SCHEMA_FLAG: &str = "x-p-static-schema-flag";
 const AUTHORIZATION_KEY: &str = "authorization";
 const UPDATE_STREAM_KEY: &str = "x-p-update-stream";
-const STREAM_TYPE_KEY: &str = "x-p-stream-type";
+pub 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";

src/parseable/mod.rs

@@ -17,17 +17,17 @@
  *
  */
 
-use std::{collections::HashMap, num::NonZeroU32, path::PathBuf, sync::Arc};
+use std::{collections::HashMap, num::NonZeroU32, path::PathBuf, str::FromStr, sync::Arc};
 
 use actix_web::http::header::HeaderMap;
 use arrow_schema::{Field, Schema};
 use bytes::Bytes;
 use chrono::Local;
 use clap::{error::ErrorKind, Parser};
-use http::StatusCode;
+use http::{header::CONTENT_TYPE, HeaderName, HeaderValue, StatusCode};
 use once_cell::sync::Lazy;
 pub use staging::StagingError;
-use streams::{Stream, StreamRef};
+use streams::StreamRef;
 pub use streams::{StreamNotFound, Streams};
 use tracing::error;
 
@@ -36,10 +36,14 @@ use crate::connectors::kafka::config::KafkaConfig;
 use crate::{
     cli::{Cli, Options, StorageOptions},
     event::format::LogSource,
-    handlers::http::{
-        ingest::PostError,
-        logstream::error::{CreateStreamError, StreamError},
-        modal::{utils::logstream_utils::PutStreamHeaders, IngestorMetadata},
+    handlers::{
+        http::{
+            cluster::{sync_streams_with_ingestors, INTERNAL_STREAM_NAME},
+            ingest::PostError,
+            logstream::error::{CreateStreamError, StreamError},
+            modal::{utils::logstream_utils::PutStreamHeaders, IngestorMetadata},
+        },
+        STREAM_TYPE_KEY,
     },
     metadata::{LogStreamMetadata, SchemaVersion},
     option::Mode,
@@ -341,6 +345,30 @@ impl Parseable {
         Ok(true)
     }
 
+    pub async fn create_internal_stream_if_not_exists(&self) -> Result<(), StreamError> {
+        match self
+            .create_stream_if_not_exists(
+                INTERNAL_STREAM_NAME,
+                StreamType::Internal,
+                LogSource::Pmeta,
+            )
+            .await
+        {
+            Err(_) | Ok(true) => return Ok(()),
+            _ => {}
+        }
+
+        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(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        sync_streams_with_ingestors(header_map, Bytes::new(), INTERNAL_STREAM_NAME).await?;
+
+        Ok(())
+    }
+
     // Check if the stream exists and create a new stream if doesn't exist
     pub async fn create_stream_if_not_exists(
         &self,