diff --git a/src/catalog/mod.rs b/src/catalog/mod.rs index 11ea747c5..c01681c7a 100644 --- a/src/catalog/mod.rs +++ b/src/catalog/mod.rs @@ -117,7 +117,7 @@ pub async fn update_snapshot( let mut meta: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; @@ -507,7 +507,7 @@ pub async fn remove_manifest_from_snapshot( let mut meta: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; diff --git a/src/enterprise/utils.rs b/src/enterprise/utils.rs index d93c60009..ad633ae0c 100644 --- a/src/enterprise/utils.rs +++ b/src/enterprise/utils.rs @@ -84,7 +84,7 @@ pub async fn fetch_parquet_file_paths( let object_store_format: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream, false, tenant_id) + .get_stream_json(stream, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index 9fc2ab299..f1c6e8f5b 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -466,7 +466,7 @@ pub async fn put_stream_hot_tier( let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(&stream_name, false, &tenant_id) + .get_stream_json(&stream_name, false, &tenant_id, false) .await?, )?; stream_metadata.hot_tier_enabled = true; @@ -572,7 +572,7 @@ pub async fn delete_stream_hot_tier( let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(&stream_name, false, &tenant_id) + .get_stream_json(&stream_name, false, &tenant_id, false) .await?, )?; stream_metadata.hot_tier_enabled = false; diff --git a/src/lib.rs b/src/lib.rs index 0e205cc0c..21030b7b0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,7 +22,7 @@ pub mod analytics; pub mod apikeys; pub mod banner; pub mod catalog; -mod cli; +pub mod cli; #[cfg(feature = "kafka")] pub mod connectors; pub mod correlation; @@ -69,7 +69,7 @@ pub use openid; use parseable::PARSEABLE; use reqwest::{Client, ClientBuilder}; pub use {opentelemetry, opentelemetry_otlp, opentelemetry_proto, opentelemetry_sdk}; -pub use {tracing_actix_web, tracing_opentelemetry, tracing_subscriber}; +pub use {tracing_actix_web, tracing_opentelemetry, tracing_subscriber, clap}; // It is very unlikely that panic will occur when dealing with locks. pub const LOCK_EXPECT: &str = "Thread shouldn't panic while holding a lock"; diff --git a/src/metastore/metastore_traits.rs b/src/metastore/metastore_traits.rs index 0df0e8a5d..4000ccee5 100644 --- a/src/metastore/metastore_traits.rs +++ b/src/metastore/metastore_traits.rs @@ -229,6 +229,7 @@ pub trait Metastore: std::fmt::Debug + Send + Sync { stream_name: &str, get_base: bool, tenant_id: &Option, + is_migration: bool ) -> Result; async fn put_stream_json( &self, diff --git a/src/metastore/metastores/object_store_metastore.rs b/src/metastore/metastores/object_store_metastore.rs index 5ae01d508..475278676 100644 --- a/src/metastore/metastores/object_store_metastore.rs +++ b/src/metastore/metastores/object_store_metastore.rs @@ -65,7 +65,7 @@ pub struct ObjectStoreMetastore { pub storage: Arc, } -fn is_missing_optional_dir(err: &ObjectStorageError) -> bool { +pub fn is_missing_optional_dir(err: &ObjectStorageError) -> bool { match err { ObjectStorageError::NoSuchKey(_) => true, ObjectStorageError::IoError(err) => err.kind() == std::io::ErrorKind::NotFound, @@ -870,6 +870,7 @@ impl Metastore for ObjectStoreMetastore { stream_name: &str, get_base: bool, tenant_id: &Option, + _is_migration: bool, ) -> Result { let tenant = tenant_id.as_deref().unwrap_or(""); let path = if get_base { diff --git a/src/migration/mod.rs b/src/migration/mod.rs index 0a0fd98e5..b0487afe1 100644 --- a/src/migration/mod.rs +++ b/src/migration/mod.rs @@ -307,7 +307,7 @@ async fn fetch_or_create_stream_metadata( ) -> anyhow::Result { if let Ok(stream_metadata) = PARSEABLE .metastore - .get_stream_json(stream, false, tenant_id) + .get_stream_json(stream, false, tenant_id, true) .await { Ok(stream_metadata) diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index b2eef36ea..a46f92984 100644 --- a/src/parseable/mod.rs +++ b/src/parseable/mod.rs @@ -33,7 +33,7 @@ use arrow_schema::{Field, Schema}; use bytes::Bytes; use chrono::Utc; use clap::{Parser, error::ErrorKind}; -use once_cell::sync::Lazy; +use once_cell::sync::{Lazy, OnceCell}; use relative_path::RelativePathBuf; pub use staging::StagingError; use streams::StreamRef; @@ -99,6 +99,9 @@ pub const JOIN_COMMUNITY: &str = "Join us on Parseable Slack community for questions : https://logg.ing/community"; pub const STREAM_EXISTS: &str = "Stream exists"; +/// OnceCell to return metastore +pub static METASTORE: OnceCell> = OnceCell::new(); + /// For OTel log sources the telemetry_type is fully determined by the log_source. /// When the user explicitly sets x-p-telemetry-type and it disagrees with the /// implied type for an otel-* source, reject the request. When they don't set it, @@ -134,6 +137,34 @@ fn resolve_telemetry_type( /// Shared state of the Parseable server. pub static PARSEABLE: Lazy = Lazy::new(|| { + let metastore = METASTORE.get_or_init(|| { + // Prompt user for missing env vars before clap validates them. + // Values are set in env but NOT saved to disk yet. + let collected_envs = crate::interactive::prompt_missing_envs(); + + // Use try_parse so we can avoid persisting bad values on failure. + let cli = match Cli::try_parse() { + Ok(cli) => { + // Clap accepted all values — safe to persist to .parseable.env + crate::interactive::save_collected_envs(&collected_envs); + cli + } + Err(e) => { + // Clap rejected something — don't save, just show the error and exit. + e.exit(); + } + }; + let storage = match &cli.storage { + StorageOptions::Local(args) => args.storage.construct_client(), + StorageOptions::S3(args) => args.storage.construct_client(), + StorageOptions::Blob(args) => args.storage.construct_client(), + StorageOptions::Gcs(args) => args.storage.construct_client(), + }; + tracing::warn!("creating objectstoremetastore"); + let metastore = ObjectStoreMetastore { storage }; + Arc::new(metastore) + }); + // Prompt user for missing env vars before clap validates them. // Values are set in env but NOT saved to disk yet. let collected_envs = crate::interactive::prompt_missing_envs(); @@ -150,7 +181,6 @@ pub static PARSEABLE: Lazy = Lazy::new(|| { e.exit(); } }; - match cli.storage { StorageOptions::Local(args) => { if args.options.staging_dir() == &args.storage.root { @@ -169,62 +199,46 @@ pub static PARSEABLE: Lazy = Lazy::new(|| { .exit(); } - // for now create a metastore without using a CLI arg - let metastore = ObjectStoreMetastore { - storage: args.storage.construct_client(), - }; let hottier_connection_pool = args.storage.construct_client(); Parseable::new( args.options, #[cfg(feature = "kafka")] args.kafka, Arc::new(args.storage), - Arc::new(metastore), + metastore.clone(), hottier_connection_pool, ) } StorageOptions::S3(args) => { - // for now create a metastore without using a CLI arg - let metastore = ObjectStoreMetastore { - storage: args.storage.construct_client(), - }; let hottier_connection_pool = args.storage.construct_client(); Parseable::new( args.options, #[cfg(feature = "kafka")] args.kafka, Arc::new(args.storage), - Arc::new(metastore), + metastore.clone(), hottier_connection_pool, ) } StorageOptions::Blob(args) => { - // for now create a metastore without using a CLI arg - let metastore = ObjectStoreMetastore { - storage: args.storage.construct_client(), - }; let hottier_connection_pool = args.storage.construct_client(); Parseable::new( args.options, #[cfg(feature = "kafka")] args.kafka, Arc::new(args.storage), - Arc::new(metastore), + metastore.clone(), hottier_connection_pool, ) } StorageOptions::Gcs(args) => { - // for now create a metastore without using a CLI arg - let metastore = ObjectStoreMetastore { - storage: args.storage.construct_client(), - }; let hottier_connection_pool = args.storage.construct_client(); Parseable::new( args.options, #[cfg(feature = "kafka")] args.kafka, Arc::new(args.storage), - Arc::new(metastore), + metastore.clone(), hottier_connection_pool, ) } diff --git a/src/query/mod.rs b/src/query/mod.rs index 5e154e5ad..f01f007e7 100644 --- a/src/query/mod.rs +++ b/src/query/mod.rs @@ -758,7 +758,7 @@ pub async fn get_manifest_list( let object_store_format: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await?, )?; diff --git a/src/query/stream_schema_provider.rs b/src/query/stream_schema_provider.rs index 2f4308a32..2a1adba65 100644 --- a/src/query/stream_schema_provider.rs +++ b/src/query/stream_schema_provider.rs @@ -543,7 +543,7 @@ impl TableProvider for StandardTableProvider { let object_store_format: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(&self.stream, false, &self.tenant_id) + .get_stream_json(&self.stream, false, &self.tenant_id, false) .await .map_err(|e| DataFusionError::Plan(e.to_string()))?, ) diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 35f1d7eba..aa3ca9730 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -467,7 +467,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { let mut format: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; @@ -490,7 +490,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { let mut format: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; @@ -513,7 +513,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { let mut format: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; @@ -537,7 +537,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { let mut format: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; @@ -583,7 +583,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { let mut format: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; @@ -606,7 +606,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; @@ -629,7 +629,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice( &PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?, )?; @@ -649,7 +649,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { ) -> Result { let stream_metadata = match PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await { Ok(data) => data, @@ -657,7 +657,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { // get the base stream metadata let bytes = PARSEABLE .metastore - .get_stream_json(stream_name, true, tenant_id) + .get_stream_json(stream_name, true, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?; @@ -705,7 +705,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { ) -> Result { if let Ok(querier_stream_json_bytes) = PARSEABLE .metastore - .get_stream_json(stream_name, true, tenant_id) + .get_stream_json(stream_name, true, tenant_id, false) .await { let querier_stream_metadata = @@ -737,7 +737,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static { if PARSEABLE.get_stream(stream_name, tenant_id).is_ok() { let stream_metadata_bytes = PARSEABLE .metastore - .get_stream_json(stream_name, false, tenant_id) + .get_stream_json(stream_name, false, tenant_id, false) .await .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?; return Ok(stream_metadata_bytes);