diff --git a/config/quickwit.yaml b/config/quickwit.yaml index 99ebda63717..0b6ba45b236 100644 --- a/config/quickwit.yaml +++ b/config/quickwit.yaml @@ -138,6 +138,7 @@ indexer: # searcher: # fast_field_cache_capacity: 1G # split_footer_cache_capacity: 500M +# split_footer_disk_cache_capacity: 1G # partial_request_cache_capacity: 64M # max_num_concurrent_split_streams: 100 # max_num_concurrent_split_searches: 100 diff --git a/docs/configuration/node-config.md b/docs/configuration/node-config.md index ed4bca51517..1a6811ba76c 100644 --- a/docs/configuration/node-config.md +++ b/docs/configuration/node-config.md @@ -200,6 +200,7 @@ This section contains the configuration options for a Searcher. | `aggregation_bucket_limit` | Determines the maximum number of buckets returned to the client. | `65000` | | `fast_field_cache_capacity` | Fast field in memory cache capacity on a Searcher. If your filter by dates, run aggregations, range queries, or even for tracing, it might worth increasing this parameter. The [metrics](../reference/metrics.md) starting by `quickwit_cache_fastfields_cache` can help you make an informed choice when setting this value. | `1G` | | `split_footer_cache_capacity` | Split footer in memory cache (it is essentially the hotcache) capacity on a Searcher.| `500M` | +| `split_footer_disk_cache_capacity` | Opt-in, persistent on-disk cache for split footers, stored in the data directory under `searcher-split-footer-cache/`. When set, split footers survive a searcher restart, avoiding refetching the hotcache from the object store on startup. Disabled if unspecified. | | | `partial_request_cache_capacity` | Partial request in memory cache capacity on a Searcher. Cache intermediate state for a request, possibly making subsequent requests faster. It can be disabled by setting the size to `0`. | `64M` | | `max_num_concurrent_split_searches` | Maximum number of concurrent split search requests running on a Searcher. | `100` | | `split_cache` | Searcher split cache configuration options defined in the section below. Cache disabled if unspecified. | | diff --git a/docs/overview/concepts/querying.md b/docs/overview/concepts/querying.md index a883ea21dce..79ce0ab90df 100644 --- a/docs/overview/concepts/querying.md +++ b/docs/overview/concepts/querying.md @@ -98,6 +98,7 @@ In memory: On disk: - The split cache stores entire splits on disk. It can be enabled by setting the `split_cache` configuration fields. This cache can help reduce object store costs and load. Searchers populate this cache when splits are created or queried and evict them with a simple LRU strategy. +- The split footer disk cache persists split footers on disk. It can be enabled by setting the `split_footer_disk_cache_capacity` configuration value. It backs the in-memory footer cache with a persistent LRU tier so that, after a searcher restart, footers do not have to be refetched from the object store. Cached footers are promoted back into memory on the first access. Learn more about cache parameters in the [searcher configuration docs](../../configuration/node-config.md#searcher-configuration). diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index f14752574a8..60423de8200 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -274,6 +274,8 @@ pub struct SearcherConfig { pub aggregation_bucket_limit: u32, pub fast_field_cache_capacity: ByteSize, pub split_footer_cache_capacity: ByteSize, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub split_footer_disk_cache_capacity: Option, pub partial_request_cache_capacity: ByteSize, pub predicate_cache_capacity: ByteSize, pub max_num_concurrent_split_searches: usize, @@ -333,6 +335,7 @@ impl Default for SearcherConfig { SearcherConfig { fast_field_cache_capacity: ByteSize::gb(1), split_footer_cache_capacity: ByteSize::mb(500), + split_footer_disk_cache_capacity: None, partial_request_cache_capacity: ByteSize::mb(64), predicate_cache_capacity: ByteSize::mb(256), max_num_concurrent_split_searches: 100, diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index 1acd648729e..5524a534139 100644 --- a/quickwit/quickwit-config/src/node_config/serialize.rs +++ b/quickwit/quickwit-config/src/node_config/serialize.rs @@ -663,6 +663,7 @@ mod tests { aggregation_bucket_limit: 500_000, fast_field_cache_capacity: ByteSize::gb(10), split_footer_cache_capacity: ByteSize::gb(1), + split_footer_disk_cache_capacity: None, partial_request_cache_capacity: ByteSize::mb(64), predicate_cache_capacity: ByteSize::mb(256), max_num_concurrent_split_searches: 150, diff --git a/quickwit/quickwit-search/src/leaf.rs b/quickwit/quickwit-search/src/leaf.rs index f184375438e..02c86520c8b 100644 --- a/quickwit/quickwit-search/src/leaf.rs +++ b/quickwit/quickwit-search/src/leaf.rs @@ -35,8 +35,8 @@ use quickwit_query::query_ast::{ }; use quickwit_query::tokenizers::TokenizerManager; use quickwit_storage::{ - BundleStorage, ByteRangeCache, MemorySizedCache, OwnedBytes, SplitCache, Storage, - StorageResolver, TimeoutAndRetryStorage, wrap_storage_with_cache, + BundleStorage, ByteRangeCache, OwnedBytes, SplitCache, Storage, StorageResolver, + TieredSizedCache, TimeoutAndRetryStorage, wrap_storage_with_cache, }; use tantivy::aggregation::agg_req::{AggregationVariants, Aggregations}; use tantivy::aggregation::{AggContextParams, AggregationLimitsGuard}; @@ -60,10 +60,10 @@ use crate::{QuickwitAggregations, SearchError}; async fn get_split_footer_from_cache_or_fetch( index_storage: Arc, split_and_footer_offsets: &SplitIdAndFooterOffsets, - footer_cache: &MemorySizedCache, + footer_cache: &TieredSizedCache, ) -> anyhow::Result { { - let possible_val = footer_cache.get(&split_and_footer_offsets.split_id); + let possible_val = footer_cache.get(&split_and_footer_offsets.split_id).await; if let Some(footer_data) = possible_val { return Ok(footer_data); } @@ -84,10 +84,12 @@ async fn get_split_footer_from_cache_or_fetch( ) })?; - footer_cache.put( - split_and_footer_offsets.split_id.to_owned(), - footer_data_opt.clone(), - ); + footer_cache + .put( + split_and_footer_offsets.split_id.to_owned(), + footer_data_opt.clone(), + ) + .await; Ok(footer_data_opt) } diff --git a/quickwit/quickwit-search/src/lib.rs b/quickwit/quickwit-search/src/lib.rs index 73d9cf05ffe..00f78e8d1c6 100644 --- a/quickwit/quickwit-search/src/lib.rs +++ b/quickwit/quickwit-search/src/lib.rs @@ -345,7 +345,7 @@ pub async fn single_node_search( let search_job_placer = SearchJobPlacer::new(searcher_pool.clone()); let cluster_client = ClusterClient::new(search_job_placer); let searcher_config = SearcherConfig::default(); - let searcher_context = Arc::new(SearcherContext::new(searcher_config, None)); + let searcher_context = Arc::new(SearcherContext::new(searcher_config, None, None)); let search_service = Arc::new(SearchServiceImpl::new( metastore.clone(), storage_resolver, diff --git a/quickwit/quickwit-search/src/service.rs b/quickwit/quickwit-search/src/service.rs index 74648c0e1eb..21ce58f6221 100644 --- a/quickwit/quickwit-search/src/service.rs +++ b/quickwit/quickwit-search/src/service.rs @@ -29,7 +29,8 @@ use quickwit_proto::search::{ SearchResponse, SnippetRequest, }; use quickwit_storage::{ - MemorySizedCache, QuickwitCache, SplitCache, StorageCache, StorageResolver, + DiskSizedCache, MemorySizedCache, QuickwitCache, SplitCache, StorageCache, StorageResolver, + TieredSizedCache, }; use tantivy::aggregation::AggregationLimitsGuard; @@ -405,8 +406,8 @@ pub struct SearcherContext { pub fast_fields_cache: Arc, /// Limit the concurrency for small snappy interactive searches. pub search_permit_provider: SearchPermitProvider, - /// Split footer cache. - pub split_footer_cache: MemorySizedCache, + /// Split footer cache. In-memory, optionally backed by a persistent on-disk tier. + pub split_footer_cache: TieredSizedCache, /// Per-split and per-query cache. pub leaf_search_cache: LeafSearchCache, /// Per-split and per-predicate cache. @@ -432,16 +433,23 @@ impl SearcherContext { #[cfg(test)] pub fn for_test() -> SearcherContext { let searcher_config = SearcherConfig::default(); - SearcherContext::new(searcher_config, None) + SearcherContext::new(searcher_config, None, None) } - /// Creates a new searcher context, given a searcher config, and an optional `SplitCache`. - pub fn new(searcher_config: SearcherConfig, split_cache_opt: Option>) -> Self { + /// Creates a new searcher context, given a searcher config, an optional `SplitCache` and an + /// optional persistent (on-disk) split footer cache. + pub fn new( + searcher_config: SearcherConfig, + split_cache_opt: Option>, + split_footer_disk_cache_opt: Option>, + ) -> Self { let capacity_in_bytes = searcher_config.split_footer_cache_capacity.as_u64() as usize; let global_split_footer_cache = MemorySizedCache::with_capacity_in_bytes( capacity_in_bytes, &quickwit_storage::STORAGE_METRICS.split_footer_cache, ); + let split_footer_cache = + TieredSizedCache::new(global_split_footer_cache, split_footer_disk_cache_opt); let leaf_search_split_semaphore = SearchPermitProvider::new( searcher_config.max_num_concurrent_split_searches, searcher_config.warmup_memory_budget, @@ -465,7 +473,7 @@ impl SearcherContext { fast_fields_cache: storage_long_term_cache, predicate_cache: predicate_cache.into(), search_permit_provider: leaf_search_split_semaphore, - split_footer_cache: global_split_footer_cache, + split_footer_cache, leaf_search_cache, list_fields_cache, split_cache_opt, diff --git a/quickwit/quickwit-search/src/tests.rs b/quickwit/quickwit-search/src/tests.rs index d20fbd881d8..a0b01a3f665 100644 --- a/quickwit/quickwit-search/src/tests.rs +++ b/quickwit/quickwit-search/src/tests.rs @@ -1125,7 +1125,7 @@ async fn test_search_util(test_sandbox: &TestSandbox, query: &str) -> Vec { ..Default::default() }); let searcher_context: Arc = - Arc::new(SearcherContext::new(SearcherConfig::default(), None)); + Arc::new(SearcherContext::new(SearcherConfig::default(), None, None)); let agg_limits = searcher_context.get_aggregation_limits(); @@ -1765,7 +1765,7 @@ async fn test_single_node_list_terms() -> anyhow::Result<()> { .into_iter() .map(|split| extract_split_and_footer_offsets(&split.split_metadata)) .collect(); - let searcher_context = Arc::new(SearcherContext::new(SearcherConfig::default(), None)); + let searcher_context = Arc::new(SearcherContext::new(SearcherConfig::default(), None, None)); { let request = ListTermsRequest { diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index 390519d33ae..bb8eefc53ad 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -113,7 +113,7 @@ use quickwit_search::{ SearchJobPlacer, SearchService, SearchServiceClient, SearcherContext, SearcherPool, create_search_client_from_channel, start_searcher_service, }; -use quickwit_storage::{SplitCache, StorageResolver}; +use quickwit_storage::{DiskSizedCache, SplitCache, StorageResolver}; use tcp_listener::TcpListenerResolver; use tokio::sync::oneshot; use tonic_health::ServingStatus; @@ -658,9 +658,32 @@ pub async fn serve_quickwit( None }; + let split_footer_disk_cache_opt: Option> = if let Some(capacity) = + node_config.searcher_config.split_footer_disk_cache_capacity + { + match DiskSizedCache::open( + node_config + .data_dir_path + .join("searcher-split-footer-cache"), + capacity.as_u64(), + &quickwit_storage::STORAGE_METRICS.split_footer_disk_cache, + ) + .await + { + Ok(disk_cache) => Some(disk_cache), + Err(error) => { + error!(%error, "failed to open the persistent split footer cache, disabling it"); + None + } + } + } else { + None + }; + let searcher_context = Arc::new(SearcherContext::new( node_config.searcher_config.clone(), split_cache_opt, + split_footer_disk_cache_opt, )); let (search_job_placer, search_service) = setup_searcher( @@ -1581,7 +1604,8 @@ mod tests { #[tokio::test] async fn test_setup_searcher() { let node_config = NodeConfig::for_test(); - let searcher_context = Arc::new(SearcherContext::new(SearcherConfig::default(), None)); + let searcher_context = + Arc::new(SearcherContext::new(SearcherConfig::default(), None, None)); let metastore = metastore_for_test(); let (change_stream, change_stream_tx) = ClusterChangeStream::new_unbounded(); let storage_resolver = StorageResolver::unconfigured(); diff --git a/quickwit/quickwit-serve/src/soft_delete_api/handler.rs b/quickwit/quickwit-serve/src/soft_delete_api/handler.rs index b7000237573..009b7805cf3 100644 --- a/quickwit/quickwit-serve/src/soft_delete_api/handler.rs +++ b/quickwit/quickwit-serve/src/soft_delete_api/handler.rs @@ -207,8 +207,11 @@ mod tests { let search_job_placer = SearchJobPlacer::new(searcher_pool.clone()); let cluster_client = ClusterClient::new(search_job_placer); let searcher_config = SearcherConfig::default(); - let searcher_context = - Arc::new(quickwit_search::SearcherContext::new(searcher_config, None)); + let searcher_context = Arc::new(quickwit_search::SearcherContext::new( + searcher_config, + None, + None, + )); let search_service: Arc = Arc::new(SearchServiceImpl::new( sandbox.metastore(), sandbox.storage_resolver(), diff --git a/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs new file mode 100644 index 00000000000..c1af2186081 --- /dev/null +++ b/quickwit/quickwit-storage/src/cache/disk_sized_cache.rs @@ -0,0 +1,516 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::fmt::Display; +use std::hash::Hasher; +use std::io; +use std::marker::PhantomData; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Instant, SystemTime}; + +use fnv::FnvHasher; +use lru::LruCache; +use tracing::{info, warn}; + +use crate::OwnedBytes; +use crate::metrics::CacheMetrics; + +/// Substring used to mark files that are being written. +const TEMP_MARKER: &str = ".tmp"; + +/// Global counter used to build unique temporary file names. +static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); + +struct DiskCacheIndex { + /// Maps the on-disk file name to the size of its content. + /// The LRU order tracks the recency of accesses for eviction. + lru_cache: LruCache, + num_bytes: u64, + capacity_in_bytes: u64, + cache_counters: &'static CacheMetrics, +} + +impl DiskCacheIndex { + fn record_item(&mut self, num_bytes: u64) { + self.num_bytes += num_bytes; + self.cache_counters.in_cache_count.inc(); + self.cache_counters.in_cache_num_bytes.add(num_bytes as i64); + } + + fn drop_item(&mut self, num_bytes: u64) { + self.num_bytes -= num_bytes; + self.cache_counters.in_cache_count.dec(); + self.cache_counters.in_cache_num_bytes.sub(num_bytes as i64); + self.cache_counters.evict_num_items.inc(); + self.cache_counters.evict_num_bytes.inc_by(num_bytes); + } + + /// Evicts the least recently used entries until `incoming` extra bytes would fit + /// under the capacity. Returns the file names that must be deleted from disk. + fn evict_to_fit(&mut self, incoming: u64) -> Vec { + let mut victims = Vec::new(); + while self.num_bytes + incoming > self.capacity_in_bytes { + if let Some((file_name, num_bytes)) = self.lru_cache.pop_lru() { + self.drop_item(num_bytes); + victims.push(file_name); + } else { + break; + } + } + victims + } +} + +impl Drop for DiskCacheIndex { + fn drop(&mut self) { + // We don't count this toward evicted entries, as we are clearing the whole index. + self.cache_counters + .in_cache_count + .sub(self.lru_cache.len() as i64); + self.cache_counters + .in_cache_num_bytes + .sub(self.num_bytes as i64); + } +} + +/// A size-bounded cache that persists its entries on disk. +/// +/// This is the on-disk counterpart to [`MemorySizedCache`](super::MemorySizedCache): entries +/// are evicted following an LRU policy once the configured capacity (in bytes) is exceeded. +/// +/// Each entry is stored as a single file named after `key.to_string()`. Keys must therefore +/// produce filesystem-safe, collision-free strings. +pub struct DiskSizedCache { + root_path: PathBuf, + index: Mutex, + _phantom: PhantomData K>, +} + +impl DiskSizedCache { + /// Opens a disk cache rooted at `root_path`. + pub async fn open( + root_path: PathBuf, + capacity_in_bytes: u64, + cache_counters: &'static CacheMetrics, + ) -> io::Result + where + K: 'static, + { + tokio::task::spawn_blocking(move || { + Self::open_blocking(root_path, capacity_in_bytes, cache_counters) + }) + .await + .map_err(io::Error::other)? + } + + fn open_blocking( + root_path: PathBuf, + capacity_in_bytes: u64, + cache_counters: &'static CacheMetrics, + ) -> io::Result { + let start = Instant::now(); + std::fs::create_dir_all(&root_path)?; + + let mut entries: Vec<(String, u64, SystemTime)> = Vec::new(); + for shard_entry_res in std::fs::read_dir(&root_path)? { + let shard_entry = shard_entry_res?; + // Entries live inside shard sub-directories; only recurse into those. + match shard_entry.file_type() { + Ok(file_type) if file_type.is_dir() => {} + _ => continue, + } + let Ok(shard_dir_iter) = std::fs::read_dir(shard_entry.path()) else { + continue; + }; + + for dir_entry_res in shard_dir_iter { + let Ok(dir_entry) = dir_entry_res else { + continue; + }; + + if let Ok(file_type) = dir_entry.file_type() + && !file_type.is_file() + { + continue; + } + let Ok(file_name) = dir_entry.file_name().into_string() else { + continue; + }; + + if file_name.contains(TEMP_MARKER) { + // Leftover temporary file from an interrupted write: clean it up. + let _ = std::fs::remove_file(dir_entry.path()); + continue; + } + let Ok(metadata) = dir_entry.metadata() else { + continue; + }; + if !metadata.is_file() { + continue; + } + let modified = metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH); + entries.push((file_name, metadata.len(), modified)); + } + } + entries.sort_by_key(|(_, _, modified)| *modified); + + let mut index = DiskCacheIndex { + lru_cache: LruCache::unbounded(), + num_bytes: 0, + capacity_in_bytes, + cache_counters, + }; + for (file_name, num_bytes, _) in entries { + index.record_item(num_bytes); + index.lru_cache.put(file_name, num_bytes); + } + let victims = index.evict_to_fit(0); + + let cache = DiskSizedCache { + root_path, + index: Mutex::new(index), + _phantom: PhantomData, + }; + remove_files(&cache.root_path, &victims); + let num_entries = { + let index = cache.index.lock().unwrap(); + index.lru_cache.len() + }; + info!( + root_path = %cache.root_path.display(), + num_entries, + num_evicted = victims.len(), + elapsed_millis = start.elapsed().as_millis(), + "opened disk cache" + ); + Ok(cache) + } + + /// Returns the cached payload for the given key, if present on disk. + pub async fn get(&self, key: &K) -> Option { + let file_name = key.to_string(); + { + let mut index = self.index.lock().unwrap(); + // `LruCache::get` refreshes the recency of the entry. + if index.lru_cache.get(&file_name).is_none() { + index.cache_counters.misses_num_items.inc(); + return None; + } + } + // Offload the blocking read so we don't stall the async runtime worker. + let path = path_for(&self.root_path, &file_name); + let read_res = tokio::task::spawn_blocking(move || std::fs::read(path)).await; + match read_res { + Ok(Ok(buffer)) => { + let index = self.index.lock().unwrap(); + index.cache_counters.hits_num_items.inc(); + index + .cache_counters + .hits_num_bytes + .inc_by(buffer.len() as u64); + Some(OwnedBytes::new(buffer)) + } + Ok(Err(_)) => { + // The file vanished (e.g. concurrent eviction or manual deletion): drop the + // stale index entry and report a miss. + let mut index = self.index.lock().unwrap(); + if let Some(num_bytes) = index.lru_cache.pop(&file_name) { + index.num_bytes -= num_bytes; + index.cache_counters.in_cache_count.dec(); + index + .cache_counters + .in_cache_num_bytes + .sub(num_bytes as i64); + } + index.cache_counters.misses_num_items.inc(); + None + } + Err(_join_error) => { + // The blocking read task failed unexpectedly. Keep the index entry (the file is + // likely still valid) and just report a miss. + let index = self.index.lock().unwrap(); + index.cache_counters.misses_num_items.inc(); + None + } + } + } + + /// Stores the given payload on disk under the given key. + /// + /// This silently does nothing if the payload is larger than the whole cache capacity. If an + /// entry already exists for the key, it is kept as-is (payloads are assumed immutable) and its + /// recency is refreshed. + pub async fn put(&self, key: K, bytes: OwnedBytes) { + let num_bytes = bytes.len() as u64; + let file_name = key.to_string(); + { + let mut index = self.index.lock().unwrap(); + if index.capacity_in_bytes < num_bytes { + if index.capacity_in_bytes != 0 { + warn!( + capacity_in_bytes = index.capacity_in_bytes, + len = num_bytes, + "payload larger than the disk cache capacity, not caching it on disk" + ); + } + return; + } + if index.lru_cache.get(&file_name).is_some() { + // Already cached: payloads are immutable, just keep the refreshed recency. + return; + } + } + + // We write the new file *before* evicting on purpose: if the write fails we return early + // without having deleted any valid entry, and the write stays outside the index lock. The + // tradeoff is that on-disk usage may transiently exceed the capacity by roughly one entry. + // The blocking write is offloaded so we don't stall the async runtime worker. + let write_res = { + let root_path = self.root_path.clone(); + let write_name = file_name.clone(); + tokio::task::spawn_blocking(move || write_file(&root_path, &write_name, &bytes)).await + }; + match write_res { + Ok(Ok(())) => {} + Ok(Err(error)) => { + warn!(%error, file_name, "failed to persist entry to disk cache"); + return; + } + Err(_join_error) => return, + } + + let victims = { + let mut index = self.index.lock().unwrap(); + if index.lru_cache.get(&file_name).is_some() { + // Lost a race with a concurrent put of the same (immutable) payload. + return; + } + let victims = index.evict_to_fit(num_bytes); + index.record_item(num_bytes); + index.lru_cache.put(file_name, num_bytes); + victims + }; + if !victims.is_empty() { + let root_path = self.root_path.clone(); + let _ = tokio::task::spawn_blocking(move || remove_files(&root_path, &victims)).await; + } + } +} + +/// Returns the shard sub-directory name a file belongs to. +fn shard_dir(file_name: &str) -> String { + // Number of shard sub-directories files are spread across. + const NUM_SHARDS: u64 = 256; + + let mut hasher = FnvHasher::default(); + hasher.write(file_name.as_bytes()); + format!("{:02x}", hasher.finish() % NUM_SHARDS) +} + +/// Returns the full on-disk path of an entry, including its shard sub-directory. +pub(crate) fn path_for(root_path: &Path, file_name: &str) -> PathBuf { + root_path.join(shard_dir(file_name)).join(file_name) +} + +fn write_file(root_path: &Path, file_name: &str, bytes: &[u8]) -> io::Result<()> { + let shard_path = root_path.join(shard_dir(file_name)); + std::fs::create_dir_all(&shard_path)?; + // Rely on a counter to guarantee uniqueness of the temporary file name. + let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); + let tmp_path = shard_path.join(format!("{file_name}{TEMP_MARKER}{counter}")); + std::fs::write(&tmp_path, bytes)?; + std::fs::rename(&tmp_path, shard_path.join(file_name)) +} + +fn remove_files(root_path: &Path, file_names: &[String]) { + for file_name in file_names { + let _ = std::fs::remove_file(path_for(root_path, file_name)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::metrics::CACHE_METRICS_FOR_TESTS; + + async fn open_cache(root_path: PathBuf, capacity_in_bytes: u64) -> DiskSizedCache { + DiskSizedCache::open(root_path, capacity_in_bytes, &CACHE_METRICS_FOR_TESTS) + .await + .unwrap() + } + + #[tokio::test] + async fn test_put_get() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; + assert!(cache.get(&"missing".to_string()).await.is_none()); + + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); + // A file should have been written on disk, inside its shard sub-directory. + assert!(path_for(tmp_dir.path(), "a").try_exists().unwrap()); + } + + #[tokio::test] + async fn test_persists_across_reopen() { + let tmp_dir = tempfile::tempdir().unwrap(); + { + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + cache + .put("b".to_string(), OwnedBytes::new(&b"world"[..])) + .await; + } + // Re-opening the cache should recover the previously stored entries. + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); + assert_eq!(cache.get(&"b".to_string()).await.unwrap(), &b"world"[..]); + } + + #[tokio::test] + async fn test_lru_eviction() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 6).await; + cache + .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) + .await; + cache + .put("b".to_string(), OwnedBytes::new(&b"bbb"[..])) + .await; + // Access "a" so that "b" becomes the least recently used entry. + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"aaa"[..]); + // Inserting a third entry must evict "b". + cache + .put("c".to_string(), OwnedBytes::new(&b"ccc"[..])) + .await; + + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"aaa"[..]); + assert!(cache.get(&"b".to_string()).await.is_none()); + assert_eq!(cache.get(&"c".to_string()).await.unwrap(), &b"ccc"[..]); + // The evicted entry's file must be gone. + assert!(!path_for(tmp_dir.path(), "b").try_exists().unwrap()); + } + + #[tokio::test] + async fn test_payload_larger_than_capacity_is_ignored() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 3).await; + cache + .put("big".to_string(), OwnedBytes::new(&b"toolarge"[..])) + .await; + assert!(cache.get(&"big".to_string()).await.is_none()); + assert!(!path_for(tmp_dir.path(), "big").try_exists().unwrap()); + } + + #[tokio::test] + async fn test_put_same_key_twice_keeps_single_entry() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 6).await; + cache + .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) + .await; + // Immutable payload: putting the same key again is a no-op and must not evict others. + cache + .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) + .await; + cache + .put("b".to_string(), OwnedBytes::new(&b"bbb"[..])) + .await; + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"aaa"[..]); + assert_eq!(cache.get(&"b".to_string()).await.unwrap(), &b"bbb"[..]); + } + + #[tokio::test] + async fn test_get_after_manual_file_deletion() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + std::fs::remove_file(path_for(tmp_dir.path(), "a")).unwrap(); + // The stale entry should be detected and reported as a miss. + assert!(cache.get(&"a".to_string()).await.is_none()); + } + + #[tokio::test] + async fn test_open_evicts_when_over_capacity() { + let tmp_dir = tempfile::tempdir().unwrap(); + { + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; + cache + .put("a".to_string(), OwnedBytes::new(&b"aaa"[..])) + .await; + cache + .put("b".to_string(), OwnedBytes::new(&b"bbb"[..])) + .await; + } + // Re-open with a capacity that can only hold one of the two entries. + let cache = open_cache(tmp_dir.path().to_path_buf(), 3).await; + let a = cache.get(&"a".to_string()).await; + let b = cache.get(&"b".to_string()).await; + // Exactly one entry should have survived. + assert_ne!(a.is_some(), b.is_some()); + } + + #[tokio::test] + async fn test_open_cleans_up_temp_files() { + let tmp_dir = tempfile::tempdir().unwrap(); + // A leftover temp file lives inside the shard directory of the entry it belongs to. + let shard_path = tmp_dir.path().join(shard_dir("a")); + std::fs::create_dir_all(&shard_path).unwrap(); + let leftover = shard_path.join("a.tmp42"); + std::fs::write(&leftover, b"partial").unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 1_000).await; + assert!(!leftover.try_exists().unwrap()); + assert!(cache.get(&"a".to_string()).await.is_none()); + } + + #[tokio::test] + async fn test_entries_are_sharded_and_survive_reopen() { + let tmp_dir = tempfile::tempdir().unwrap(); + let cache = open_cache(tmp_dir.path().to_path_buf(), 100_000).await; + // Insert enough entries that at least two distinct shard directories get used. + for i in 0..64 { + let key = format!("key-{i}"); + cache.put(key, OwnedBytes::new(&b"payload"[..])).await; + } + // Files must be nested under shard sub-directories, not directly in the root. + let mut shard_dirs = 0; + for entry in std::fs::read_dir(tmp_dir.path()).unwrap() { + let entry = entry.unwrap(); + assert!( + entry.file_type().unwrap().is_dir(), + "root should only contain shard directories" + ); + shard_dirs += 1; + } + assert!( + shard_dirs > 1, + "entries should be spread over several shards" + ); + + // Every entry is still retrievable, including after a reopen (which walks the shards). + let cache = open_cache(tmp_dir.path().to_path_buf(), 100_000).await; + for i in 0..64 { + let key = format!("key-{i}"); + assert_eq!(cache.get(&key).await.unwrap(), &b"payload"[..]); + } + } +} diff --git a/quickwit/quickwit-storage/src/cache/mod.rs b/quickwit/quickwit-storage/src/cache/mod.rs index fe2a45d2f4f..5c6fa20ff61 100644 --- a/quickwit/quickwit-storage/src/cache/mod.rs +++ b/quickwit/quickwit-storage/src/cache/mod.rs @@ -13,11 +13,13 @@ // limitations under the License. mod byte_range_cache; +mod disk_sized_cache; mod memory_sized_cache; mod quickwit_cache; mod slice_address; mod storage_with_cache; mod stored_item; +mod tiered_sized_cache; use std::ops::Range; use std::path::{Path, PathBuf}; @@ -28,7 +30,9 @@ pub use quickwit_cache::QuickwitCache; pub use storage_with_cache::StorageWithCache; pub use self::byte_range_cache::ByteRangeCache; +pub use self::disk_sized_cache::DiskSizedCache; pub use self::memory_sized_cache::MemorySizedCache; +pub use self::tiered_sized_cache::TieredSizedCache; use crate::{OwnedBytes, Storage}; /// Wraps the given directory with a slice cache that is actually global diff --git a/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs new file mode 100644 index 00000000000..9546ee8174f --- /dev/null +++ b/quickwit/quickwit-storage/src/cache/tiered_sized_cache.rs @@ -0,0 +1,133 @@ +// Copyright 2021-Present Datadog, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::fmt::Display; +use std::hash::Hash; + +use crate::OwnedBytes; +use crate::cache::disk_sized_cache::DiskSizedCache; +use crate::cache::memory_sized_cache::MemorySizedCache; + +/// A two-tier, size-bounded cache combining a fast in-memory tier (L1) with an optional +/// disk-backed tier (L2). +/// +/// Lookups check memory first, then disk; a disk hit is promoted back into memory. Writes +/// populate both tiers. When no disk tier is configured this behaves exactly like the +/// underlying [`MemorySizedCache`], which makes it a drop-in, opt-in replacement. +pub struct TieredSizedCache { + memory: MemorySizedCache, + disk: Option>, +} + +impl TieredSizedCache { + /// Creates a tiered cache from an in-memory tier and an optional disk tier. + pub fn new(memory: MemorySizedCache, disk: Option>) -> Self { + TieredSizedCache { memory, disk } + } + + /// Returns the cached payload for the given key, checking memory first, then disk. + /// + /// A disk hit is promoted into the memory tier before being returned. Only the disk tier + /// performs (off-thread) I/O, so an in-memory hit stays fully synchronous and cheap. + pub async fn get(&self, key: &K) -> Option { + if let Some(bytes) = self.memory.get(key) { + return Some(bytes); + } + let bytes = self.disk.as_ref()?.get(key).await?; + self.memory.put(key.clone(), bytes.clone()); + Some(bytes) + } + + /// Stores the given payload in both the memory tier and, if configured, the disk tier. + pub async fn put(&self, key: K, bytes: OwnedBytes) { + // Populate L1 first so callers benefit immediately even if the disk tier is slow. + self.memory.put(key.clone(), bytes.clone()); + if let Some(disk) = &self.disk { + disk.put(key, bytes).await; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::cache::disk_sized_cache::path_for; + use crate::metrics::CACHE_METRICS_FOR_TESTS; + + fn memory_cache() -> MemorySizedCache { + MemorySizedCache::with_capacity_in_bytes(1_000, &CACHE_METRICS_FOR_TESTS) + } + + #[tokio::test] + async fn test_memory_only() { + let cache = TieredSizedCache::new(memory_cache(), None); + assert!(cache.get(&"a".to_string()).await.is_none()); + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); + } + + #[tokio::test] + async fn test_put_writes_to_both_tiers() { + let tmp_dir = tempfile::tempdir().unwrap(); + let disk = DiskSizedCache::open( + tmp_dir.path().to_path_buf(), + 1_000, + &CACHE_METRICS_FOR_TESTS, + ) + .await + .unwrap(); + let cache = TieredSizedCache::new(memory_cache(), Some(disk)); + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); + // The payload must have been persisted on disk as well. + assert!(path_for(tmp_dir.path(), "a").try_exists().unwrap()); + } + + #[tokio::test] + async fn test_disk_hit_is_promoted_to_memory() { + let tmp_dir = tempfile::tempdir().unwrap(); + { + let disk = DiskSizedCache::open( + tmp_dir.path().to_path_buf(), + 1_000, + &CACHE_METRICS_FOR_TESTS, + ) + .await + .unwrap(); + let cache = TieredSizedCache::new(memory_cache(), Some(disk)); + cache + .put("a".to_string(), OwnedBytes::new(&b"hello"[..])) + .await; + } + // Simulate a fresh process: only the disk tier still holds the data. + let disk = DiskSizedCache::open( + tmp_dir.path().to_path_buf(), + 1_000, + &CACHE_METRICS_FOR_TESTS, + ) + .await + .unwrap(); + let memory = memory_cache(); + let cache = TieredSizedCache::new(memory, Some(disk)); + + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); + // After a disk hit, deleting the file must not lose the value: it lives in memory now. + std::fs::remove_file(path_for(tmp_dir.path(), "a")).unwrap(); + assert_eq!(cache.get(&"a".to_string()).await.unwrap(), &b"hello"[..]); + } +} diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs index 31bbddcdd89..18737ec7ad0 100644 --- a/quickwit/quickwit-storage/src/lib.rs +++ b/quickwit/quickwit-storage/src/lib.rs @@ -64,7 +64,8 @@ pub use self::bundle_storage::{BundleStorage, BundleStorageFileOffsets}; #[cfg(any(test, feature = "testsuite"))] pub use self::cache::MockStorageCache; pub use self::cache::{ - ByteRangeCache, MemorySizedCache, QuickwitCache, StorageCache, wrap_storage_with_cache, + ByteRangeCache, DiskSizedCache, MemorySizedCache, QuickwitCache, StorageCache, + TieredSizedCache, wrap_storage_with_cache, }; pub use self::local_file_storage::{LocalFileStorage, LocalFileStorageFactory}; #[cfg(feature = "azure")] diff --git a/quickwit/quickwit-storage/src/metrics.rs b/quickwit/quickwit-storage/src/metrics.rs index a4de7a53db0..4a94ac4e899 100644 --- a/quickwit/quickwit-storage/src/metrics.rs +++ b/quickwit/quickwit-storage/src/metrics.rs @@ -28,6 +28,7 @@ pub struct StorageMetrics { pub fd_cache_metrics: CacheMetrics, pub fast_field_cache: CacheMetrics, pub split_footer_cache: CacheMetrics, + pub split_footer_disk_cache: CacheMetrics, pub searcher_split_cache: CacheMetrics, pub get_slice_timeout_successes: [IntCounter; 3], pub get_slice_timeout_all_timeouts: IntCounter, @@ -66,6 +67,7 @@ impl Default for StorageMetrics { searcher_split_cache: CacheMetrics::for_component("searcher_split"), shortlived_cache: CacheMetrics::for_component("shortlived"), split_footer_cache: CacheMetrics::for_component("splitfooter"), + split_footer_disk_cache: CacheMetrics::for_component("splitfooter_disk"), get_slice_timeout_successes, get_slice_timeout_all_timeouts, object_storage_requests_total: new_counter_vec(