remove pmeta creation (#1822)

Author: nikhilsinhaparseable

src/handlers/http/modal/query_server.rs

@@ -120,9 +120,6 @@ impl ParseableServer for QueryServer {
             .await;
         migration::run_migration(&PARSEABLE).await?;
 
-        // create internal stream at server start
-        // Multi-tenant mode is not allowed in OSS hence no need to have knowledge of tenant id
-        PARSEABLE.create_internal_stream_if_not_exists().await?;
         // load on init
         load_on_init().await?;
         // track all parquet files already in the data directory

src/parseable/mod.rs

@@ -21,14 +21,10 @@ use std::{
     collections::{HashMap, HashSet},
     num::NonZeroU32,
     path::PathBuf,
-    str::FromStr,
     sync::{Arc, LazyLock, RwLock},
 };
 
-use actix_web::http::{
-    StatusCode,
-    header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue},
-};
+use actix_web::http::{StatusCode, header::HeaderMap};
 use arrow_schema::{Field, Schema};
 use bytes::Bytes;
 use chrono::Utc;
@@ -53,12 +49,10 @@ use crate::{
         format::{LogSource, LogSourceEntry},
     },
     handlers::{
-        DatasetTag, STREAM_TYPE_KEY, TelemetryType,
+        DatasetTag, TelemetryType,
         http::{
-            cluster::{PMETA_STREAM_NAME, sync_streams_with_ingestors},
             ingest::PostError,
             logstream::error::{CreateStreamError, StreamError},
-            middleware::{CLUSTER_SECRET, CLUSTER_SECRET_HEADER},
             modal::{
                 ingest_server::INGESTOR_META,
                 utils::{logstream_utils::PutStreamHeaders, rbac_utils::get_metadata},
@@ -570,72 +564,6 @@ impl Parseable {
         Ok(true)
     }
 
-    pub async fn create_internal_stream_if_not_exists(&self) -> Result<(), StreamError> {
-        let log_source_entry = LogSourceEntry::new(LogSource::Pmeta, HashSet::new());
-        let tenants = if let Some(tenants) = PARSEABLE.list_tenants() {
-            tenants.into_iter().map(Some).collect()
-        } else {
-            vec![None]
-        };
-        for tenant_id in tenants {
-            let internal_stream_result = self
-                .create_stream_if_not_exists(
-                    PMETA_STREAM_NAME,
-                    StreamType::Internal,
-                    None,
-                    vec![log_source_entry.clone()],
-                    TelemetryType::Logs,
-                    &tenant_id,
-                    vec![],
-                    vec![],
-                )
-                .await;
-
-            // Check if either stream creation failed
-            if let Err(e) = &internal_stream_result {
-                tracing::error!("Failed to create pmeta stream: {:?}", e);
-            }
-
-            // Check if both streams already existed
-            if matches!(internal_stream_result, Ok(true)) {
-                continue;
-            }
-
-            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"));
-
-            if let Some((_, hash)) = CLUSTER_SECRET.get() {
-                header_map.insert(
-                    HeaderName::from_static(CLUSTER_SECRET_HEADER),
-                    HeaderValue::from_str(hash).unwrap(),
-                );
-                header_map.insert(
-                    HeaderName::from_static("intra-cluster-tenant"),
-                    HeaderValue::from_str(tenant_id.as_deref().unwrap_or(DEFAULT_TENANT)).unwrap(),
-                );
-            }
-
-            // Sync only the streams that were created successfully
-            if matches!(internal_stream_result, Ok(false))
-                && let Err(e) = sync_streams_with_ingestors(
-                    header_map.clone(),
-                    Bytes::new(),
-                    PMETA_STREAM_NAME,
-                    &tenant_id,
-                )
-                .await
-            {
-                tracing::error!("Failed to sync pmeta stream with ingestors: {:?}", e);
-            }
-        }
-
-        Ok(())
-    }
-
     // Check if the stream exists and create a new stream if doesn't exist
     #[allow(clippy::too_many_arguments)]
     pub async fn create_stream_if_not_exists(