Skip to content

Metastore static #1704

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nikhilsinhaparseable merged 2 commits into from
Jun 26, 2026
Merged

Metastore static #1704

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6587254
create static for metastore
parmesant Jun 25, 2026
669b6ca
introduce flag for migration
parmesant Jun 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/catalog/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())))?,
)?;
Expand Down Expand Up @@ -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())))?,
)?;
Expand Down
2 changes: 1 addition & 1 deletion src/enterprise/utils.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())))?,
)?;
Expand Down
4 changes: 2 additions & 2 deletions src/handlers/http/logstream.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
4 changes: 2 additions & 2 deletions src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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";
Expand Down
1 change: 1 addition & 0 deletions src/metastore/metastore_traits.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,6 +229,7 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
stream_name: &str,
get_base: bool,
tenant_id: &Option<String>,
is_migration: bool
) -> Result<Bytes, MetastoreError>;
async fn put_stream_json(
&self,
Expand Down
3 changes: 2 additions & 1 deletion src/metastore/metastores/object_store_metastore.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ pub struct ObjectStoreMetastore {
pub storage: Arc<dyn ObjectStorage>,
}

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,
Expand Down Expand Up @@ -870,6 +870,7 @@ impl Metastore for ObjectStoreMetastore {
stream_name: &str,
get_base: bool,
tenant_id: &Option<String>,
_is_migration: bool,
) -> Result<Bytes, MetastoreError> {
let tenant = tenant_id.as_deref().unwrap_or("");
let path = if get_base {
Expand Down
2 changes: 1 addition & 1 deletion src/migration/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -307,7 +307,7 @@ async fn fetch_or_create_stream_metadata(
) -> anyhow::Result<Bytes> {
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)
Expand Down
58 changes: 36 additions & 22 deletions src/parseable/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<Arc<dyn Metastore>> = 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,
Expand Down Expand Up @@ -134,6 +137,34 @@ fn resolve_telemetry_type(

/// Shared state of the Parseable server.
pub static PARSEABLE: Lazy<Parseable> = 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();
Expand All @@ -150,7 +181,6 @@ pub static PARSEABLE: Lazy<Parseable> = Lazy::new(|| {
e.exit();
}
};

match cli.storage {
StorageOptions::Local(args) => {
if args.options.staging_dir() == &args.storage.root {
Expand All @@ -169,62 +199,46 @@ pub static PARSEABLE: Lazy<Parseable> = 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,
)
}
Expand Down
2 changes: 1 addition & 1 deletion src/query/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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?,
)?;

Expand Down
2 changes: 1 addition & 1 deletion src/query/stream_schema_provider.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()))?,
)
Expand Down
22 changes: 11 additions & 11 deletions src/storage/object_storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())))?,
)?;
Expand All @@ -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())))?,
)?;
Expand All @@ -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())))?,
)?;
Expand All @@ -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())))?,
)?;
Expand Down Expand Up @@ -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())))?,
)?;
Expand All @@ -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())))?,
)?;
Expand All @@ -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())))?,
)?;
Expand All @@ -649,15 +649,15 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
) -> Result<ObjectStoreFormat, ObjectStorageError> {
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,
Err(_) => {
// 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())))?;

Expand Down Expand Up @@ -705,7 +705,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
) -> Result<Bytes, ObjectStorageError> {
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 =
Expand Down Expand Up @@ -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);
Expand Down
Loading