diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs index fac667202f5..0aa79c1d21a 100644 --- a/quickwit/quickwit-storage/src/lib.rs +++ b/quickwit/quickwit-storage/src/lib.rs @@ -58,6 +58,7 @@ mod versioned_component; use quickwit_common::uri::Uri; pub use split_cache::SearchSplitCache; +pub use split_range_cache::FoyerSplitRangeCache; pub use tantivy::directory::OwnedBytes; pub use versioned_component::VersionedComponent; diff --git a/quickwit/quickwit-storage/src/split_range_cache/mod.rs b/quickwit/quickwit-storage/src/split_range_cache/mod.rs index e404115a030..18265ca9c93 100644 --- a/quickwit/quickwit-storage/src/split_range_cache/mod.rs +++ b/quickwit/quickwit-storage/src/split_range_cache/mod.rs @@ -13,7 +13,174 @@ // limitations under the License. mod key; +#[cfg(test)] +mod tests; -// Later split-range cache tasks import this handle from the module root. -#[allow(unused_imports)] +use std::path::Path; +use std::sync::Arc; + +use anyhow::Context; +use bytes::Bytes; +use bytesize::ByteSize; +use foyer::{Code, DeviceBuilder}; pub(crate) use key::SplitRangeCacheKey; +use quickwit_config::{ + CachePolicy, DiskCompression, RecoverMode, SplitRangeCacheWritePolicy, + SplitRangeDiskCacheConfig, +}; + +/// Process-wide Foyer hybrid cache for exact split byte-range payloads. +pub struct FoyerSplitRangeCache { + pub(crate) cache: foyer::HybridCache, + #[allow(dead_code)] + pub(crate) max_entry_size: usize, + #[allow(dead_code)] + pub(crate) block_size: usize, +} + +impl FoyerSplitRangeCache { + /// Create `config.path` if needed, open Foyer's cache files there, recover + /// any existing image, and start flushers/reclaimers. + /// + /// This does not mount a volume; the path must already sit on a usable + /// filesystem (for example an EBS mount). + pub async fn open(config: &SplitRangeDiskCacheConfig) -> anyhow::Result { + tokio::fs::create_dir_all(&config.path) + .await + .with_context(|| { + format!( + "failed to create split range cache directory `{}`", + config.path.display() + ) + })?; + let device = build_foyer_fs_device( + &config.path, + bytesize_to_usize(config.disk_capacity, "disk_capacity")?, + )?; + let engine = build_block_engine(config, device)?; + let memory_capacity = bytesize_to_usize(config.memory_capacity, "memory_capacity")?; + let cache = foyer::HybridCacheBuilder::new() + .with_name("split-range-v1") + .with_policy(foyer_write_policy(config.write_policy)) + .with_flush_on_close(foyer_flush_on_close(config.write_policy)) + .memory(memory_capacity) + .with_eviction_config(foyer_memory_eviction_config(config.memory_eviction_policy)?) + .with_weighter(|key: &SplitRangeCacheKey, value: &Bytes| { + key.estimated_size() + value.len() + }) + .storage() + .with_engine_config(engine) + .with_recover_mode(foyer_recover_mode(config.recover_mode)) + .with_compression(foyer_compression(config.compression)) + .build() + .await?; + Ok(Self { + cache, + max_entry_size: bytesize_to_usize(config.max_entry_size, "max_entry_size")?, + block_size: bytesize_to_usize(config.block_size, "block_size")?, + }) + } + + /// Stop new disk writes and wait for in-flight flush/reclaim work. + /// + /// Cache files stay on disk for the next [`Self::open`]. This drops file + /// descriptors only; it does not unmount the volume. + pub async fn close(&self) -> anyhow::Result<()> { + self.cache.close().await.map_err(Into::into) + } +} + +pub(crate) fn foyer_write_policy(policy: SplitRangeCacheWritePolicy) -> foyer::HybridCachePolicy { + match policy { + SplitRangeCacheWritePolicy::WriteOnEviction => foyer::HybridCachePolicy::WriteOnEviction, + SplitRangeCacheWritePolicy::WriteOnInsertion => foyer::HybridCachePolicy::WriteOnInsertion, + } +} + +/// Flush the memory tier on close under write-on-eviction so a graceful +/// restart can recover hot entries. Write-on-insertion already submitted +/// those entries to disk. +pub(crate) fn foyer_flush_on_close(policy: SplitRangeCacheWritePolicy) -> bool { + matches!(policy, SplitRangeCacheWritePolicy::WriteOnEviction) +} + +fn foyer_recover_mode(recover_mode: RecoverMode) -> foyer::RecoverMode { + match recover_mode { + RecoverMode::Quiet => foyer::RecoverMode::Quiet, + } +} + +fn foyer_compression(compression: DiskCompression) -> foyer::Compression { + match compression { + DiskCompression::Lz4 => foyer::Compression::Lz4, + } +} + +fn foyer_memory_eviction_config(policy: CachePolicy) -> anyhow::Result { + match policy { + CachePolicy::S3Fifo => Ok(foyer::S3FifoConfig::default()), + CachePolicy::Lru | CachePolicy::TinyLfu => { + anyhow::bail!( + "split_range_disk_cache.memory_eviction_policy must be s3-fifo in phase 1" + ) + } + } +} + +fn build_foyer_fs_device( + path: &Path, + disk_capacity: usize, +) -> anyhow::Result> { + let device = foyer::FsDeviceBuilder::new(path) + .with_capacity(disk_capacity) + .with_throttle(foyer::Throttle::default()) + .build()?; + Ok(device) +} + +fn build_block_engine( + config: &SplitRangeDiskCacheConfig, + device: Arc, +) -> anyhow::Result> +{ + Ok(foyer::BlockEngineConfig::new(device) + .with_block_size(bytesize_to_usize(config.block_size, "block_size")?) + .with_flushers(config.flushers) + .with_reclaimers(config.reclaimers) + .with_buffer_pool_size(bytesize_to_usize( + config.buffer_pool_size, + "buffer_pool_size", + )?) + .with_submit_queue_size_threshold(bytesize_to_usize( + config.submit_queue_size_threshold, + "submit_queue_size_threshold", + )?) + .with_eviction_pickers(vec![ + Box::new(foyer::InvalidRatioPicker::new(0.8)), + Box::::default(), + ])) +} + +fn bytesize_to_usize(size: ByteSize, field: &'static str) -> anyhow::Result { + usize::try_from(size.as_u64()) + .with_context(|| format!("split_range_disk_cache.{field} does not fit usize")) +} + +#[cfg(test)] +pub(crate) fn config_for_test(path: impl AsRef) -> SplitRangeDiskCacheConfig { + SplitRangeDiskCacheConfig { + path: path.as_ref().to_path_buf(), + disk_capacity: ByteSize::mb(64), + memory_capacity: ByteSize::mb(8), + buffer_pool_size: ByteSize::mb(4), + submit_queue_size_threshold: ByteSize::mb(8), + memory_eviction_policy: CachePolicy::S3Fifo, + write_policy: SplitRangeCacheWritePolicy::WriteOnEviction, + compression: DiskCompression::Lz4, + recover_mode: RecoverMode::Quiet, + block_size: ByteSize::mb(4), + max_entry_size: ByteSize::mb(2), + flushers: 1, + reclaimers: 1, + } +} diff --git a/quickwit/quickwit-storage/src/split_range_cache/tests.rs b/quickwit/quickwit-storage/src/split_range_cache/tests.rs new file mode 100644 index 00000000000..e39b5750546 --- /dev/null +++ b/quickwit/quickwit-storage/src/split_range_cache/tests.rs @@ -0,0 +1,53 @@ +// 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 quickwit_config::SplitRangeCacheWritePolicy; + +use super::*; + +#[test] +fn test_flush_on_close_pairs_with_write_policy() { + assert!(foyer_flush_on_close( + SplitRangeCacheWritePolicy::WriteOnEviction + )); + assert!(!foyer_flush_on_close( + SplitRangeCacheWritePolicy::WriteOnInsertion + )); +} + +#[tokio::test] +async fn test_split_range_cache_builder_uses_configured_policy_and_throttle() { + let temp_dir = tempfile::tempdir().unwrap(); + let cache = FoyerSplitRangeCache::open(&config_for_test(temp_dir.path())) + .await + .unwrap(); + assert_eq!( + cache.cache.policy(), + foyer::HybridCachePolicy::WriteOnEviction + ); + cache.close().await.unwrap(); +} + +#[tokio::test] +async fn test_split_range_cache_builder_write_on_insertion() { + let temp_dir = tempfile::tempdir().unwrap(); + let mut config = config_for_test(temp_dir.path()); + config.write_policy = SplitRangeCacheWritePolicy::WriteOnInsertion; + let cache = FoyerSplitRangeCache::open(&config).await.unwrap(); + assert_eq!( + cache.cache.policy(), + foyer::HybridCachePolicy::WriteOnInsertion + ); + cache.close().await.unwrap(); +}