From 78fd7c50872eec8efd34c1a161e8fd0f1d215094 Mon Sep 17 00:00:00 2001 From: Michael-Keith Bernard Date: Tue, 6 Oct 2026 23:15:21 -0700 Subject: [PATCH] fix: centralize tenant-aware object path construction Object paths were built ad hoc at each call site, and several sites ignored the tenant component: - Retention built `{stream}/{date}` instead of `{tenant}/{stream}/{date}`, so a named tenant's expired data was never deleted while a same-named stream under the default tenant could be deleted instead (cross-tenant data loss). The named tenant's manifests were removed from the snapshot first, leaving the surviving objects unreferenced. - LocalFS/GCS/Azure `list_dates`/`list_hours`/`list_minutes` and `delete_stream` ignored `tenant_id`, so tenant listings were empty or operated on the wrong prefix. Introduce `tenant_prefix`/`stream_prefix` as the single source of truth for tenant path scoping (default tenant = no prefix) and use them from retention, all four backends, and the metadata/path builders (stream data, schema, stream.json, tombstone, filters, alerts, dashboards, manifest partitions, and the base metastore stream.json path). Add regression tests covering `stream_prefix` scoping and tenant-scoped retention deletion. Refs: parseablehq/parseable#1812 --- src/catalog/mod.rs | 4 +- .../metastores/object_store_metastore.rs | 6 +- src/storage/azure_blob.rs | 29 +++-- src/storage/gcs.rs | 29 +++-- src/storage/localfs.rs | 103 ++++++++++++++++-- src/storage/object_storage.rs | 66 ++++++----- src/storage/retention.rs | 6 +- src/storage/s3.rs | 35 ++---- src/users/dashboards.rs | 3 +- 9 files changed, 185 insertions(+), 96 deletions(-) diff --git a/src/catalog/mod.rs b/src/catalog/mod.rs index 98ab48003..8894629cd 100644 --- a/src/catalog/mod.rs +++ b/src/catalog/mod.rs @@ -40,7 +40,7 @@ use crate::{ stats::{event_labels_date, get_current_stats, storage_size_labels_date, update_deleted_stats}, storage::{ ObjectStorage, ObjectStorageError, ObjectStoreFormat, - object_storage::{manifest_segment_matches, own_manifest_file_name}, + object_storage::{manifest_segment_matches, own_manifest_file_name, tenant_prefix}, }, }; pub use manifest::create_from_parquet_file; @@ -568,7 +568,7 @@ pub fn partition_path( upper_bound: DateTime, tenant_id: &Option, ) -> RelativePathBuf { - let root = tenant_id.as_deref().unwrap_or(""); + let root = tenant_prefix(tenant_id); let lower = lower_bound.date_naive().format("%Y-%m-%d").to_string(); let upper = upper_bound.date_naive().format("%Y-%m-%d").to_string(); if lower == upper { diff --git a/src/metastore/metastores/object_store_metastore.rs b/src/metastore/metastores/object_store_metastore.rs index 867544601..63cbca800 100644 --- a/src/metastore/metastores/object_store_metastore.rs +++ b/src/metastore/metastores/object_store_metastore.rs @@ -59,7 +59,7 @@ use crate::{ object_storage::{ alert_json_path, alert_state_json_path, filter_path, manifest_path, mttr_json_path, outbound_http_policy_json_path, parseable_json_path, schema_path, stream_json_path, - to_bytes, + tenant_prefix, to_bytes, }, }, users::filters::{Filter, migrate_v1_v2}, @@ -777,7 +777,7 @@ impl Metastore for ObjectStoreMetastore { tenant_id: &Option, _is_migration: bool, ) -> Result { - let tenant = tenant_id.as_deref().unwrap_or(""); + let tenant = tenant_prefix(tenant_id); let path = if get_base { RelativePathBuf::from_iter([ tenant, @@ -799,7 +799,7 @@ impl Metastore for ObjectStoreMetastore { tenant_id: &Option, _is_migration: bool, ) -> Result, MetastoreError> { - let root = tenant_id.as_deref().unwrap_or(""); + let root = tenant_prefix(tenant_id); let path = RelativePathBuf::from_iter([root, stream_name, STREAM_ROOT_DIRECTORY]); if let Some(mode) = mode { if mode.eq(&Mode::Ingest) { diff --git a/src/storage/azure_blob.rs b/src/storage/azure_blob.rs index 6c0bc776b..24f53a76b 100644 --- a/src/storage/azure_blob.rs +++ b/src/storage/azure_blob.rs @@ -63,7 +63,8 @@ use crate::{ use super::{ CONNECT_TIMEOUT_SECS, DEFAULT_MAX_OBJECT_STORE_REQUESTS, ObjectStorage, ObjectStorageError, ObjectStorageProvider, PARSEABLE_ROOT_DIRECTORY, REQUEST_TIMEOUT_SECS, - STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer, object_storage::parseable_json_path, + STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer, + object_storage::{parseable_json_path, stream_prefix}, partial_path, to_object_store_path, }; use crate::storage::GET_OBJECTS_CONCURRENCY; @@ -459,9 +460,10 @@ impl BlobStore { stream: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { + let prefix = format!("{}/", stream_prefix(stream, tenant_id)); let resp: Result = self .client - .list_with_delimiter(Some(&(stream.into()))) + .list_with_delimiter(Some(&object_store::path::Path::from(prefix.as_str()))) .await; let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); increment_object_store_calls_by_date("LIST", &Utc::now().date_naive().to_string(), tenant); @@ -485,7 +487,7 @@ impl BlobStore { // return prefixes at the root level let dates: Vec<_> = common_prefixes .iter() - .filter_map(|path| path.as_ref().strip_prefix(&format!("{stream}/"))) + .filter_map(|path| path.as_ref().strip_prefix(&prefix)) .map(String::from) .collect(); tracing::Span::current().record("dates", dates.len()); @@ -889,12 +891,8 @@ impl ObjectStorage for BlobStore { stream_name: &str, tenant_id: &Option, ) -> Result<(), ObjectStorageError> { - let prefix = if let Some(tenant) = tenant_id.as_ref() { - &format!("{tenant}/{stream_name}") - } else { - stream_name - }; - self._delete_prefix(prefix, tenant_id).await?; + let prefix = stream_prefix(stream_name, tenant_id); + self._delete_prefix(&prefix, tenant_id).await?; Ok(()) } @@ -998,7 +996,8 @@ impl ObjectStorage for BlobStore { date: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { - let pre = object_store::path::Path::from(format!("{}/{}/", stream_name, date)); + let prefix = format!("{}/{date}/", stream_prefix(stream_name, tenant_id)); + let pre = object_store::path::Path::from(prefix.as_str()); let resp = self.client.list_with_delimiter(Some(&pre)).await?; let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); increment_files_scanned_in_object_store_calls_by_date( @@ -1014,8 +1013,7 @@ impl ObjectStorage for BlobStore { .iter() .filter_map(|path| { let path_str = path.as_ref(); - if let Some(stripped) = path_str.strip_prefix(&format!("{}/{}/", stream_name, date)) - { + if let Some(stripped) = path_str.strip_prefix(&prefix) { // Remove trailing slash if present, otherwise use as is let clean_path = stripped.strip_suffix('/').unwrap_or(stripped); Some(clean_path.to_string()) @@ -1036,7 +1034,8 @@ impl ObjectStorage for BlobStore { hour: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { - let pre = object_store::path::Path::from(format!("{}/{}/{}/", stream_name, date, hour)); + let prefix = format!("{}/{date}/{hour}/", stream_prefix(stream_name, tenant_id)); + let pre = object_store::path::Path::from(prefix.as_str()); let resp = self.client.list_with_delimiter(Some(&pre)).await?; let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); increment_files_scanned_in_object_store_calls_by_date( @@ -1051,9 +1050,7 @@ impl ObjectStorage for BlobStore { .iter() .filter_map(|path| { let path_str = path.as_ref(); - if let Some(stripped) = - path_str.strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour)) - { + if let Some(stripped) = path_str.strip_prefix(&prefix) { // Remove trailing slash if present, otherwise use as is let clean_path = stripped.strip_suffix('/').unwrap_or(stripped); Some(clean_path.to_string()) diff --git a/src/storage/gcs.rs b/src/storage/gcs.rs index 8ac87f763..ecb19414e 100644 --- a/src/storage/gcs.rs +++ b/src/storage/gcs.rs @@ -62,7 +62,8 @@ use tracing::error; use super::{ CONNECT_TIMEOUT_SECS, DEFAULT_MAX_OBJECT_STORE_REQUESTS, ObjectStorage, ObjectStorageError, ObjectStorageProvider, PARSEABLE_ROOT_DIRECTORY, REQUEST_TIMEOUT_SECS, - STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer, object_storage::parseable_json_path, + STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer, + object_storage::{parseable_json_path, stream_prefix}, partial_path, to_object_store_path, }; use crate::storage::GET_OBJECTS_CONCURRENCY; @@ -425,9 +426,10 @@ impl Gcs { stream: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { + let prefix = format!("{}/", stream_prefix(stream, tenant_id)); let resp: Result = self .client - .list_with_delimiter(Some(&(stream.into()))) + .list_with_delimiter(Some(&object_store::path::Path::from(prefix.as_str()))) .await; let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); increment_object_store_calls_by_date("LIST", &Utc::now().date_naive().to_string(), tenant); @@ -451,7 +453,7 @@ impl Gcs { // return prefixes at the root level let dates: Vec<_> = common_prefixes .iter() - .filter_map(|path| path.as_ref().strip_prefix(&format!("{stream}/"))) + .filter_map(|path| path.as_ref().strip_prefix(&prefix)) .map(String::from) .collect(); tracing::Span::current().record("dates", dates.len()); @@ -872,12 +874,8 @@ impl ObjectStorage for Gcs { stream_name: &str, tenant_id: &Option, ) -> Result<(), ObjectStorageError> { - let prefix = if let Some(tenant) = tenant_id.as_ref() { - &format!("{tenant}/{stream_name}") - } else { - stream_name - }; - self._delete_prefix(prefix, tenant_id).await?; + let prefix = stream_prefix(stream_name, tenant_id); + self._delete_prefix(&prefix, tenant_id).await?; Ok(()) } @@ -980,7 +978,8 @@ impl ObjectStorage for Gcs { date: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { - let pre = object_store::path::Path::from(format!("{}/{}/", stream_name, date)); + let prefix = format!("{}/{date}/", stream_prefix(stream_name, tenant_id)); + let pre = object_store::path::Path::from(prefix.as_str()); let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); let resp = self.client.list_with_delimiter(Some(&pre)).await?; increment_files_scanned_in_object_store_calls_by_date( @@ -996,8 +995,7 @@ impl ObjectStorage for Gcs { .iter() .filter_map(|path| { let path_str = path.as_ref(); - if let Some(stripped) = path_str.strip_prefix(&format!("{}/{}/", stream_name, date)) - { + if let Some(stripped) = path_str.strip_prefix(&prefix) { // Remove trailing slash if present, otherwise use as is let clean_path = stripped.strip_suffix('/').unwrap_or(stripped); Some(clean_path.to_string()) @@ -1018,7 +1016,8 @@ impl ObjectStorage for Gcs { hour: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { - let pre = object_store::path::Path::from(format!("{}/{}/{}/", stream_name, date, hour)); + let prefix = format!("{}/{date}/{hour}/", stream_prefix(stream_name, tenant_id)); + let pre = object_store::path::Path::from(prefix.as_str()); let resp = self.client.list_with_delimiter(Some(&pre)).await?; let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); increment_files_scanned_in_object_store_calls_by_date( @@ -1033,9 +1032,7 @@ impl ObjectStorage for Gcs { .iter() .filter_map(|path| { let path_str = path.as_ref(); - if let Some(stripped) = - path_str.strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour)) - { + if let Some(stripped) = path_str.strip_prefix(&prefix) { // Remove trailing slash if present, otherwise use as is let clean_path = stripped.strip_suffix('/').unwrap_or(stripped); Some(clean_path.to_string()) diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index d66714023..933e78aea 100644 --- a/src/storage/localfs.rs +++ b/src/storage/localfs.rs @@ -49,7 +49,7 @@ use crate::{ use super::{ ALERTS_ROOT_DIRECTORY, ObjectStorage, ObjectStorageError, ObjectStorageProvider, PARSEABLE_ROOT_DIRECTORY, STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY, - TOMBSTONE_ROOT_DIRECTORY, + TOMBSTONE_ROOT_DIRECTORY, object_storage::stream_prefix, }; #[derive(Debug, Clone, clap::Args)] @@ -491,7 +491,7 @@ impl ObjectStorage for LocalFS { stream_name: &str, tenant_id: &Option, ) -> Result<(), ObjectStorageError> { - let path = self.root.join(stream_name); + let path = self.root.join(stream_prefix(stream_name, tenant_id)); let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); let result = fs::remove_dir_all(path).await; @@ -683,7 +683,7 @@ impl ObjectStorage for LocalFS { stream_name: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { - let path = self.root.join(stream_name); + let path = self.root.join(stream_prefix(stream_name, tenant_id)); let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); let result = fs::read_dir(&path).await; @@ -717,9 +717,12 @@ impl ObjectStorage for LocalFS { &self, stream_name: &str, date: &str, - _tenant_id: &Option, + tenant_id: &Option, ) -> Result, ObjectStorageError> { - let path = self.root.join(stream_name).join(date); + let path = self + .root + .join(stream_prefix(stream_name, tenant_id)) + .join(date); let directories = ReadDirStream::new(fs::read_dir(&path).await?); let entries: Vec = directories.try_collect().await?; let entries = entries.into_iter().map(dir_name); @@ -736,9 +739,13 @@ impl ObjectStorage for LocalFS { stream_name: &str, date: &str, hour: &str, - _tenant_id: &Option, + tenant_id: &Option, ) -> Result, ObjectStorageError> { - let path = self.root.join(stream_name).join(date).join(hour); + let path = self + .root + .join(stream_prefix(stream_name, tenant_id)) + .join(date) + .join(hour); // Propagate any read_dir errors instead of swallowing them let directories = ReadDirStream::new(fs::read_dir(&path).await?); let entries: Vec = directories.try_collect().await?; @@ -916,3 +923,85 @@ impl From for ObjectStorageError { ObjectStorageError::UnhandledError(Box::new(e)) } } + +#[cfg(test)] +mod tenant_path_tests { + //! Regression tests for tenant-scoped object paths. + //! + //! Regression tests for the centralized tenant-scoped path construction + //! (`stream_prefix`/`tenant_prefix`) and the backend list/delete paths. + //! They cover issues/0001 (retention deletes the wrong prefix) and + //! issues/0002 (list APIs ignore the tenant) at the LocalFS layer, without + //! needing the global `PARSEABLE` server bootstrap. + + use super::*; + + fn make_dir(root: &Path, rel: &str) { + std::fs::create_dir_all(root.join(rel)).unwrap(); + } + + /// `list_dates` must list the requested tenant's prefix, not the bucket root. + #[tokio::test] + async fn list_dates_must_be_tenant_scoped() { + let tmp = temp_dir::TempDir::new().unwrap(); + let root = tmp.path().to_path_buf(); + + // Same-named stream under two tenants, with distinguishable partitions. + make_dir(&root, "t1/logs/date=2024-01-01"); + make_dir(&root, "logs/date=2024-03-03"); + + let store = LocalFS::new(root.clone()); + let dates = store + .list_dates("logs", &Some("t1".to_owned())) + .await + .unwrap(); + + assert_eq!( + dates, + vec!["date=2024-01-01".to_owned()], + "list_dates for tenant t1 returned {dates:?}; it is listing the \ + tenant-less prefix (see issues/0002)" + ); + } + + /// `stream_prefix` is the single source of truth for tenant path scoping. + #[test] + fn stream_prefix_is_tenant_scoped() { + assert_eq!(stream_prefix("logs", &None), "logs"); + assert_eq!(stream_prefix("logs", &Some("t1".to_owned())), "t1/logs"); + // The default-tenant sentinel is treated as "no tenant". + assert_eq!( + stream_prefix("logs", &Some(DEFAULT_TENANT.to_owned())), + "logs" + ); + } + + /// A retention delete for tenant `t1` must delete exactly `t1`'s expired + /// partition and must not touch the default tenant's same-named stream. + #[tokio::test] + async fn retention_delete_is_tenant_scoped() { + let tmp = temp_dir::TempDir::new().unwrap(); + let root = tmp.path().to_path_buf(); + + make_dir(&root, "t1/logs/date=2024-01-01"); + make_dir(&root, "logs/date=2024-01-01"); + + let store = LocalFS::new(root.clone()); + let tenant = Some("t1".to_owned()); + let date = "date=2024-01-01".to_owned(); + + // The path retention now builds (see `src/storage/retention.rs`). + let prefix = stream_prefix("logs", &tenant); + let path = RelativePathBuf::from_iter([prefix.as_str(), &date]); + store.delete_prefix(&path, &tenant).await.unwrap(); + + assert!( + !root.join("t1/logs/date=2024-01-01").exists(), + "tenant t1's expired partition should be deleted" + ); + assert!( + root.join("logs/date=2024-01-01").exists(), + "the default tenant's same-named stream must not be touched" + ); + } +} diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 368326a67..6d54a90ae 100644 --- a/src/storage/object_storage.rs +++ b/src/storage/object_storage.rs @@ -1374,13 +1374,7 @@ fn stream_relative_path( let custom_partition_list = custom_partition_fields.split(',').collect::>(); file_suffix = str::replacen(filename, ".", "/", 3 + custom_partition_list.len()); } - if let Some(tenant) = tenant_id - && !tenant.eq(DEFAULT_TENANT) - { - format!("{tenant}/{stream_name}/{file_suffix}") - } else { - format!("{stream_name}/{file_suffix}") - } + format!("{}/{file_suffix}", stream_prefix(stream_name, tenant_id)) } pub fn sync_all_streams(joinset: &mut JoinSet>) { @@ -1449,6 +1443,30 @@ pub async fn commit_schema_to_storage( .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail()))) } +/// The tenant component of an object path, or `""` for the default tenant. +/// +/// Single source of truth for tenant path scoping: a named tenant is prefixed +/// as `{tenant}/...`, while the default tenant (`None`) has no prefix. Use this +/// (or [`stream_prefix`]) instead of matching on `tenant_id` ad hoc, so +/// list/delete/metadata paths cannot drift apart again. +#[inline] +pub fn tenant_prefix(tenant_id: &Option) -> &str { + match tenant_id.as_deref() { + Some(tenant) if tenant != DEFAULT_TENANT => tenant, + _ => "", + } +} + +/// Tenant-qualified stream prefix (no trailing slash): `{tenant}/{stream}` for +/// a named tenant, `{stream}` for the default tenant. +#[inline] +pub fn stream_prefix(stream_name: &str, tenant_id: &Option) -> String { + match tenant_prefix(tenant_id) { + "" => stream_name.to_owned(), + tenant => format!("{tenant}/{stream_name}"), + } +} + #[inline(always)] pub fn to_bytes(any: &(impl ?Sized + serde::Serialize)) -> Bytes { serde_json::to_vec(any) @@ -1457,7 +1475,7 @@ pub fn to_bytes(any: &(impl ?Sized + serde::Serialize)) -> Bytes { } pub fn schema_path(stream_name: &str, tenant_id: &Option) -> RelativePathBuf { - let tenant = tenant_id.as_deref().unwrap_or(""); + let tenant = tenant_prefix(tenant_id); if PARSEABLE.options.mode == Mode::Ingest { let id = INGESTOR_META .get() @@ -1473,7 +1491,7 @@ pub fn schema_path(stream_name: &str, tenant_id: &Option) -> RelativePat #[inline(always)] pub fn stream_json_path(stream_name: &str, tenant_id: &Option) -> RelativePathBuf { - let tenant = tenant_id.as_deref().unwrap_or(""); + let tenant = tenant_prefix(tenant_id); if PARSEABLE.options.mode == Mode::Ingest { let id = INGESTOR_META .get() @@ -1503,7 +1521,7 @@ pub fn stream_json_path(stream_name: &str, tenant_id: &Option) -> Relati /// scan to find it. #[inline(always)] pub fn tombstone_path(stream_name: &str, tenant_id: &Option) -> RelativePathBuf { - let tenant = tenant_id.as_deref().unwrap_or(""); + let tenant = tenant_prefix(tenant_id); RelativePathBuf::from_iter([ TOMBSTONE_ROOT_DIRECTORY, tenant, @@ -1553,7 +1571,7 @@ pub async fn list_tombstoned_streams( storage: &(impl ObjectStorage + ?Sized), tenant_id: &Option, ) -> Result, ObjectStorageError> { - let tenant = tenant_id.as_deref().unwrap_or(""); + let tenant = tenant_prefix(tenant_id); let root = RelativePathBuf::from_iter([TOMBSTONE_ROOT_DIRECTORY, tenant]); let candidates = storage.list_dirs_relative(&root, tenant_id).await?; @@ -1574,7 +1592,7 @@ pub fn filter_path( filter_file_name: &str, tenant_id: &Option, ) -> RelativePathBuf { - let root = tenant_id.as_deref().unwrap_or(""); + let root = tenant_prefix(tenant_id); RelativePathBuf::from_iter([ root, USERS_ROOT_DIR, @@ -1594,21 +1612,20 @@ pub fn parseable_json_path() -> RelativePathBuf { /// TODO: Needs to be updated for distributed mode #[inline(always)] pub fn alert_json_path(alert_id: Ulid, tenant_id: &Option) -> RelativePathBuf { - if let Some(tenant_id) = tenant_id.as_ref() { - RelativePathBuf::from_iter([ - tenant_id, + match tenant_prefix(tenant_id) { + "" => RelativePathBuf::from_iter([ALERTS_ROOT_DIRECTORY, &format!("{alert_id}.json")]), + tenant => RelativePathBuf::from_iter([ + tenant, ALERTS_ROOT_DIRECTORY, &format!("{alert_id}.json"), - ]) - } else { - RelativePathBuf::from_iter([ALERTS_ROOT_DIRECTORY, &format!("{alert_id}.json")]) + ]), } } /// TODO: Needs to be updated for distributed mode #[inline(always)] pub fn target_json_path(target_id: &Ulid, tenant_id: &Option) -> RelativePathBuf { - let root = tenant_id.as_deref().unwrap_or(""); + let root = tenant_prefix(tenant_id); RelativePathBuf::from_iter([ root, SETTINGS_ROOT_DIRECTORY, @@ -1621,7 +1638,7 @@ pub fn target_json_path(target_id: &Ulid, tenant_id: &Option) -> Relativ /// Format: "/settings/outbound_http_policy.json" #[inline(always)] pub fn outbound_http_policy_json_path(tenant_id: &Option) -> RelativePathBuf { - let root = tenant_id.as_deref().unwrap_or(""); + let root = tenant_prefix(tenant_id); RelativePathBuf::from_iter([root, SETTINGS_ROOT_DIRECTORY, "outbound_http_policy.json"]) } @@ -1629,7 +1646,7 @@ pub fn outbound_http_policy_json_path(tenant_id: &Option) -> RelativePat /// Format: ".alerts/alert_state_{alert_id}.json" #[inline(always)] pub fn alert_state_json_path(alert_id: Ulid, tenant_id: &Option) -> RelativePathBuf { - let root = tenant_id.as_deref().unwrap_or(""); + let root = tenant_prefix(tenant_id); RelativePathBuf::from_iter([ root, ALERTS_ROOT_DIRECTORY, @@ -1641,10 +1658,9 @@ pub fn alert_state_json_path(alert_id: Ulid, tenant_id: &Option) -> Rela /// Format: ".alerts/mttr.json" #[inline(always)] pub fn mttr_json_path(tenant_id: &Option) -> RelativePathBuf { - if let Some(tenant) = tenant_id.as_ref() { - RelativePathBuf::from_iter([tenant, ALERTS_ROOT_DIRECTORY, "mttr.json"]) - } else { - RelativePathBuf::from_iter([ALERTS_ROOT_DIRECTORY, "mttr.json"]) + match tenant_prefix(tenant_id) { + "" => RelativePathBuf::from_iter([ALERTS_ROOT_DIRECTORY, "mttr.json"]), + tenant => RelativePathBuf::from_iter([tenant, ALERTS_ROOT_DIRECTORY, "mttr.json"]), } } diff --git a/src/storage/retention.rs b/src/storage/retention.rs index 64faf95a9..81dc84073 100644 --- a/src/storage/retention.rs +++ b/src/storage/retention.rs @@ -209,6 +209,7 @@ impl From for Vec { mod action { use crate::catalog::remove_manifest_from_snapshot; use crate::parseable::PARSEABLE; + use crate::storage::object_storage::stream_prefix; use crate::tenants::TENANT_METADATA; use chrono::{Days, NaiveDate, Utc}; use futures::{StreamExt, stream::FuturesUnordered}; @@ -251,7 +252,10 @@ mod action { } for date in dates_to_delete { - let path = RelativePathBuf::from_iter([&stream_name, &date]); + // Build the tenant-qualified path: retention must never operate on + // the default tenant's same-named stream (see issues/0001). + let prefix = stream_prefix(&stream_name, tenant_id); + let path = RelativePathBuf::from_iter([prefix.as_str(), &date]); delete_tasks.push(async move { PARSEABLE .storage diff --git a/src/storage/s3.rs b/src/storage/s3.rs index a92923172..c69744326 100644 --- a/src/storage/s3.rs +++ b/src/storage/s3.rs @@ -64,7 +64,8 @@ use crate::{ use super::{ CONNECT_TIMEOUT_SECS, DEFAULT_MAX_OBJECT_STORE_REQUESTS, ObjectStorage, ObjectStorageError, ObjectStorageProvider, PARSEABLE_ROOT_DIRECTORY, REQUEST_TIMEOUT_SECS, - STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer, object_storage::parseable_json_path, + STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer, + object_storage::{parseable_json_path, stream_prefix}, partial_path, to_object_store_path, }; use crate::storage::GET_OBJECTS_CONCURRENCY; @@ -639,11 +640,8 @@ impl S3 { stream: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { - let (prefix, tenant_str) = if let Some(tenant) = tenant_id { - (format!("{}/{}/", tenant, stream), tenant.as_str()) - } else { - (format!("{}/", stream), DEFAULT_TENANT) - }; + let prefix = format!("{}/", stream_prefix(stream, tenant_id)); + let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); let resp: Result = self .client .list_with_delimiter(Some(&(prefix.as_str().into()))) @@ -1125,12 +1123,8 @@ impl ObjectStorage for S3 { stream_name: &str, tenant_id: &Option, ) -> Result<(), ObjectStorageError> { - let prefix = if let Some(tenant) = tenant_id.as_ref() { - &format!("{tenant}/{stream_name}") - } else { - stream_name - }; - self._delete_prefix(prefix, tenant_id).await?; + let prefix = stream_prefix(stream_name, tenant_id); + self._delete_prefix(&prefix, tenant_id).await?; Ok(()) } @@ -1235,11 +1229,8 @@ impl ObjectStorage for S3 { date: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { - let (pre, tenant_str) = if let Some(tenant) = tenant_id { - (format!("{tenant}/{stream_name}/{date}/"), tenant.as_str()) - } else { - (format!("{stream_name}/{date}/"), DEFAULT_TENANT) - }; + let pre = format!("{}/{date}/", stream_prefix(stream_name, tenant_id)); + let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); let resp = self .client .list_with_delimiter(Some(&object_store::path::Path::from(pre.as_str()))) @@ -1282,14 +1273,8 @@ impl ObjectStorage for S3 { hour: &str, tenant_id: &Option, ) -> Result, ObjectStorageError> { - let (pre, tenant_str) = if let Some(tenant) = tenant_id { - ( - format!("{tenant}/{stream_name}/{date}/{hour}/"), - tenant.as_str(), - ) - } else { - (format!("{stream_name}/{date}/{hour}/"), DEFAULT_TENANT) - }; + let pre = format!("{}/{date}/{hour}/", stream_prefix(stream_name, tenant_id)); + let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); let resp = self .client .list_with_delimiter(Some(&object_store::path::Path::from(pre.as_str()))) diff --git a/src/users/dashboards.rs b/src/users/dashboards.rs index 5e0446bb3..fb49ecb72 100644 --- a/src/users/dashboards.rs +++ b/src/users/dashboards.rs @@ -30,6 +30,7 @@ use crate::{ handlers::http::users::{DASHBOARDS_DIR, USERS_ROOT_DIR, dashboards::DashboardError}, metastore::metastore_traits::MetastoreObject, parseable::{DEFAULT_TENANT, PARSEABLE}, + storage::object_storage::tenant_prefix, }; pub static DASHBOARDS: Lazy = Lazy::new(Dashboards::default); @@ -75,7 +76,7 @@ pub struct Dashboard { impl MetastoreObject for Dashboard { fn get_object_path(&self) -> String { - let root = self.tenant_id.as_deref().unwrap_or(""); + let root = tenant_prefix(&self.tenant_id); RelativePathBuf::from_iter([ root, USERS_ROOT_DIR,