diff --git a/LICENSE-3rdparty.csv b/LICENSE-3rdparty.csv index b3c307aa803..76cd483b558 100644 --- a/LICENSE-3rdparty.csv +++ b/LICENSE-3rdparty.csv @@ -161,6 +161,7 @@ clap_builder,https://github.com/clap-rs/clap,MIT OR Apache-2.0,The clap_builder clap_lex,https://github.com/clap-rs/clap,MIT OR Apache-2.0,The clap_lex Authors cmac,https://github.com/RustCrypto/MACs,MIT OR Apache-2.0,RustCrypto Developers cmov,https://github.com/RustCrypto/utils,Apache-2.0 OR MIT,RustCrypto Developers +cmsketch,https://github.com/mrcroxx/cmsketch-rs,Apache-2.0,MrCroxx coarsetime,https://github.com/jedisct1/rust-coarsetime,BSD-2-Clause,Frank Denis cobs,https://github.com/jamesmunns/cobs.rs,MIT OR Apache-2.0,"Allen Welkie <>, James Munns " codespan-reporting,https://github.com/brendanzab/codespan,Apache-2.0,Brendan Zabarauskas @@ -186,6 +187,7 @@ constant_time_eq,https://github.com/cesarb/constant_time_eq,CC0-1.0 OR MIT-0 OR convert_case,https://github.com/rutrum/convert-case,MIT,rutrum core-foundation,https://github.com/servo/core-foundation-rs,MIT OR Apache-2.0,The Servo Project Developers core-foundation-sys,https://github.com/servo/core-foundation-rs,MIT OR Apache-2.0,The Servo Project Developers +core_affinity,https://github.com/Elzair/core_affinity_rs,MIT OR Apache-2.0,Philip Woods cpp_demangle,https://github.com/gimli-rs/cpp_demangle,MIT OR Apache-2.0,"Nick Fitzgerald , Jim Blandy , Kyle Huey " cpufeatures,https://github.com/RustCrypto/utils,MIT OR Apache-2.0,RustCrypto Developers crc,https://github.com/mrhooray/crc-rs,MIT OR Apache-2.0,"Rui Hu , Akhil Velagapudi <4@4khil.com>" @@ -306,6 +308,7 @@ event-listener-strategy,https://github.com/smol-rs/event-listener-strategy,Apach evmap,https://github.com/jonhoo/evmap,MIT OR Apache-2.0,Jon Gjengset fail,https://github.com/tikv/fail-rs,Apache-2.0,The TiKV Project Developers fancy-regex,https://github.com/fancy-regex/fancy-regex,MIT,"Raph Levien , Robin Stocker , Keith Hall " +fastant,https://github.com/fast/fastant,MIT,The fastant Authors fastdivide,https://github.com/fulmicoton/fastdivide,zlib-acknowledgement OR MIT,Paul Masurel fastrand,https://github.com/smol-rs/fastrand,Apache-2.0 OR MIT,Stjepan Glavina ff,https://github.com/zkcrypto/ff,MIT OR Apache-2.0,"Sean Bowe , Jack Grigg " @@ -323,6 +326,12 @@ foldhash,https://github.com/orlp/foldhash,Zlib,Orson Peters foreign-types-shared,https://github.com/sfackler/foreign-types,MIT OR Apache-2.0,Steven Fackler form_urlencoded,https://github.com/servo/rust-url,MIT OR Apache-2.0,The rust-url developers +foyer,https://github.com/foyer-rs/foyer,Apache-2.0,MrCroxx +foyer-common,https://github.com/foyer-rs/foyer,Apache-2.0,MrCroxx +foyer-intrusive-collections,https://github.com/foyer-rs/intrusive-rs,Apache-2.0 OR MIT,Amanieu d'Antras +foyer-memory,https://github.com/foyer-rs/foyer,Apache-2.0,MrCroxx +foyer-storage,https://github.com/foyer-rs/foyer,Apache-2.0,MrCroxx +foyer-tokio,https://github.com/foyer-rs/foyer,Apache-2.0,MrCroxx fraction,https://github.com/dnsl48/fraction,MIT OR Apache-2.0,dnsl48 fragile,https://github.com/mitsuhiko/fragile,Apache-2.0,Armin Ronacher fs4,https://github.com/al8n/fs4-rs,MIT OR Apache-2.0,"Dan Burkert , Al Liu " @@ -409,6 +418,7 @@ inout,https://github.com/RustCrypto/utils,MIT OR Apache-2.0,RustCrypto Developer instant,https://github.com/sebcrozet/instant,BSD-3-Clause,sebcrozet integer-encoding,https://github.com/dermesser/integer-encoding-rs,MIT,Lewin Bormann inventory,https://github.com/dtolnay/inventory,MIT OR Apache-2.0,David Tolnay +io-uring,https://github.com/tokio-rs/io-uring,MIT OR Apache-2.0,quininer ipcrypt-rs,https://github.com/jedisct1/rust-ipcrypt2,ISC,Frank Denis ipnet,https://github.com/krisprice/ipnet,MIT OR Apache-2.0,Kris Price ipnetwork,https://github.com/achanda/ipnetwork,MIT OR Apache-2.0,"Abhishek Chanda , Linus Färnstrand " @@ -473,6 +483,7 @@ mea,https://github.com/fast/mea,Apache-2.0,The mea Authors measure_time,https://github.com/PSeitz/rust_measure_time,MIT,Pascal Seitz memchr,https://github.com/BurntSushi/memchr,Unlicense OR MIT,"Andrew Gallant , bluss" memmap2,https://github.com/RazrFalcon/memmap2-rs,MIT OR Apache-2.0,"Dan Burkert , Yevhenii Reizner , The Contributors" +memoffset,https://github.com/Gilnaa/memoffset,MIT,Gilad Naaman metrics,https://github.com/metrics-rs/metrics,MIT,Toby Lawrence metrics-exporter-prometheus,https://github.com/metrics-rs/metrics,MIT AND Apache-2.0,Toby Lawrence metrics-opentelemetry,https://github.com/DoumanAsh/metrics-opentelemetry,BSL-1.0,The metrics-opentelemetry Authors @@ -485,6 +496,7 @@ minimal-lexical,https://github.com/Alexhuszagh/minimal-lexical,MIT OR Apache-2.0 miniserde,https://github.com/dtolnay/miniserde,MIT OR Apache-2.0,David Tolnay miniz_oxide,https://github.com/Frommi/miniz_oxide/tree/master/miniz_oxide,MIT OR Zlib OR Apache-2.0,"Frommi , oyvindln , Rich Geldreich richgel99@gmail.com" mio,https://github.com/tokio-rs/mio,MIT,"Carl Lerche , Thomas de Zeeuw , Tokio Contributors " +mixtrics,https://github.com/foyer-rs/mixtrics,Apache-2.0,MrCroxx mockall,https://github.com/asomers/mockall,MIT OR Apache-2.0,Alan Somers mockall_derive,https://github.com/asomers/mockall,MIT OR Apache-2.0,Alan Somers moka,https://github.com/moka-rs/moka,(MIT OR Apache-2.0) AND Apache-2.0,The moka Authors @@ -776,6 +788,7 @@ simple_asn1,https://github.com/acw/simple_asn1,ISC,Adam Wick siphasher,https://github.com/jedisct1/rust-siphash,MIT OR Apache-2.0,Frank Denis sketches-ddsketch,https://github.com/mheffner/rust-sketches-ddsketch,Apache-2.0,Mike Heffner slab,https://github.com/tokio-rs/slab,MIT,Carl Lerche +small_ctor,https://github.com/mitsuhiko/small-ctor,Apache-2.0,Armin Ronacher smallvec,https://github.com/servo/rust-smallvec,MIT OR Apache-2.0,The Servo Project Developers snafu,https://github.com/shepmaster/snafu,MIT OR Apache-2.0,Jake Goulding snafu-derive,https://github.com/shepmaster/snafu,MIT OR Apache-2.0,Jake Goulding diff --git a/config/quickwit.yaml b/config/quickwit.yaml index 58bdd4942ff..3652035ebaa 100644 --- a/config/quickwit.yaml +++ b/config/quickwit.yaml @@ -199,6 +199,8 @@ indexer: # # service. Searchers require at least one `metastore_read_replica` node at # # startup and do not fall back to the primary metastore. # use_metastore_read_replica: false +# # Fast field RAM cache. Default 1G. Omitted while split_range_disk_cache +# # is set disables this cache; set a capacity explicitly to keep both. # fast_field_cache_capacity: 1G # split_footer_cache_capacity: 500M # partial_request_cache_capacity: 64M @@ -210,6 +212,22 @@ indexer: # max_num_bytes: 1G # max_num_splits: 10000 # num_concurrent_downloads: 1 +# # Process-wide Foyer disk cache for exact split footer and body ranges. +# # Omitted or null disables the cache. write_policy defaults to write-on-eviction. +# split_range_disk_cache: +# path: /quickwit/qwdata/split-range-v1 +# disk_capacity: 1500G +# memory_capacity: 15G +# memory_eviction_policy: s3-fifo +# compression: lz4 +# recover_mode: quiet +# block_size: 64M +# max_entry_size: 60M +# flushers: 24 +# buffer_pool_size: 2G +# submit_queue_size_threshold: 3G +# reclaimers: 8 +# clean_block_threshold: 16 # -------------------------------- Jaeger settings -------------------------------- jaeger: diff --git a/docs/configuration/node-config.md b/docs/configuration/node-config.md index cbe3f8b0022..38fc1b4ef93 100644 --- a/docs/configuration/node-config.md +++ b/docs/configuration/node-config.md @@ -305,11 +305,12 @@ This section contains the configuration options for a Searcher. | --- | --- | --- | | `aggregation_memory_limit` | Controls the maximum amount of memory that can be used for aggregations before aborting. This limit is per searcher node. A node may run concurrent queries, which share the limit. The first query that will hit the limit will be aborted and frees its memory. It is used to prevent excessive memory usage during the aggregation phase, which can lead to performance degradation or crashes. | `500M`| | `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` | +| `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. Default is `1G` when `split_range_disk_cache` is unset. If `split_range_disk_cache` is set and this key is omitted, the RAM cache is disabled. Set a capacity explicitly to keep both. | `1G` | | `split_footer_cache_capacity` | Split footer in memory cache (it is essentially the hotcache) capacity on a Searcher.| `500M` | | `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. | | +| `split_range_disk_cache` | Process-wide on-disk cache for exact split footer and body byte ranges. Configuration options are defined in the section below. Cache disabled if unspecified. | | | `request_timeout_secs` | The time before a search request is cancelled. This should match the timeout of the stack calling into quickwit if there is one set. | `30` | | `use_metastore_read_replica` | If true, routes read-only metastore requests from searchers, including DataFusion when enabled, to nodes running the `metastore_read_replica` service. Searchers require at least one `metastore_read_replica` node at startup and do not fall back to the primary metastore. | `false` | @@ -323,6 +324,26 @@ This section contains the configuration options for the on-disk searcher split c | `max_num_splits` | Maximum number of splits allowed in the split cache. | `10000` | | `num_concurrent_downloads` | Maximum number of concurrent download of splits. | `1` | +### Searcher split range disk cache configuration + +This section contains the configuration options for the process-wide on-disk cache of exact split footer and body ranges. The cache is disabled when this section is omitted or set to `null`. If it is set and `fast_field_cache_capacity` is omitted, the long-lived fast field RAM cache is disabled; set a capacity explicitly to keep both. + +| Property | Description | Default value | +| --- | --- | --- | +| `path` | Directory used to store cache files. Created if missing. Must already sit on a usable filesystem. | | +| `disk_capacity` | Maximum on-disk size of the cache. | | +| `memory_capacity` | In-memory tier size in front of the disk cache. | | +| `buffer_pool_size` | Size of the disk write buffer pool. | | +| `submit_queue_size_threshold` | Maximum amount of data waiting to be flushed to disk. | | +| `memory_eviction_policy` | Eviction policy for the memory tier. Currently only `s3-fifo` is accepted. | | +| `write_policy` | When admitted values are written to disk: `write-on-eviction` or `write-on-insertion`. | `write-on-eviction` | +| `compression` | On-disk compression. Currently only `lz4` is accepted. | | +| `recover_mode` | How existing cache files are recovered on startup. Currently only `quiet` is accepted. | | +| `block_size` | Disk block size. Must be larger than `max_entry_size`. | | +| `max_entry_size` | Maximum uncompressed payload stored as one disk entry. Larger ranges stay in memory only. | | +| `flushers` | Number of flush worker threads. Must be positive. | | +| `reclaimers` | Number of reclaim worker threads. Must be positive. | | +| `clean_block_threshold` | Clean disk block count at or below which reclaimers start working. Must be positive. | `16` | Example: @@ -336,6 +357,20 @@ searcher: max_num_bytes: 1G max_num_splits: 10000 num_concurrent_downloads: 1 + split_range_disk_cache: + path: /quickwit/qwdata/split-range-v1 + disk_capacity: 1500G + memory_capacity: 15G + memory_eviction_policy: s3-fifo + compression: lz4 + recover_mode: quiet + block_size: 64M + max_entry_size: 60M + flushers: 24 + buffer_pool_size: 2G + submit_queue_size_threshold: 3G + reclaimers: 8 + clean_block_threshold: 16 ``` ## Jaeger configuration diff --git a/docs/reference/metrics.md b/docs/reference/metrics.md index 33a49854895..f26013fe8b4 100644 --- a/docs/reference/metrics.md +++ b/docs/reference/metrics.md @@ -91,3 +91,9 @@ PostgreSQL-backed metastores also expose connection pool gauges: | `quickwit_storage` | `object_storage_puts_total` | Number of objects uploaded. May differ from object_storage_requests_parts due to multipart upload | `counter` | | `quickwit_storage` | `object_storage_puts_parts` | Number of object parts uploaded | `counter` | | `quickwit_storage` | `object_storage_download_num_bytes` | Amount of data downloaded from an object storage | `counter` | +| `quickwit_storage` | `split_range_disk_cache_requests_total` | Split range disk cache requests by `result` (`memory`, `disk`, `miss`, or `error`) | `counter` | +| `quickwit_storage` | `split_range_disk_cache_requested_bytes_total` | Requested bytes by `result` | `counter` | +| `quickwit_storage` | `split_range_disk_cache_admission_bypasses_total` | Entries kept memory-only, labeled by `reason` (`max_entry_size` or `encoded_too_large`) | `counter` | +| `quickwit_storage` | `split_range_disk_cache_fail_open_total` | Foyer failures served from object storage | `counter` | + +Foyer also exports its own hybrid-cache metrics on `/metrics` when `split_range_disk_cache` is enabled, including `foyer_memory_op_total`, `foyer_memory_usage`, `foyer_memory_entries`, `foyer_storage_op_total`, and `foyer_storage_disk_io_bytes_total`, labeled by cache `name` (`split-range-v1`). Object-storage GET counters cover actual remote fetches on a cache miss. diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index 630e108c2ea..8731ca5a860 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -2152,6 +2152,12 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" +[[package]] +name = "cmsketch" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7ee2cfacbd29706479902b06d75ad8f1362900836aa32799eabc7e004bfd854" + [[package]] name = "coarsetime" version = "0.1.37" @@ -2407,6 +2413,17 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "core_affinity" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a034b3a7b624016c6e13f5df875747cc25f884156aad2abd12b6c46797971342" +dependencies = [ + "libc", + "num_cpus", + "winapi", +] + [[package]] name = "cpp_demangle" version = "0.4.5" @@ -4113,6 +4130,16 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "fastant" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e825441bfb2d831c47c97d05821552db8832479f44c571b97fededbf0099c07" +dependencies = [ + "small_ctor", + "web-time", +] + [[package]] name = "fastdivide" version = "0.4.2" @@ -4283,6 +4310,117 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "foyer" +version = "0.22.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0abc0b87814989efa711f9becd9f26969820e2d3905db27d10969c4bd45890" +dependencies = [ + "anyhow", + "equivalent", + "foyer-common", + "foyer-memory", + "foyer-storage", + "foyer-tokio", + "futures-util", + "mea", + "mixtrics", + "pin-project", + "serde", + "tracing", +] + +[[package]] +name = "foyer-common" +version = "0.22.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3db80d5dece93adb7ad709c84578794724a9cba342a7e566c3551c7ec626789" +dependencies = [ + "anyhow", + "bytes", + "cfg-if", + "foyer-tokio", + "mixtrics", + "parking_lot", + "pin-project", + "twox-hash", +] + +[[package]] +name = "foyer-intrusive-collections" +version = "0.10.0-dev" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e4fee46bea69e0596130e3210e65d3424e0ac1e6df3bde6636304bdf1ca4a3b" +dependencies = [ + "memoffset", +] + +[[package]] +name = "foyer-memory" +version = "0.22.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db907f40a527ca2aa2f40a5f68b32ea58aa70f050cd233518e9ffd402cfba6ce" +dependencies = [ + "anyhow", + "bitflags 2.13.0", + "cmsketch", + "equivalent", + "foyer-common", + "foyer-intrusive-collections", + "foyer-tokio", + "futures-util", + "hashbrown 0.16.1", + "itertools 0.14.0", + "mea", + "mixtrics", + "parking_lot", + "paste", + "pin-project", + "serde", + "tracing", +] + +[[package]] +name = "foyer-storage" +version = "0.22.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1983f1db3d0710e9c9d5fc116d9202dccd41a2d1e032572224f1aff5520aa958" +dependencies = [ + "allocator-api2", + "anyhow", + "bytes", + "core_affinity", + "equivalent", + "fastant", + "foyer-common", + "foyer-memory", + "foyer-tokio", + "fs4", + "futures-core", + "futures-util", + "hashbrown 0.16.1", + "io-uring", + "itertools 0.14.0", + "libc", + "lz4", + "mea", + "parking_lot", + "pin-project", + "rand 0.9.4", + "tracing", + "twox-hash", + "zstd", +] + +[[package]] +name = "foyer-tokio" +version = "0.22.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6577b05a7ffad0db555aedf00bfe52af818220fc4c1c3a7a12520896fc38627" +dependencies = [ + "tokio", +] + [[package]] name = "fraction" version = "0.15.4" @@ -5399,6 +5537,17 @@ dependencies = [ "rustversion", ] +[[package]] +name = "io-uring" +version = "0.7.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d64d8ca234d152948ceaede1f419b6a83983a5ecccaac05fb337a809c96d3aa6" +dependencies = [ + "bitflags 2.13.0", + "cfg-if", + "libc", +] + [[package]] name = "ipcrypt-rs" version = "0.9.4" @@ -6116,6 +6265,15 @@ dependencies = [ "libc", ] +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "metrics" version = "0.24.6" @@ -6260,6 +6418,16 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "mixtrics" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c46b5adfb7a3ae4996d327a5bdc90e78fec025806dd312bdbe6f07a755e0ec9" +dependencies = [ + "itertools 0.15.0", + "parking_lot", +] + [[package]] name = "mockall" version = "0.14.0" @@ -9301,6 +9469,7 @@ dependencies = [ "serde_json", "tantivy", "tantivy-fst", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-util", @@ -9410,13 +9579,18 @@ dependencies = [ "base64 0.22.1", "bytes", "bytesize", + "fail", + "foyer", "futures", "http 1.4.2", "http-body-util", "hyper 1.10.1", "lru 0.18.0", "md5", + "metrics", + "metrics-util", "mini-moka", + "mixtrics", "mockall", "opendal", "opendal-http-transport-reqwest", @@ -11165,6 +11339,12 @@ version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" +[[package]] +name = "small_ctor" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88414a5ca1f85d82cc34471e975f0f74f6aa54c40f062efa42c0080e7f763f81" + [[package]] name = "smallvec" version = "1.15.2" @@ -12638,6 +12818,9 @@ name = "twox-hash" version = "2.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ea3136b675547379c4bd395ca6b938e5ad3c3d20fad76e7fe85f9e0d011419c" +dependencies = [ + "rand 0.9.4", +] [[package]] name = "typeid" diff --git a/quickwit/Cargo.toml b/quickwit/Cargo.toml index 8904ed1dd3c..dad8da2d47e 100644 --- a/quickwit/Cargo.toml +++ b/quickwit/Cargo.toml @@ -136,6 +136,7 @@ fail = "0.5" flate2 = "1.1" flume = "0.12" fnv = "1" +foyer = { version = "=0.22.3", default-features = false, features = ["runtime-tokio"] } futures = "0.3" futures-util = { version = "0.3", default-features = false } glob = "0.3" @@ -180,6 +181,7 @@ metrics-exporter-prometheus = { version = "0.18", default-features = false } metrics-util = "0.20" mime_guess = "2.0" mini-moka = "0.10" +mixtrics = "=0.2.5" mockall = "0.14" mrecordlog = { git = "https://github.com/quickwit-oss/mrecordlog", rev = "3b3562ef" } new_string_template = "1.5" diff --git a/quickwit/quickwit-config/src/lib.rs b/quickwit/quickwit-config/src/lib.rs index 904e65f8180..2a27bddcecc 100644 --- a/quickwit/quickwit-config/src/lib.rs +++ b/quickwit/quickwit-config/src/lib.rs @@ -77,10 +77,11 @@ pub use crate::metastore_config::{ MetastoreBackend, MetastoreConfig, MetastoreConfigs, PostgresMetastoreConfig, }; pub use crate::node_config::{ - CacheConfig, CachePolicy, CompactorConfig, DEFAULT_QW_CONFIG_PATH, GrpcConfig, HealthConfig, - IndexerConfig, IngestApiConfig, JaegerConfig, KeepAliveConfig, LambdaConfig, - LambdaDeployConfig, MAX_GOSSIP_PROTOCOL_VERSION, NodeConfig, RestConfig, SearcherConfig, - SplitCacheLimits, StorageTimeoutPolicy, TlsConfig, + CacheConfig, CachePolicy, CompactorConfig, DEFAULT_QW_CONFIG_PATH, DiskCompression, GrpcConfig, + HealthConfig, IndexerConfig, IngestApiConfig, JaegerConfig, KeepAliveConfig, LambdaConfig, + LambdaDeployConfig, MAX_GOSSIP_PROTOCOL_VERSION, NodeConfig, RecoverMode, RestConfig, + SearcherConfig, SplitCacheLimits, SplitRangeCacheWritePolicy, SplitRangeDiskCacheConfig, + StorageTimeoutPolicy, TlsConfig, }; pub use crate::serde_utils::HumanDuration; use crate::source_config::serialize::{SourceConfigV0_7, SourceConfigV0_8, VersionedSourceConfig}; diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index 1f8dd2b81e4..999e6f3e460 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -403,17 +403,125 @@ impl SplitCacheLimits { } } +/// Admission policy for the split-range Foyer cache. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum SplitRangeCacheWritePolicy { + /// Keep admitted values in memory and persist them when memory eviction occurs. + #[default] + WriteOnEviction, + /// Persist admitted values as soon as they are inserted. + WriteOnInsertion, +} + +/// On-disk compression for the split-range Foyer cache. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum DiskCompression { + Lz4, +} + +/// Recovery mode for the split-range Foyer cache. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum RecoverMode { + Quiet, +} + +/// Disabled-by-default searcher disk cache for exact split byte ranges. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SplitRangeDiskCacheConfig { + pub path: PathBuf, + #[serde(with = "crate::serde_utils::bytesize_serde")] + pub disk_capacity: ByteSize, + #[serde(with = "crate::serde_utils::bytesize_serde")] + pub memory_capacity: ByteSize, + #[serde(with = "crate::serde_utils::bytesize_serde")] + pub buffer_pool_size: ByteSize, + #[serde(with = "crate::serde_utils::bytesize_serde")] + pub submit_queue_size_threshold: ByteSize, + pub memory_eviction_policy: CachePolicy, + #[serde(default)] + pub write_policy: SplitRangeCacheWritePolicy, + pub compression: DiskCompression, + pub recover_mode: RecoverMode, + #[serde(with = "crate::serde_utils::bytesize_serde")] + pub block_size: ByteSize, + #[serde(with = "crate::serde_utils::bytesize_serde")] + pub max_entry_size: ByteSize, + pub flushers: usize, + pub reclaimers: usize, + #[serde(default = "SplitRangeDiskCacheConfig::default_clean_block_threshold")] + pub clean_block_threshold: usize, +} + +impl SplitRangeDiskCacheConfig { + fn default_clean_block_threshold() -> usize { + 16 + } + + fn validate(&self) -> anyhow::Result<()> { + if self.path.as_os_str().is_empty() { + bail!("split_range_disk_cache.path must not be empty"); + } + if self.disk_capacity.as_u64() == 0 || self.memory_capacity.as_u64() == 0 { + bail!("split range disk and memory capacities must be positive"); + } + if self.buffer_pool_size.as_u64() == 0 || self.submit_queue_size_threshold.as_u64() == 0 { + bail!("split range buffer pool and submit queue sizes must be positive"); + } + if self.block_size.as_u64() == 0 || self.max_entry_size.as_u64() == 0 { + bail!("split range block and entry sizes must be positive"); + } + if self.max_entry_size >= self.block_size { + bail!("split_range_disk_cache.max_entry_size must be smaller than block_size"); + } + if self.memory_eviction_policy != CachePolicy::S3Fifo { + bail!("split_range_disk_cache.memory_eviction_policy must be s3-fifo in phase 1"); + } + if self.flushers == 0 || self.reclaimers == 0 || self.clean_block_threshold == 0 { + bail!( + "split range disk cache flushers, reclaimers, and clean block threshold must be \ + positive" + ); + } + Ok(()) + } + + #[cfg(any(test, feature = "testsuite"))] + pub fn for_test() -> Self { + Self { + path: PathBuf::from("/tmp/quickwit-split-range-disk-cache"), + disk_capacity: ByteSize::gb(300), + memory_capacity: ByteSize::gb(1), + buffer_pool_size: ByteSize::mb(512), + submit_queue_size_threshold: ByteSize::gb(1), + memory_eviction_policy: CachePolicy::S3Fifo, + write_policy: SplitRangeCacheWritePolicy::WriteOnEviction, + compression: DiskCompression::Lz4, + recover_mode: RecoverMode::Quiet, + block_size: ByteSize::mb(16), + max_entry_size: ByteSize::mb(15), + flushers: 4, + reclaimers: 4, + clean_block_threshold: Self::default_clean_block_threshold(), + } + } +} + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields, default)] pub struct SearcherConfig { pub aggregation_memory_limit: ByteSize, pub aggregation_bucket_limit: u32, + /// Long-lived `.fast` RAM cache. Omitted is `None` and resolved in + /// [`Self::resolved_fast_field_cache`]. #[serde(alias = "fast_field_cache_capacity")] - #[serde( - deserialize_with = "CacheConfig::deserialize_with_default::<_, {ByteSize::gb(1).as_u64()}>" - )] - pub fast_field_cache: CacheConfig, + #[serde(default, deserialize_with = "deserialize_optional_fast_field_cache")] + #[serde(skip_serializing_if = "Option::is_none")] + pub fast_field_cache: Option, #[serde(alias = "split_footer_cache_capacity")] #[serde(deserialize_with = "CacheConfig::deserialize_with_default::<_, \ {ByteSize::mb(500).as_u64()}>")] @@ -437,6 +545,10 @@ pub struct SearcherConfig { // TODO document and fix if necessary. #[serde(default, skip_serializing_if = "Option::is_none")] pub split_cache: Option, + /// Process-wide Foyer disk cache for exact split footer and body ranges. + /// Omitted or `null` disables the cache. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub split_range_disk_cache: Option, #[serde(default = "SearcherConfig::default_request_timeout_secs")] request_timeout_secs: NonZeroU64, #[serde(default = "SearcherConfig::default_request_timeout_secs")] @@ -597,6 +709,13 @@ impl CacheConfig { } } +fn deserialize_optional_fast_field_cache<'de, D>( + deserializer: D, +) -> Result, D::Error> +where D: Deserializer<'de> { + CacheConfig::deserialize_with_default::(deserializer).map(Some) +} + impl From for CacheConfig { fn from(capacity: ByteSize) -> Self { CacheConfig::default_with_capacity(capacity) @@ -658,7 +777,7 @@ impl StorageTimeoutPolicy { impl Default for SearcherConfig { fn default() -> Self { SearcherConfig { - fast_field_cache: CacheConfig::default_with_capacity(ByteSize::gb(1)), + fast_field_cache: None, split_footer_cache: CacheConfig::default_with_capacity(ByteSize::mb(500)), partial_request_cache: CacheConfig::default_with_capacity(ByteSize::mb(64)), predicate_cache: CacheConfig::default_with_capacity(ByteSize::mb(256)), @@ -668,6 +787,7 @@ impl Default for SearcherConfig { aggregation_memory_limit: ByteSize::mb(500), aggregation_bucket_limit: 65000, split_cache: None, + split_range_disk_cache: None, request_timeout_secs: Self::default_request_timeout_secs(), leaf_request_timeout_secs: Self::default_request_timeout_secs(), storage_timeout_policy: None, @@ -691,6 +811,20 @@ impl SearcherConfig { fn default_request_timeout_secs() -> NonZeroU64 { NonZeroU64::new(30).unwrap() } + + /// Long-lived `.fast` RAM cache after applying defaults. + /// + /// An explicit config is used as-is. If omitted, Foyer disables the cache + /// and otherwise it is 1 GiB. + pub fn resolved_fast_field_cache(&self) -> CacheConfig { + match &self.fast_field_cache { + Some(cache_config) => cache_config.clone(), + None => match &self.split_range_disk_cache { + Some(_) => CacheConfig::no_cache(), + None => CacheConfig::default_with_capacity(ByteSize::gb(1)), + }, + } + } fn validate(&self) -> anyhow::Result<()> { if let Some(split_cache_limits) = self.split_cache { if self.max_num_concurrent_split_searches @@ -712,6 +846,9 @@ impl SearcherConfig { ); } } + if let Some(split_range_disk_cache) = &self.split_range_disk_cache { + split_range_disk_cache.validate()?; + } Ok(()) } } @@ -1320,4 +1457,181 @@ mod tests { }; assert!(grpc_config.validate().is_err()); } + + #[test] + fn test_split_range_disk_cache_config_is_disabled_by_default() { + assert!(SearcherConfig::default().split_range_disk_cache.is_none()); + assert_eq!( + SearcherConfig::default().resolved_fast_field_cache(), + CacheConfig::default_with_capacity(ByteSize::gb(1)) + ); + } + + #[test] + fn test_omitted_fast_field_cache_stays_1g_without_split_range_disk_cache() { + let config: SearcherConfig = serde_yaml::from_str("{}").unwrap(); + assert!(config.split_range_disk_cache.is_none()); + assert!(config.fast_field_cache.is_none()); + assert_eq!( + config.resolved_fast_field_cache(), + CacheConfig::default_with_capacity(ByteSize::gb(1)) + ); + } + + #[test] + fn test_omitted_fast_field_cache_is_disabled_when_split_range_disk_cache_is_set() { + let yaml = r#" +split_range_disk_cache: + path: /var/cache/quickwit/split-range-v1 + disk_capacity: 300G + memory_capacity: 1G + buffer_pool_size: 512M + submit_queue_size_threshold: 1G + memory_eviction_policy: s3-fifo + compression: lz4 + recover_mode: quiet + block_size: 16M + max_entry_size: 15M + flushers: 4 + reclaimers: 4 +"#; + let config: SearcherConfig = serde_yaml::from_str(yaml).unwrap(); + assert!(config.split_range_disk_cache.is_some()); + assert!(config.fast_field_cache.is_none()); + assert_eq!(config.resolved_fast_field_cache(), CacheConfig::no_cache()); + } + + #[test] + fn test_explicit_fast_field_cache_is_kept_when_split_range_disk_cache_is_set() { + let yaml = r#" +fast_field_cache_capacity: 1G +split_range_disk_cache: + path: /var/cache/quickwit/split-range-v1 + disk_capacity: 300G + memory_capacity: 1G + buffer_pool_size: 512M + submit_queue_size_threshold: 1G + memory_eviction_policy: s3-fifo + compression: lz4 + recover_mode: quiet + block_size: 16M + max_entry_size: 15M + flushers: 4 + reclaimers: 4 +"#; + let config: SearcherConfig = serde_yaml::from_str(yaml).unwrap(); + assert!(config.split_range_disk_cache.is_some()); + assert_eq!( + config.resolved_fast_field_cache(), + CacheConfig::default_with_capacity(ByteSize::gb(1)) + ); + } + + #[test] + fn test_explicit_zero_fast_field_cache_disables_without_split_range_disk_cache() { + let config: SearcherConfig = serde_yaml::from_str("fast_field_cache_capacity: 0").unwrap(); + assert!(config.split_range_disk_cache.is_none()); + assert_eq!( + config.resolved_fast_field_cache().capacity(), + ByteSize::b(0) + ); + } + + #[test] + fn test_split_range_disk_cache_config_round_trip() { + let yaml = r#" +split_range_disk_cache: + path: /var/cache/quickwit/split-range-v1 + disk_capacity: 300G + memory_capacity: 1G + buffer_pool_size: 512M + submit_queue_size_threshold: 1G + memory_eviction_policy: s3-fifo + compression: lz4 + recover_mode: quiet + block_size: 16M + max_entry_size: 15M + flushers: 4 + reclaimers: 4 +"#; + let config: SearcherConfig = serde_yaml::from_str(yaml).unwrap(); + let disk_cache = config.split_range_disk_cache.as_ref().unwrap(); + let serialized_disk_cache = serde_yaml::to_string(disk_cache).unwrap(); + assert!(serialized_disk_cache.contains("clean_block_threshold: 16")); + assert_eq!(disk_cache.clean_block_threshold, 16); + assert_eq!( + disk_cache.path, + PathBuf::from("/var/cache/quickwit/split-range-v1") + ); + assert_eq!(disk_cache.memory_eviction_policy, CachePolicy::S3Fifo); + assert_eq!( + disk_cache.write_policy, + SplitRangeCacheWritePolicy::WriteOnEviction + ); + assert_eq!(disk_cache.block_size, ByteSize::mb(16)); + assert_eq!(disk_cache.max_entry_size, ByteSize::mb(15)); + // Round-trip the nested cache config. SearcherConfig's other ByteSize + // fields serialize as display strings and do not round-trip exactly. + assert_eq!( + serde_yaml::from_str::(&serialized_disk_cache).unwrap(), + *disk_cache + ); + } + + #[test] + fn test_split_range_disk_cache_accepts_write_on_insertion() { + let yaml = r#" +split_range_disk_cache: + path: /var/cache/quickwit/split-range-v1 + disk_capacity: 300G + memory_capacity: 1G + buffer_pool_size: 512M + submit_queue_size_threshold: 1G + memory_eviction_policy: s3-fifo + write_policy: write-on-insertion + compression: lz4 + recover_mode: quiet + block_size: 16M + max_entry_size: 15M + flushers: 4 + reclaimers: 4 +"#; + let config: SearcherConfig = serde_yaml::from_str(yaml).unwrap(); + assert_eq!( + config.split_range_disk_cache.unwrap().write_policy, + SplitRangeCacheWritePolicy::WriteOnInsertion + ); + } + + #[test] + fn test_split_range_disk_cache_rejects_invalid_sizes() { + let mut disk_cache = SplitRangeDiskCacheConfig::for_test(); + disk_cache.max_entry_size = ByteSize::mb(16); + let config = SearcherConfig { + split_range_disk_cache: Some(disk_cache), + ..Default::default() + }; + let error = config.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("max_entry_size must be smaller than block_size") + ); + } + + #[test] + fn test_split_range_disk_cache_rejects_zero_clean_block_threshold() { + let mut disk_cache = SplitRangeDiskCacheConfig::for_test(); + disk_cache.clean_block_threshold = 0; + let config = SearcherConfig { + split_range_disk_cache: Some(disk_cache), + ..Default::default() + }; + let error = config.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("clean block threshold must be positive") + ); + } } diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index 48fa2854509..60e8f518c08 100644 --- a/quickwit/quickwit-config/src/node_config/serialize.rs +++ b/quickwit/quickwit-config/src/node_config/serialize.rs @@ -883,7 +883,7 @@ mod tests { SearcherConfig { aggregation_memory_limit: ByteSize::gb(1), aggregation_bucket_limit: 500_000, - fast_field_cache: CacheConfig::default_with_capacity(ByteSize::gb(10)), + fast_field_cache: Some(CacheConfig::default_with_capacity(ByteSize::gb(10))), split_footer_cache: CacheConfig::default_with_capacity(ByteSize::gb(1)), partial_request_cache: CacheConfig::default_with_capacity(ByteSize::mb(64)), predicate_cache: CacheConfig::default_with_capacity(ByteSize::mb(256)), @@ -891,6 +891,7 @@ mod tests { max_splits_per_search: None, _max_num_concurrent_split_streams: Some(serde::de::IgnoredAny), split_cache: None, + split_range_disk_cache: None, request_timeout_secs: NonZeroU64::new(30).unwrap(), leaf_request_timeout_secs: NonZeroU64::new(30).unwrap(), storage_timeout_policy: Some(crate::StorageTimeoutPolicy { diff --git a/quickwit/quickwit-lambda-server/src/context.rs b/quickwit/quickwit-lambda-server/src/context.rs index d3b9167414f..579d01a6036 100644 --- a/quickwit/quickwit-lambda-server/src/context.rs +++ b/quickwit/quickwit-lambda-server/src/context.rs @@ -33,8 +33,11 @@ impl LambdaSearcherContext { info!("initializing lambda searcher context"); let searcher_config = try_searcher_config_from_env()?; - let searcher_context = - Arc::new(SearcherContext::new_without_invoker(searcher_config, None)); + let searcher_context = Arc::new(SearcherContext::new_without_invoker( + searcher_config, + None, + None, + )); let storage_resolver = StorageResolver::configured(&Default::default()); Ok(Self { @@ -62,7 +65,7 @@ fn try_searcher_config_from_env() -> anyhow::Result { let mut searcher_config = SearcherConfig::default(); searcher_config.max_num_concurrent_split_searches = 20; searcher_config.warmup_memory_budget = warmup_memory_budget; - searcher_config.fast_field_cache = CacheConfig::no_cache(); + searcher_config.fast_field_cache = Some(CacheConfig::no_cache()); searcher_config.split_footer_cache = CacheConfig::no_cache(); searcher_config.predicate_cache = CacheConfig::no_cache(); searcher_config.partial_request_cache = CacheConfig::no_cache(); diff --git a/quickwit/quickwit-search/Cargo.toml b/quickwit/quickwit-search/Cargo.toml index cbbe2b269d5..4c95fc62214 100644 --- a/quickwit/quickwit-search/Cargo.toml +++ b/quickwit/quickwit-search/Cargo.toml @@ -53,6 +53,7 @@ assert-json-diff = { workspace = true } proptest = { workspace = true } rand = { workspace = true } serde_json = { workspace = true } +tempfile = { workspace = true } quickwit-indexing = { workspace = true, features = ["testsuite"] } quickwit-metastore = { workspace = true, features = ["testsuite"] } diff --git a/quickwit/quickwit-search/src/leaf.rs b/quickwit/quickwit-search/src/leaf.rs index 99ff06c312b..36ae3986f9c 100644 --- a/quickwit/quickwit-search/src/leaf.rs +++ b/quickwit/quickwit-search/src/leaf.rs @@ -46,6 +46,7 @@ use quickwit_query::tokenizers::TokenizerManager; use quickwit_storage::{ BundleStorage, ByteRangeCache, CountingStorage, MemorySizedCache, OwnedBytes, SearchSplitCache, Storage, StorageResolver, TimeoutAndRetryStorage, wrap_storage_with_cache, + wrap_storage_with_split_range_cache, }; use tantivy::aggregation::AggContextParams; use tantivy::aggregation::agg_req::{AggregationVariants, Aggregations}; @@ -160,16 +161,19 @@ async fn get_split_footer_from_cache_or_fetch( Ok(footer_data_opt) } -/// Returns hotcache_bytes and the split directory (`BundleStorage`) with cache layer: -/// - A split footer cache given by `SearcherContext.split_footer_cache`. +/// Returns hotcache_bytes and the split directory (`BundleStorage`). pub(crate) async fn open_split_bundle( searcher_context: &SearcherContext, index_storage: Arc, split_and_footer_offsets: &SplitIdAndFooterOffsets, ) -> anyhow::Result<(FileSlice, BundleStorage)> { let split_file = PathBuf::from(format!("{}.split", split_and_footer_offsets.split_id)); + let foyer_storage: Arc = match &searcher_context.split_range_disk_cache_opt { + Some(cache) => wrap_storage_with_split_range_cache(cache.clone(), index_storage.clone()), + None => index_storage.clone(), + }; let footer_data = get_split_footer_from_cache_or_fetch( - index_storage.clone(), + foyer_storage.clone(), split_and_footer_offsets, &searcher_context.split_footer_cache, ) @@ -179,9 +183,9 @@ pub(crate) async fn open_split_bundle( // This is before the bundle storage: at this point, this storage is reading `.split` files. let index_storage_with_split_cache = if let Some(split_cache) = searcher_context.split_cache_opt.as_ref() { - SearchSplitCache::wrap_storage(split_cache.clone(), index_storage.clone()) + SearchSplitCache::wrap_storage(split_cache.clone(), foyer_storage) } else { - index_storage.clone() + foyer_storage }; let (hotcache_bytes, bundle_storage) = BundleStorage::open_from_split_data( @@ -3113,7 +3117,8 @@ mod tests { offload_threshold: 3, ..LambdaConfig::for_test() }); - let searcher_context = SearcherContext::new(config, None, Some(Arc::new(DummyInvoker))); + let searcher_context = + SearcherContext::new(config, None, None, Some(Arc::new(DummyInvoker))); let splits = make_splits_with_requests(7); let result = super::schedule_search_tasks(splits, &searcher_context).await; assert_eq!(result.local_search_tasks.len(), 3); @@ -3133,7 +3138,8 @@ mod tests { offload_threshold: 0, ..LambdaConfig::for_test() }); - let searcher_context = SearcherContext::new(config, None, Some(Arc::new(DummyInvoker))); + let searcher_context = + SearcherContext::new(config, None, None, Some(Arc::new(DummyInvoker))); let splits = make_splits_with_requests(5); let result = super::schedule_search_tasks(splits, &searcher_context).await; assert!(result.local_search_tasks.is_empty()); @@ -3147,7 +3153,8 @@ mod tests { offload_threshold: 100, ..LambdaConfig::for_test() }); - let searcher_context = SearcherContext::new(config, None, Some(Arc::new(DummyInvoker))); + let searcher_context = + SearcherContext::new(config, None, None, Some(Arc::new(DummyInvoker))); let splits = make_splits_with_requests(5); let result = super::schedule_search_tasks(splits, &searcher_context).await; assert_eq!(result.local_search_tasks.len(), 5); diff --git a/quickwit/quickwit-search/src/lib.rs b/quickwit/quickwit-search/src/lib.rs index 2d891dbfa65..9d15c7f6c12 100644 --- a/quickwit/quickwit-search/src/lib.rs +++ b/quickwit/quickwit-search/src/lib.rs @@ -42,6 +42,8 @@ pub(crate) mod top_k_collector; mod metrics; mod search_permit_provider; +#[cfg(test)] +mod split_range_cache_layer_tests; #[cfg(test)] mod tests; @@ -289,7 +291,11 @@ 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_without_invoker(searcher_config, None)); + let searcher_context = Arc::new(SearcherContext::new_without_invoker( + 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 52bfc846696..2647f7e6edb 100644 --- a/quickwit/quickwit-search/src/service.rs +++ b/quickwit/quickwit-search/src/service.rs @@ -29,7 +29,8 @@ use quickwit_proto::search::{ SearchPlanResponse, SearchRequest, SearchResponse, SnippetRequest, }; use quickwit_storage::{ - MemorySizedCache, QuickwitCache, SearchSplitCache, StorageCache, StorageResolver, + FoyerSplitRangeCache, MemorySizedCache, QuickwitCache, SearchSplitCache, StorageCache, + StorageResolver, }; use tantivy::aggregation::AggregationLimitsGuard; @@ -417,6 +418,8 @@ pub struct SearcherContext { pub predicate_cache: Arc, /// Search split cache. `None` if no split cache is configured. pub split_cache_opt: Option>, + /// Process-wide split range disk cache. `None` if not configured. + pub split_range_disk_cache_opt: Option>, /// List fields cache. Caches the raw fields-metadata blob for a given split. pub list_fields_cache: ListFieldsCache, /// The aggregation limits are passed to limit the memory usage. @@ -439,17 +442,19 @@ impl SearcherContext { #[cfg(test)] pub fn for_test() -> SearcherContext { let searcher_config = SearcherConfig::default(); - SearcherContext::new_without_invoker(searcher_config, None) + SearcherContext::new_without_invoker(searcher_config, None, None) } /// Creates a new searcher context without a lambda invoker. pub fn new_without_invoker( searcher_config: SearcherConfig, split_cache_opt: Option>, + split_range_disk_cache_opt: Option>, ) -> Self { Self::new( searcher_config, split_cache_opt, + split_range_disk_cache_opt, None::>, ) } @@ -458,6 +463,7 @@ impl SearcherContext { pub fn new( searcher_config: SearcherConfig, split_cache_opt: Option>, + split_range_disk_cache_opt: Option>, lambda_invoker: Option, ) -> Self { let global_split_footer_cache = MemorySizedCache::from_config( @@ -468,8 +474,12 @@ impl SearcherContext { searcher_config.max_num_concurrent_split_searches, searcher_config.warmup_memory_budget, ); - let storage_long_term_cache = - Arc::new(QuickwitCache::new(&searcher_config.fast_field_cache)); + let fast_field_cache = searcher_config.resolved_fast_field_cache(); + let storage_long_term_cache = if fast_field_cache.capacity().as_u64() == 0 { + Arc::new(QuickwitCache::empty()) + } else { + Arc::new(QuickwitCache::new(&fast_field_cache)) + }; let leaf_search_cache = LeafSearchCache::new(&searcher_config.partial_request_cache); let predicate_cache = PredicateCacheImpl::new(&searcher_config.predicate_cache); let list_fields_cache = ListFieldsCache::new(&searcher_config.partial_request_cache); @@ -490,6 +500,7 @@ impl SearcherContext { leaf_search_cache, list_fields_cache, split_cache_opt, + split_range_disk_cache_opt, aggregation_limit, lambda_invoker, } @@ -500,3 +511,52 @@ impl SearcherContext { self.aggregation_limit.clone() } } + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + + use quickwit_config::{CacheConfig, SearcherConfig}; + use quickwit_storage::OwnedBytes; + + use super::SearcherContext; + + #[tokio::test] + async fn test_zero_capacity_fast_field_cache_does_not_retain_entries() { + let mut searcher_config = SearcherConfig::default(); + searcher_config.fast_field_cache = Some(CacheConfig::no_cache()); + let searcher_context = SearcherContext::new_without_invoker(searcher_config, None, None); + let path = PathBuf::from("segment.fast"); + searcher_context + .fast_fields_cache + .put(path.clone(), 0..3, OwnedBytes::new(&b"abc"[..])) + .await; + assert!( + searcher_context + .fast_fields_cache + .get(path.as_path(), 0..3) + .await + .is_none() + ); + } + + #[tokio::test] + async fn test_nonzero_fast_field_cache_retains_entries() { + let searcher_context = + SearcherContext::new_without_invoker(SearcherConfig::default(), None, None); + let path = PathBuf::from("segment.fast"); + searcher_context + .fast_fields_cache + .put(path.clone(), 0..3, OwnedBytes::new(&b"abc"[..])) + .await; + assert_eq!( + searcher_context + .fast_fields_cache + .get(path.as_path(), 0..3) + .await + .unwrap() + .as_slice(), + b"abc" + ); + } +} diff --git a/quickwit/quickwit-search/src/split_range_cache_layer_tests.rs b/quickwit/quickwit-search/src/split_range_cache_layer_tests.rs new file mode 100644 index 00000000000..8379e6bce0f --- /dev/null +++ b/quickwit/quickwit-search/src/split_range_cache_layer_tests.rs @@ -0,0 +1,226 @@ +// 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::num::NonZeroU32; +use std::path::Path; +use std::sync::Arc; + +use bytesize::ByteSize; +use quickwit_config::{ + CachePolicy, DiskCompression, RecoverMode, SearcherConfig, SplitCacheLimits, + SplitRangeCacheWritePolicy, SplitRangeDiskCacheConfig, +}; +use quickwit_proto::search::SplitIdAndFooterOffsets; +use quickwit_storage::{ + CountingStorage, DownloadCounters, FoyerSplitRangeCache, OwnedBytes, PutPayload, + RamStorageBuilder, SearchSplitCache, SplitPayloadBuilder, Storage, StorageResolver, +}; + +use crate::SearcherContext; +use crate::leaf::open_split_bundle; + +const SPLIT_ID: &str = "range-cache-split"; +const BODY_FILE: &str = "segment.fast"; +const BODY_BYTES: &[u8] = b"FASTDATA"; +const HOTCACHE_BYTES: &[u8] = b"HOT"; + +fn range_cache_config(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, + clean_block_threshold: 16, + } +} + +fn lower_reads(counters: &DownloadCounters) -> u64 { + counters.snapshot().1 +} + +struct SplitBundle { + split_bytes: OwnedBytes, + footer_offsets: SplitIdAndFooterOffsets, +} + +async fn build_split() -> SplitBundle { + let temp_dir = tempfile::tempdir().unwrap(); + let body_path = temp_dir.path().join(BODY_FILE); + std::fs::write(&body_path, BODY_BYTES).unwrap(); + let payload = + SplitPayloadBuilder::get_split_payload(&[body_path], &[], HOTCACHE_BYTES).unwrap(); + let footer_range = payload.footer_range.clone(); + let split_bytes = payload.read_all().await.unwrap(); + SplitBundle { + split_bytes, + footer_offsets: SplitIdAndFooterOffsets { + split_id: SPLIT_ID.to_string(), + split_footer_start: footer_range.start, + split_footer_end: footer_range.end, + ..Default::default() + }, + } +} + +fn split_file_name() -> String { + format!("{SPLIT_ID}.split") +} + +async fn open_range_cache(dir: &Path) -> Arc { + Arc::new( + FoyerSplitRangeCache::open(&range_cache_config(dir)) + .await + .unwrap(), + ) +} + +fn wrap_counted_ram(split_bytes: &OwnedBytes) -> (Arc, Arc) { + let ram = RamStorageBuilder::default() + .put(&split_file_name(), split_bytes.as_slice()) + .build(); + CountingStorage::instrument_storage(Arc::new(ram)) +} + +fn context_with_range_cache(cache: Arc) -> SearcherContext { + SearcherContext::new_without_invoker(SearcherConfig::default(), None, Some(cache)) +} + +#[tokio::test] +async fn test_open_split_bundle_footer_ram_hit_bypasses_lower_tiers() { + let split = build_split().await; + let cache_dir = tempfile::tempdir().unwrap(); + let cache = open_range_cache(cache_dir.path()).await; + let context = context_with_range_cache(cache.clone()); + let footer = split.split_bytes.slice( + split.footer_offsets.split_footer_start as usize + ..split.footer_offsets.split_footer_end as usize, + ); + context.split_footer_cache.put(SPLIT_ID.to_string(), footer); + + let (counted, counters) = wrap_counted_ram(&split.split_bytes); + let (_hotcache, _bundle) = open_split_bundle(&context, counted, &split.footer_offsets) + .await + .unwrap(); + assert_eq!(lower_reads(&counters), 0); + cache.close().await.unwrap(); +} + +#[tokio::test] +async fn test_open_split_bundle_footer_miss_uses_foyer_then_reuses_storage_for_body() { + let split = build_split().await; + let cache_dir = tempfile::tempdir().unwrap(); + let cache = open_range_cache(cache_dir.path()).await; + let context = context_with_range_cache(cache.clone()); + let (counted, counters) = wrap_counted_ram(&split.split_bytes); + + let (_hotcache, bundle) = open_split_bundle(&context, counted, &split.footer_offsets) + .await + .unwrap(); + assert_eq!(lower_reads(&counters), 1, "cold footer is one lower read"); + + bundle.get_slice(Path::new(BODY_FILE), 0..4).await.unwrap(); + bundle.get_slice(Path::new(BODY_FILE), 0..4).await.unwrap(); + assert_eq!( + lower_reads(&counters), + 2, + "second exact body range must hit Foyer" + ); + cache.close().await.unwrap(); +} + +#[tokio::test] +async fn test_open_split_bundle_footer_skips_whole_split_cache() { + let split = build_split().await; + let split_cache_dir = tempfile::tempdir().unwrap(); + std::fs::write( + split_cache_dir.path().join(split_file_name()), + split.split_bytes.as_slice(), + ) + .unwrap(); + let split_cache = SearchSplitCache::with_root_path( + split_cache_dir.path().to_path_buf(), + StorageResolver::unconfigured(), + SplitCacheLimits { + max_num_bytes: ByteSize::mb(64), + max_num_splits: NonZeroU32::new(8).unwrap(), + num_concurrent_downloads: NonZeroU32::new(1).unwrap(), + max_file_descriptors: NonZeroU32::new(8).unwrap(), + }, + ) + .unwrap(); + + let range_cache_dir = tempfile::tempdir().unwrap(); + let range_cache = open_range_cache(range_cache_dir.path()).await; + let context = SearcherContext::new_without_invoker( + SearcherConfig::default(), + Some(split_cache), + Some(range_cache.clone()), + ); + let (counted, counters) = wrap_counted_ram(&split.split_bytes); + let (_hotcache, bundle) = open_split_bundle(&context, counted, &split.footer_offsets) + .await + .unwrap(); + assert_eq!( + lower_reads(&counters), + 1, + "footer fetch skips SplitCache and reads through Foyer" + ); + + bundle.get_slice(Path::new(BODY_FILE), 0..4).await.unwrap(); + assert_eq!( + lower_reads(&counters), + 1, + "body read must hit the on-disk whole-split cache" + ); + range_cache.close().await.unwrap(); +} + +#[tokio::test] +async fn test_open_split_bundle_recovers_footer_from_foyer() { + let split = build_split().await; + let cache_dir = tempfile::tempdir().unwrap(); + let config = range_cache_config(cache_dir.path()); + { + let cache = Arc::new(FoyerSplitRangeCache::open(&config).await.unwrap()); + let context = context_with_range_cache(cache.clone()); + let (counted, counters) = wrap_counted_ram(&split.split_bytes); + open_split_bundle(&context, counted, &split.footer_offsets) + .await + .unwrap(); + assert_eq!(lower_reads(&counters), 1); + cache.close().await.unwrap(); + } + + let recovered = Arc::new(FoyerSplitRangeCache::open(&config).await.unwrap()); + let context = context_with_range_cache(recovered.clone()); + let (counted, counters) = wrap_counted_ram(&split.split_bytes); + open_split_bundle(&context, counted, &split.footer_offsets) + .await + .unwrap(); + assert_eq!( + lower_reads(&counters), + 0, + "recovered footer range must not read lower storage" + ); + recovered.close().await.unwrap(); +} diff --git a/quickwit/quickwit-search/src/tests.rs b/quickwit/quickwit-search/src/tests.rs index 94183d16c2a..63c6c3ed08a 100644 --- a/quickwit/quickwit-search/src/tests.rs +++ b/quickwit/quickwit-search/src/tests.rs @@ -1031,6 +1031,7 @@ async fn test_search_util(test_sandbox: &TestSandbox, query: &str) -> Vec { let searcher_context: Arc = Arc::new(SearcherContext::new_without_invoker( SearcherConfig::default(), None, + None, )); let search_response = single_doc_mapping_leaf_search( @@ -1671,6 +1672,7 @@ async fn test_single_node_list_terms() -> anyhow::Result<()> { let searcher_context = Arc::new(SearcherContext::new_without_invoker( SearcherConfig::default(), None, + None, )); { diff --git a/quickwit/quickwit-serve/src/lib.rs b/quickwit/quickwit-serve/src/lib.rs index dd45d678d95..951ea64e111 100644 --- a/quickwit/quickwit-serve/src/lib.rs +++ b/quickwit/quickwit-serve/src/lib.rs @@ -119,7 +119,7 @@ use quickwit_search::{ SearchJobPlacer, SearchService, SearchServiceClient, SearcherContext, SearcherPool, create_search_client_from_channel, start_searcher_service, }; -use quickwit_storage::{SearchSplitCache, StorageResolver}; +use quickwit_storage::{FoyerSplitRangeCache, SearchSplitCache, StorageResolver}; pub use quickwit_telemetry_exporters::{EnvFilterReloadFn, do_nothing_env_filter_reload_fn}; pub use quickwit_transport::reload_tls_cert; use tcp_listener::TcpListenerResolver; @@ -729,6 +729,20 @@ pub async fn serve_quickwit( None }; + let split_range_disk_cache_opt = if node_config.is_service_enabled(QuickwitService::Searcher) { + match &node_config.searcher_config.split_range_disk_cache { + Some(config) => Some(Arc::new( + FoyerSplitRangeCache::open(config) + .await + .context("failed to open searcher split range disk cache")?, + )), + None => None, + } + } else { + None + }; + let split_range_disk_cache_for_shutdown = split_range_disk_cache_opt.clone(); + // Initialize Lambda invoker if enabled and searcher service is running let searcher_context = if node_config.is_service_enabled(QuickwitService::Searcher) { if let Some(lambda_config) = &node_config.searcher_config.lambda { @@ -741,6 +755,7 @@ pub async fn serve_quickwit( Arc::new(SearcherContext::new( node_config.searcher_config.clone(), search_split_cache_opt, + split_range_disk_cache_opt, Some(invoker), )) } @@ -753,12 +768,14 @@ pub async fn serve_quickwit( Arc::new(SearcherContext::new_without_invoker( node_config.searcher_config.clone(), search_split_cache_opt, + split_range_disk_cache_opt, )) } } else { Arc::new(SearcherContext::new_without_invoker( node_config.searcher_config.clone(), search_split_cache_opt, + split_range_disk_cache_opt, )) }; @@ -1077,6 +1094,12 @@ pub async fn serve_quickwit( let actor_exit_statuses = shutdown_handle .await .context("failed to gracefully shutdown services")?; + if let Some(cache) = split_range_disk_cache_for_shutdown { + cache + .close() + .await + .context("failed to close searcher split range disk cache")?; + } Ok(actor_exit_statuses) } @@ -2086,6 +2109,7 @@ mod tests { let searcher_context = Arc::new(SearcherContext::new_without_invoker( SearcherConfig::default(), None, + None, )); let metastore = metastore_for_test(); let (change_stream, change_stream_tx) = ClusterChangeStream::new_unbounded(); diff --git a/quickwit/quickwit-storage/Cargo.toml b/quickwit/quickwit-storage/Cargo.toml index 620e44b9ff7..52231bd2970 100644 --- a/quickwit/quickwit-storage/Cargo.toml +++ b/quickwit/quickwit-storage/Cargo.toml @@ -17,12 +17,15 @@ async-trait = { workspace = true } base64 = { workspace = true } bytes = { workspace = true } bytesize = { workspace = true } +foyer = { workspace = true } futures = { workspace = true } http-body-util = { workspace = true} hyper = { workspace = true } lru = { workspace = true } md5 = { workspace = true } +metrics = { workspace = true } mini-moka = { workspace = true } +mixtrics = { workspace = true } mockall = { workspace = true, optional = true } pin-project = { workspace = true } quick_cache = { workspace = true } @@ -34,6 +37,7 @@ stable_deref_trait = { workspace = true } tantivy = { workspace = true } tempfile = { workspace = true } thiserror = { workspace = true } +fail = { workspace = true } tokio = { workspace = true, features = ["test-util"] } tokio-stream = { workspace = true } tokio-util = { workspace = true } @@ -61,6 +65,7 @@ reqwest = { workspace = true, optional = true } [dev-dependencies] http = { workspace = true } +metrics-util = { workspace = true } mockall = { workspace = true } proptest = { workspace = true } # Match OpenDAL's internal reqwest major. `default-features = false` is @@ -94,6 +99,7 @@ azure = [ ] gcs = ["dep:opendal", "opendal/services-gcs"] ci-test = [] +failpoints = ["fail/failpoints"] integration-testsuite = [ "azure", "azure_core/azurite_workaround", diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs index 13bc7069f60..894a18711fe 100644 --- a/quickwit/quickwit-storage/src/lib.rs +++ b/quickwit/quickwit-storage/src/lib.rs @@ -50,6 +50,7 @@ mod prefix_storage; mod ram_storage; mod split; mod split_cache; +mod split_range_cache; mod stable_deref_bytes; mod storage_factory; mod storage_resolver; @@ -57,6 +58,9 @@ mod versioned_component; use quickwit_common::uri::Uri; pub use split_cache::SearchSplitCache; +pub use split_range_cache::{ + FoyerSplitRangeCache, FoyerSplitRangeStorage, wrap_storage_with_split_range_cache, +}; pub use tantivy::directory::OwnedBytes; pub use versioned_component::VersionedComponent; diff --git a/quickwit/quickwit-storage/src/split_range_cache/key.rs b/quickwit/quickwit-storage/src/split_range_cache/key.rs new file mode 100644 index 00000000000..1bac1c61bfe --- /dev/null +++ b/quickwit/quickwit-storage/src/split_range_cache/key.rs @@ -0,0 +1,130 @@ +// 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::io::{Read, Write}; +use std::ops::Range; + +/// Exact cache key for a split object URI and a half-open byte range. +/// +/// The on-disk layout is little-endian `u64` fields: +/// `[uri_len][uri_bytes][range_start][range_end]`. +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +pub struct SplitRangeCacheKey { + pub object_uri: String, + pub byte_range: Range, +} + +const U64_LEN: usize = 8; + +fn write_u64(writer: &mut impl Write, value: u64) -> foyer::Result<()> { + writer + .write_all(&value.to_le_bytes()) + .map_err(foyer::Error::io_error) +} + +fn read_u64(reader: &mut impl Read) -> foyer::Result { + let mut buf = [0u8; U64_LEN]; + reader + .read_exact(&mut buf) + .map_err(foyer::Error::io_error)?; + Ok(u64::from_le_bytes(buf)) +} + +/// Manual `Code` impl: keep Foyer's serde feature off so bincode is not pulled +/// in for keys or values. Foyer 0.22.3 already implements `Code` for `Bytes`, +/// so only this key needs a codec. Enabling serde would encode each value byte +/// as an integer; for a 15 MiB payload that is millions of serializer visits +/// instead of one `write_all`. +impl foyer::Code for SplitRangeCacheKey { + fn encode(&self, writer: &mut impl Write) -> foyer::Result<()> { + let uri_bytes = self.object_uri.as_bytes(); + write_u64(writer, uri_bytes.len() as u64)?; + writer + .write_all(uri_bytes) + .map_err(foyer::Error::io_error)?; + write_u64(writer, self.byte_range.start as u64)?; + write_u64(writer, self.byte_range.end as u64) + } + + fn decode(reader: &mut impl Read) -> foyer::Result { + let uri_len = read_u64(reader)? as usize; + let mut uri_bytes = vec![0; uri_len]; + reader + .read_exact(&mut uri_bytes) + .map_err(foyer::Error::io_error)?; + let object_uri = String::from_utf8(uri_bytes).map_err(|error| { + foyer::Error::new(foyer::ErrorKind::Parse, "object URI is not UTF-8").with_source(error) + })?; + let start = read_u64(reader)? as usize; + let end = read_u64(reader)? as usize; + Ok(Self { + object_uri, + byte_range: start..end, + }) + } + + fn estimated_size(&self) -> usize { + U64_LEN + self.object_uri.len() + U64_LEN + U64_LEN + } +} + +#[cfg(test)] +mod tests { + use foyer::Code; + + use super::*; + + #[test] + fn test_split_range_cache_key_codec_round_trip() { + let expected = SplitRangeCacheKey { + object_uri: "s3://bucket/prefix/a.split".to_string(), + byte_range: 10..42, + }; + let mut encoded = Vec::new(); + expected.encode(&mut encoded).unwrap(); + assert_eq!( + SplitRangeCacheKey::decode(&mut encoded.as_slice()).unwrap(), + expected + ); + assert_eq!(expected.estimated_size(), encoded.len()); + } + + #[test] + fn test_split_range_cache_key_codec_empty_uri() { + let expected = SplitRangeCacheKey { + object_uri: String::new(), + byte_range: 0..0, + }; + let mut encoded = Vec::new(); + expected.encode(&mut encoded).unwrap(); + assert_eq!( + SplitRangeCacheKey::decode(&mut encoded.as_slice()).unwrap(), + expected + ); + assert_eq!(encoded.len(), 24); + } + + #[test] + fn test_split_range_cache_key_codec_rejects_truncated_buffer() { + let key = SplitRangeCacheKey { + object_uri: "s3://bucket/a.split".to_string(), + byte_range: 1..2, + }; + let mut encoded = Vec::new(); + key.encode(&mut encoded).unwrap(); + encoded.pop(); + let error = SplitRangeCacheKey::decode(&mut encoded.as_slice()).unwrap_err(); + assert_eq!(error.kind(), foyer::ErrorKind::Io); + } +} diff --git a/quickwit/quickwit-storage/src/split_range_cache/metrics.rs b/quickwit/quickwit-storage/src/split_range_cache/metrics.rs new file mode 100644 index 00000000000..d8af2705864 --- /dev/null +++ b/quickwit/quickwit-storage/src/split_range_cache/metrics.rs @@ -0,0 +1,347 @@ +// 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::borrow::Cow; +use std::fmt; + +use mixtrics::metrics::{ + BoxedCounter, BoxedCounterVec, BoxedGauge, BoxedGaugeVec, BoxedHistogram, BoxedHistogramVec, + CounterOps, CounterVecOps, GaugeOps, GaugeVecOps, HistogramOps, HistogramVecOps, RegistryOps, +}; +use quickwit_metrics::{LazyCounter, label_names, label_values, lazy_counter}; + +use super::storage::AdmissionBypass; + +const CACHE_RESULT: quickwit_metrics::LabelNames<1> = label_names!("result"); +const ADMISSION_REASON: quickwit_metrics::LabelNames<1> = label_names!("reason"); + +static REQUESTS: LazyCounter = lazy_counter!( + name: "split_range_disk_cache_requests_total", + description: "Split range disk cache requests by result", + subsystem: "storage", +); +static REQUESTED_BYTES: LazyCounter = lazy_counter!( + name: "split_range_disk_cache_requested_bytes_total", + description: "Split range disk cache requested bytes by result", + subsystem: "storage", +); +static ADMISSION_BYPASSES: LazyCounter = lazy_counter!( + name: "split_range_disk_cache_admission_bypasses_total", + description: "Entries kept memory-only by admission checks", + subsystem: "storage", +); +static FAIL_OPEN_TOTAL: LazyCounter = lazy_counter!( + name: "split_range_disk_cache_fail_open_total", + description: "Foyer failures bypassed through lower storage", + subsystem: "storage", +); + +pub(crate) static REQUESTS_MEMORY: LazyCounter = lazy_counter!( + parent: REQUESTS, + labels: [label_values!(CACHE_RESULT => "memory")] +); +pub(crate) static REQUESTS_DISK: LazyCounter = lazy_counter!( + parent: REQUESTS, + labels: [label_values!(CACHE_RESULT => "disk")] +); +pub(crate) static REQUESTS_MISS: LazyCounter = lazy_counter!( + parent: REQUESTS, + labels: [label_values!(CACHE_RESULT => "miss")] +); +pub(crate) static REQUESTS_ERROR: LazyCounter = lazy_counter!( + parent: REQUESTS, + labels: [label_values!(CACHE_RESULT => "error")] +); +static REQUESTED_BYTES_MEMORY: LazyCounter = lazy_counter!( + parent: REQUESTED_BYTES, + labels: [label_values!(CACHE_RESULT => "memory")] +); +static REQUESTED_BYTES_DISK: LazyCounter = lazy_counter!( + parent: REQUESTED_BYTES, + labels: [label_values!(CACHE_RESULT => "disk")] +); +static REQUESTED_BYTES_MISS: LazyCounter = lazy_counter!( + parent: REQUESTED_BYTES, + labels: [label_values!(CACHE_RESULT => "miss")] +); +static REQUESTED_BYTES_ERROR: LazyCounter = lazy_counter!( + parent: REQUESTED_BYTES, + labels: [label_values!(CACHE_RESULT => "error")] +); +pub(crate) static ADMISSION_MAX_ENTRY_SIZE: LazyCounter = lazy_counter!( + parent: ADMISSION_BYPASSES, + labels: [label_values!(ADMISSION_REASON => "max_entry_size")] +); +pub(crate) static ADMISSION_ENCODED_TOO_LARGE: LazyCounter = lazy_counter!( + parent: ADMISSION_BYPASSES, + labels: [label_values!(ADMISSION_REASON => "encoded_too_large")] +); + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum FetchOutcome { + MemoryHit, + DiskHit, + RemoteMiss, + Error, +} + +pub(crate) fn record_request(outcome: FetchOutcome, num_bytes: u64) { + match outcome { + FetchOutcome::MemoryHit => { + REQUESTS_MEMORY.inc(); + REQUESTED_BYTES_MEMORY.inc_by(num_bytes); + } + FetchOutcome::DiskHit => { + REQUESTS_DISK.inc(); + REQUESTED_BYTES_DISK.inc_by(num_bytes); + } + FetchOutcome::RemoteMiss => { + REQUESTS_MISS.inc(); + REQUESTED_BYTES_MISS.inc_by(num_bytes); + } + FetchOutcome::Error => { + REQUESTS_ERROR.inc(); + REQUESTED_BYTES_ERROR.inc_by(num_bytes); + } + } +} + +pub(crate) fn record_admission_bypass(reason: AdmissionBypass) { + match reason { + AdmissionBypass::MaxEntrySize => ADMISSION_MAX_ENTRY_SIZE.inc(), + AdmissionBypass::EncodedTooLarge => ADMISSION_ENCODED_TOO_LARGE.inc(), + } +} + +pub(crate) fn record_fail_open() { + FAIL_OPEN_TOTAL.inc(); +} + +/// Mixtrics registry that forwards Foyer metrics to the process `metrics` recorder. +#[derive(Debug)] +pub(crate) struct QuickwitMetricsRegistry; + +impl RegistryOps for QuickwitMetricsRegistry { + fn register_counter_vec( + &self, + name: Cow<'static, str>, + desc: Cow<'static, str>, + label_names: &'static [&'static str], + ) -> BoxedCounterVec { + ::metrics::describe_counter!(name.clone(), desc.clone()); + Box::new(MetricsCounterVec { name, label_names }) + } + + fn register_gauge_vec( + &self, + name: Cow<'static, str>, + desc: Cow<'static, str>, + label_names: &'static [&'static str], + ) -> BoxedGaugeVec { + ::metrics::describe_gauge!(name.clone(), desc.clone()); + Box::new(MetricsGaugeVec { name, label_names }) + } + + fn register_histogram_vec( + &self, + name: Cow<'static, str>, + desc: Cow<'static, str>, + label_names: &'static [&'static str], + ) -> BoxedHistogramVec { + ::metrics::describe_histogram!(name.clone(), desc.clone()); + Box::new(MetricsHistogramVec { name, label_names }) + } + + fn register_histogram_vec_with_buckets( + &self, + name: Cow<'static, str>, + desc: Cow<'static, str>, + label_names: &'static [&'static str], + _buckets: Vec, + ) -> BoxedHistogramVec { + self.register_histogram_vec(name, desc, label_names) + } +} + +#[derive(Debug)] +struct MetricsCounterVec { + name: Cow<'static, str>, + label_names: &'static [&'static str], +} + +impl CounterVecOps for MetricsCounterVec { + fn counter(&self, labels: &[Cow<'static, str>]) -> BoxedCounter { + Box::new(MetricsCounter(::metrics::counter!( + self.name.clone(), + labeled(self.label_names, labels) + ))) + } +} + +#[derive(Debug)] +struct MetricsGaugeVec { + name: Cow<'static, str>, + label_names: &'static [&'static str], +} + +impl GaugeVecOps for MetricsGaugeVec { + fn gauge(&self, labels: &[Cow<'static, str>]) -> BoxedGauge { + Box::new(MetricsGauge(::metrics::gauge!( + self.name.clone(), + labeled(self.label_names, labels) + ))) + } +} + +#[derive(Debug)] +struct MetricsHistogramVec { + name: Cow<'static, str>, + label_names: &'static [&'static str], +} + +impl HistogramVecOps for MetricsHistogramVec { + fn histogram(&self, labels: &[Cow<'static, str>]) -> BoxedHistogram { + Box::new(MetricsHistogram(::metrics::histogram!( + self.name.clone(), + labeled(self.label_names, labels) + ))) + } +} + +struct MetricsCounter(::metrics::Counter); +struct MetricsGauge(::metrics::Gauge); +struct MetricsHistogram(::metrics::Histogram); + +impl fmt::Debug for MetricsCounter { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("MetricsCounter").finish() + } +} + +impl fmt::Debug for MetricsGauge { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("MetricsGauge").finish() + } +} + +impl fmt::Debug for MetricsHistogram { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("MetricsHistogram").finish() + } +} + +impl CounterOps for MetricsCounter { + fn increase(&self, val: u64) { + self.0.increment(val); + } +} + +impl GaugeOps for MetricsGauge { + fn increase(&self, val: u64) { + self.0.increment(val as f64); + } + + fn decrease(&self, val: u64) { + self.0.decrement(val as f64); + } + + fn absolute(&self, val: u64) { + self.0.set(val as f64); + } +} + +impl HistogramOps for MetricsHistogram { + fn record(&self, val: f64) { + self.0.record(val); + } +} + +fn labeled( + label_names: &'static [&'static str], + labels: &[Cow<'static, str>], +) -> Vec<::metrics::Label> { + debug_assert_eq!( + label_names.len(), + labels.len(), + "Foyer mixtrics label names and values must have the same length" + ); + let mut metric_labels = Vec::with_capacity(label_names.len()); + for (name, value) in label_names.iter().zip(labels.iter()) { + metric_labels.push(::metrics::Label::new(*name, value.clone().into_owned())); + } + metric_labels +} + +#[cfg(test)] +mod tests { + use ::metrics::with_local_recorder; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; + use mixtrics::metrics::RegistryOps; + + use super::*; + + #[test] + fn test_quickwit_metrics_registry_records_counter_gauge_histogram() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + with_local_recorder(&recorder, || { + let registry = QuickwitMetricsRegistry; + let counters = registry.register_counter_vec( + "foyer_memory_op_total".into(), + "foyer in-memory cache operations".into(), + &["name", "op"], + ); + counters + .counter(&["split-range-v1".into(), "hit".into()]) + .increase(1); + let gauges = registry.register_gauge_vec( + "foyer_memory_usage".into(), + "foyer in-memory cache usage".into(), + &["name"], + ); + gauges.gauge(&["split-range-v1".into()]).absolute(7); + let histograms = registry.register_histogram_vec_with_buckets( + "foyer_storage_op_duration".into(), + "foyer storage op duration".into(), + &["name", "op"], + vec![0.1, 1.0], + ); + histograms + .histogram(&["split-range-v1".into(), "hit".into()]) + .record(0.5); + }); + let snapshot = snapshotter.snapshot().into_vec(); + let has_counter = snapshot.iter().any(|(key, _, _, value)| { + key.key().name() == "foyer_memory_op_total" && *value == DebugValue::Counter(1) + }); + let has_gauge = snapshot + .iter() + .any(|(key, _, _, _)| key.key().name() == "foyer_memory_usage"); + let has_histogram = snapshot + .iter() + .any(|(key, _, _, _)| key.key().name() == "foyer_storage_op_duration"); + assert!( + has_counter, + "Foyer counter must register through the metrics recorder" + ); + assert!( + has_gauge, + "Foyer gauge must register through the metrics recorder" + ); + assert!( + has_histogram, + "Foyer histogram must register through the metrics recorder" + ); + } +} diff --git a/quickwit/quickwit-storage/src/split_range_cache/mod.rs b/quickwit/quickwit-storage/src/split_range_cache/mod.rs new file mode 100644 index 00000000000..d495a745277 --- /dev/null +++ b/quickwit/quickwit-storage/src/split_range_cache/mod.rs @@ -0,0 +1,190 @@ +// 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. + +mod key; +mod metrics; +mod storage; +#[cfg(test)] +mod tests; + +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, +}; +pub use storage::{FoyerSplitRangeStorage, wrap_storage_with_split_range_cache}; + +/// Process-wide Foyer hybrid cache for exact split byte-range payloads. +pub struct FoyerSplitRangeCache { + pub(crate) cache: foyer::HybridCache, + pub(crate) max_entry_size: usize, + 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_metrics_registry(Box::new(metrics::QuickwitMetricsRegistry)) + .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_clean_block_threshold(config.clean_block_threshold) + .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, + clean_block_threshold: 16, + } +} diff --git a/quickwit/quickwit-storage/src/split_range_cache/storage.rs b/quickwit/quickwit-storage/src/split_range_cache/storage.rs new file mode 100644 index 00000000000..80d6fd3821e --- /dev/null +++ b/quickwit/quickwit-storage/src/split_range_cache/storage.rs @@ -0,0 +1,307 @@ +// 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::future::Future; +use std::ops::Range; +use std::path::Path; +use std::sync::Arc; +use std::{fmt, io}; + +use async_trait::async_trait; +use bytes::Bytes; +use fail::fail_point; +use foyer::Code; +use quickwit_common::uri::Uri; +use tokio::io::AsyncRead; +use tracing::{error, warn}; + +use super::metrics::{FetchOutcome, record_admission_bypass, record_fail_open, record_request}; +use super::{FoyerSplitRangeCache, SplitRangeCacheKey}; +use crate::stable_deref_bytes::into_owned_bytes; +use crate::storage::SendableAsync; +use crate::{ + BulkDeleteError, OwnedBytes, PutPayload, Storage, StorageError, StorageErrorKind, StorageResult, +}; + +/// Foyer hybrid-cache entry header size in the 0.22.3 block engine. +pub(crate) const FOYER_ENTRY_HEADER_SIZE: usize = 36; +/// Foyer blob index reserved at the end of each block. +pub(crate) const FOYER_BLOB_INDEX_SIZE: usize = 4 * 1024; +/// Foyer disk page size used to align encoded entries. +pub(crate) const FOYER_PAGE_SIZE: usize = 4 * 1024; + +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub(crate) enum AdmissionBypass { + MaxEntrySize, + EncodedTooLarge, +} + +pub(crate) fn admission_bypass_reason( + key_size: usize, + value: &Bytes, + max_entry_size: usize, + block_size: usize, +) -> Option { + if value.len() > max_entry_size { + return Some(AdmissionBypass::MaxEntrySize); + } + // `max_entry_size < block_size` is not enough: the disk slot is + // `block_size - blob index` after header, key, and page alignment. + let encoded_len = FOYER_ENTRY_HEADER_SIZE + key_size + Bytes::estimated_size(value); + let aligned_len = encoded_len.div_ceil(FOYER_PAGE_SIZE) * FOYER_PAGE_SIZE; + if aligned_len > block_size - FOYER_BLOB_INDEX_SIZE { + return Some(AdmissionBypass::EncodedTooLarge); + } + None +} + +#[derive(Debug, thiserror::Error)] +#[error(transparent)] +struct LowerStorageError(StorageError); + +pub(crate) enum CacheFetchError { + Lower(StorageError), + Foyer, +} + +impl FoyerSplitRangeCache { + pub(crate) async fn get_or_fetch( + &self, + key: SplitRangeCacheKey, + fetch: F, + ) -> Result + where + F: FnOnce() -> Fut + Send + 'static, + Fut: Future> + Send + 'static, + { + let key_size = key.estimated_size(); + let requested_num_bytes = (key.byte_range.end - key.byte_range.start) as u64; + let max_entry_size = self.max_entry_size; + let block_size = self.block_size; + match self + .cache + .get_or_fetch(&key, || async move { + let bytes = fetch().await.map_err(LowerStorageError)?; + if let Some(reason) = + admission_bypass_reason(key_size, &bytes, max_entry_size, block_size) + { + record_admission_bypass(reason); + // Foyer keeps this tag on the RAM entry and skips disk enqueue + // on eviction (write-on-eviction). + Ok::<_, LowerStorageError>(( + bytes, + foyer::HybridCacheProperties::default() + .with_location(foyer::Location::InMem), + )) + } else { + Ok((bytes, foyer::HybridCacheProperties::default())) + } + }) + .await + { + Ok(entry) => { + let outcome = match entry.source() { + foyer::Source::Memory => FetchOutcome::MemoryHit, + foyer::Source::Disk => FetchOutcome::DiskHit, + foyer::Source::Outer => FetchOutcome::RemoteMiss, + }; + let bytes = entry.value().clone(); + record_request(outcome, bytes.len() as u64); + Ok(bytes) + } + Err(error) => { + record_request(FetchOutcome::Error, requested_num_bytes); + if let Some(lower_error) = error.downcast_ref::() { + Err(CacheFetchError::Lower(lower_error.0.clone())) + } else { + warn!( + error = ?error, + "split range cache fetch failed, reading from storage" + ); + Err(CacheFetchError::Foyer) + } + } + } + } +} + +/// Read-only [`Storage`] decorator that caches exact split byte-range payloads. +#[derive(Clone)] +pub struct FoyerSplitRangeStorage { + inner: Arc, + cache: Arc, +} + +/// Wraps `storage` so [`Storage::get_slice`] is served from `cache` on an exact +/// `{object URI, byte range}` key. +pub fn wrap_storage_with_split_range_cache( + cache: Arc, + storage: Arc, +) -> Arc { + Arc::new(FoyerSplitRangeStorage { + inner: storage, + cache, + }) +} + +impl fmt::Debug for FoyerSplitRangeStorage { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("FoyerSplitRangeStorage") + .field("uri", self.inner.uri()) + .finish() + } +} + +fn unsupported_operation(paths: &[&Path]) -> StorageError { + let msg = "Unsupported operation. FoyerSplitRangeStorage only supports async reads"; + error!(paths=?paths, msg); + io::Error::other(format!("{msg}: {paths:?}")).into() +} + +#[async_trait] +impl Storage for FoyerSplitRangeStorage { + async fn check_connectivity(&self) -> anyhow::Result<()> { + self.inner.check_connectivity().await + } + + async fn put(&self, path: &Path, _payload: Box) -> StorageResult<()> { + Err(unsupported_operation(&[path])) + } + + async fn copy_to(&self, path: &Path, output: &mut dyn SendableAsync) -> StorageResult<()> { + self.inner.copy_to(path, output).await + } + + async fn get_slice(&self, path: &Path, byte_range: Range) -> StorageResult { + if byte_range.is_empty() { + return Ok(OwnedBytes::empty()); + } + if should_bypass_cache() { + return self.inner.get_slice(path, byte_range).await; + } + let object_uri = self + .inner + .uri() + .join(path) + .map_err(|error| StorageErrorKind::Internal.with_error(error))? + .into_string(); + let key = SplitRangeCacheKey { + object_uri, + byte_range: byte_range.clone(), + }; + let inner = self.inner.clone(); + let owned_path = path.to_owned(); + let fetch_range = byte_range.clone(); + let fetch_result = self + .cache + .get_or_fetch(key, move || async move { + inner + .get_slice(&owned_path, fetch_range) + .await + .map(Bytes::from_owner) + }) + .await; + match fetch_result { + Ok(bytes) => Ok(into_owned_bytes(bytes)), + Err(CacheFetchError::Lower(storage_error)) => Err(storage_error), + Err(CacheFetchError::Foyer) => { + record_fail_open(); + self.inner.get_slice(path, byte_range).await + } + } + } + + async fn get_slice_stream( + &self, + path: &Path, + range: Range, + ) -> StorageResult> { + self.inner.get_slice_stream(path, range).await + } + + async fn get_all(&self, path: &Path) -> StorageResult { + self.inner.get_all(path).await + } + + async fn delete(&self, path: &Path) -> StorageResult<()> { + Err(unsupported_operation(&[path])) + } + + async fn bulk_delete<'a>(&self, paths: &[&'a Path]) -> Result<(), BulkDeleteError> { + Err(BulkDeleteError { + error: Some(unsupported_operation(paths)), + ..Default::default() + }) + } + + async fn file_num_bytes(&self, path: &Path) -> StorageResult { + self.inner.file_num_bytes(path).await + } + + fn uri(&self) -> &Uri { + self.inner.uri() + } +} + +fn should_bypass_cache() -> bool { + fail_point!("split-range-cache-before-get", |_| true); + false +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_admission_bypass_pinned_foyer_block_format() { + let key_size = 40; + let block_size = 2 * FOYER_PAGE_SIZE; + // encoded = 36 + 40 + (usize_len + value_len) = 84 + value_len on 64-bit. + // 4012 => encoded 4096, one page, fits in block_size - blob index. + // 4013 => encoded 4097, two pages, exceeds that slot. + assert_eq!( + admission_bypass_reason( + key_size, + &Bytes::from(vec![0; 4012]), + usize::MAX, + block_size + ), + None + ); + assert_eq!( + admission_bypass_reason( + key_size, + &Bytes::from(vec![0; 4013]), + usize::MAX, + block_size + ), + Some(AdmissionBypass::EncodedTooLarge) + ); + assert_eq!( + admission_bypass_reason(key_size, &Bytes::from(vec![0; 101]), 100, 4 * 1024 * 1024), + Some(AdmissionBypass::MaxEntrySize) + ); + // 5 KiB < max_entry_size 7 KiB < block_size 8 KiB, but the disk slot is + // only 4 KiB after the blob index. + assert_eq!( + admission_bypass_reason( + key_size, + &Bytes::from(vec![0; 5 * 1024]), + 7 * 1024, + 8 * 1024 + ), + Some(AdmissionBypass::EncodedTooLarge) + ); + } +} 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..d9800f2034b --- /dev/null +++ b/quickwit/quickwit-storage/src/split_range_cache/tests.rs @@ -0,0 +1,437 @@ +// 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; +use std::ops::Range; +use std::path::Path; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use async_trait::async_trait; +use quickwit_common::uri::Uri; +use quickwit_config::SplitRangeCacheWritePolicy; +use tokio::io::AsyncRead; +use tokio::sync::watch; + +use super::metrics::{ADMISSION_MAX_ENTRY_SIZE, REQUESTS_ERROR, REQUESTS_MEMORY, REQUESTS_MISS}; +use super::*; +use crate::storage::SendableAsync; +use crate::{ + BulkDeleteError, OwnedBytes, PutPayload, RamStorageBuilder, Storage, StorageErrorKind, + StorageResult, wrap_storage_with_split_range_cache, +}; + +const SPLIT_PATH: &str = "a.split"; +const SPLIT_BYTES: &[u8] = b"abcde"; + +#[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(); +} + +struct LowerProbe { + inner: Arc, + get_slice_calls: AtomicUsize, + get_slice_completed: AtomicUsize, + gate: watch::Receiver, +} + +impl fmt::Debug for LowerProbe { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("LowerProbe") + .field("uri", self.inner.uri()) + .finish() + } +} + +#[async_trait] +impl Storage for LowerProbe { + async fn check_connectivity(&self) -> anyhow::Result<()> { + self.inner.check_connectivity().await + } + + async fn put(&self, path: &Path, payload: Box) -> StorageResult<()> { + self.inner.put(path, payload).await + } + + async fn copy_to(&self, path: &Path, output: &mut dyn SendableAsync) -> StorageResult<()> { + self.inner.copy_to(path, output).await + } + + async fn get_slice(&self, path: &Path, range: Range) -> StorageResult { + self.get_slice_calls.fetch_add(1, Ordering::Relaxed); + let mut gate = self.gate.clone(); + let _ = gate.wait_for(|open| *open).await; + let result = self.inner.get_slice(path, range).await; + self.get_slice_completed.fetch_add(1, Ordering::Relaxed); + result + } + + async fn get_slice_stream( + &self, + path: &Path, + range: Range, + ) -> StorageResult> { + self.inner.get_slice_stream(path, range).await + } + + async fn get_all(&self, path: &Path) -> StorageResult { + self.inner.get_all(path).await + } + + async fn delete(&self, path: &Path) -> StorageResult<()> { + self.inner.delete(path).await + } + + async fn bulk_delete<'a>(&self, paths: &[&'a Path]) -> Result<(), BulkDeleteError> { + self.inner.bulk_delete(paths).await + } + + async fn file_num_bytes(&self, path: &Path) -> StorageResult { + self.inner.file_num_bytes(path).await + } + + fn uri(&self) -> &Uri { + self.inner.uri() + } +} + +struct Fixture { + storage: Arc, + cache: Arc, + lower: Arc, + gate_tx: watch::Sender, + _temp_dir: tempfile::TempDir, +} + +impl Fixture { + async fn new() -> Self { + Self::with_payload(SPLIT_BYTES, true).await + } + + async fn new_with_blocked_lower_read() -> Self { + Self::with_payload(SPLIT_BYTES, false).await + } + + async fn with_payload(payload: &[u8], gate_open: bool) -> Self { + let temp_dir = tempfile::tempdir().unwrap(); + let cache = Arc::new( + FoyerSplitRangeCache::open(&config_for_test(temp_dir.path())) + .await + .unwrap(), + ); + let ram: Arc = Arc::new( + RamStorageBuilder::default() + .put(SPLIT_PATH, payload) + .build(), + ); + let (gate_tx, gate_rx) = watch::channel(gate_open); + let lower = Arc::new(LowerProbe { + inner: ram, + get_slice_calls: AtomicUsize::new(0), + get_slice_completed: AtomicUsize::new(0), + gate: gate_rx, + }); + let storage = wrap_storage_with_split_range_cache(cache.clone(), lower.clone()); + Self { + storage, + cache, + lower, + gate_tx, + _temp_dir: temp_dir, + } + } + + fn release_lower_read(&self) { + self.gate_tx.send(true).unwrap(); + } + + fn lower_reads(&self) -> usize { + self.lower.get_slice_calls.load(Ordering::Relaxed) + } + + fn lower_completed(&self) -> usize { + self.lower.get_slice_completed.load(Ordering::Relaxed) + } + + async fn wait_until_lower_read_started(&self) { + wait_until(|| self.lower_reads() > 0, "lower read start").await; + } + + async fn wait_until_lower_read_completed(&self) { + wait_until(|| self.lower_completed() > 0, "lower read completion").await; + } + + async fn close(&self) { + self.cache.close().await.unwrap(); + } +} + +async fn wait_until(predicate: impl Fn() -> bool, what: &str) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + while !predicate() { + if tokio::time::Instant::now() >= deadline { + panic!("timed out waiting for {what}"); + } + tokio::time::sleep(Duration::from_millis(1)).await; + } +} + +#[tokio::test] +async fn test_empty_range_and_exact_hit_behavior() { + let fixture = Fixture::new().await; + let path = Path::new(SPLIT_PATH); + assert!( + fixture + .storage + .get_slice(path, 4..4) + .await + .unwrap() + .is_empty() + ); + assert_eq!(fixture.lower_reads(), 0); + let misses_before = REQUESTS_MISS.get(); + let memory_hits_before = REQUESTS_MEMORY.get(); + assert_eq!( + fixture + .storage + .get_slice(path, 1..4) + .await + .unwrap() + .as_slice(), + b"bcd" + ); + assert!(REQUESTS_MISS.get() > misses_before); + assert_eq!( + fixture + .storage + .get_slice(path, 1..4) + .await + .unwrap() + .as_slice(), + b"bcd" + ); + assert!(REQUESTS_MEMORY.get() > memory_hits_before); + assert_eq!(fixture.lower_reads(), 1); + fixture.storage.get_slice(path, 0..5).await.unwrap(); + assert_eq!( + fixture.lower_reads(), + 2, + "covering ranges are distinct keys" + ); + fixture.close().await; +} + +#[tokio::test] +async fn test_identical_concurrent_misses_fetch_once() { + let fixture = Fixture::new_with_blocked_lower_read().await; + let path = Path::new(SPLIT_PATH); + let first = fixture.storage.get_slice(path, 0..4); + let second = fixture.storage.get_slice(path, 0..4); + let release = async { + fixture.wait_until_lower_read_started().await; + fixture.release_lower_read(); + }; + let (first_result, second_result, _) = tokio::join!(first, second, release); + assert_eq!(first_result.unwrap(), second_result.unwrap()); + assert_eq!(fixture.lower_reads(), 1); + fixture.close().await; +} + +#[tokio::test] +async fn test_remote_error_is_not_cached_or_rewritten() { + let fixture = Fixture::new().await; + let errors_before = REQUESTS_ERROR.get(); + for _ in 0..2 { + let error = fixture + .storage + .get_slice(Path::new("missing.split"), 0..4) + .await + .unwrap_err(); + assert_eq!(error.kind(), StorageErrorKind::NotFound); + } + assert!(REQUESTS_ERROR.get() >= errors_before + 2); + assert_eq!(fixture.lower_reads(), 2); + fixture.close().await; +} + +#[tokio::test] +async fn test_writes_are_unsupported() { + let fixture = Fixture::new().await; + let path = Path::new(SPLIT_PATH); + let put_error = fixture + .storage + .put(path, Box::new(b"x".to_vec())) + .await + .unwrap_err(); + assert_eq!(put_error.kind(), StorageErrorKind::Io); + assert!( + put_error + .to_string() + .contains("Unsupported operation. FoyerSplitRangeStorage only supports async reads") + ); + let delete_error = fixture.storage.delete(path).await.unwrap_err(); + assert_eq!(delete_error.kind(), StorageErrorKind::Io); + let bulk_error = fixture.storage.bulk_delete(&[path]).await.unwrap_err(); + assert_eq!( + bulk_error.error.as_ref().unwrap().kind(), + StorageErrorKind::Io + ); + fixture.close().await; +} + +#[tokio::test] +async fn test_get_all_is_not_cached() { + let fixture = Fixture::new().await; + let path = Path::new(SPLIT_PATH); + assert_eq!( + fixture.storage.get_all(path).await.unwrap().as_slice(), + SPLIT_BYTES + ); + assert_eq!(fixture.lower_reads(), 0); + fixture.storage.get_slice(path, 0..5).await.unwrap(); + assert_eq!(fixture.lower_reads(), 1); + fixture.close().await; +} + +#[tokio::test] +async fn test_initiating_caller_drop_surviving_waiter_succeeds() { + let fixture = Fixture::new_with_blocked_lower_read().await; + let path = Path::new(SPLIT_PATH); + let mut initiating = Box::pin(fixture.storage.get_slice(path, 0..4)); + tokio::select! { + biased; + result = &mut initiating => panic!("fetch completed before release: {result:?}"), + () = fixture.wait_until_lower_read_started() => {} + } + drop(initiating); + let waiter = fixture.storage.get_slice(path, 0..4); + fixture.release_lower_read(); + assert_eq!(waiter.await.unwrap().as_slice(), b"abcd"); + assert_eq!(fixture.lower_reads(), 1); + fixture.close().await; +} + +#[tokio::test] +async fn test_waiter_drop_does_not_cancel_fetch() { + let fixture = Fixture::new_with_blocked_lower_read().await; + let path = Path::new(SPLIT_PATH); + let mut initiating = Box::pin(fixture.storage.get_slice(path, 0..4)); + tokio::select! { + biased; + result = &mut initiating => panic!("fetch completed before release: {result:?}"), + () = fixture.wait_until_lower_read_started() => {} + } + let mut waiter = Box::pin(fixture.storage.get_slice(path, 0..4)); + for _ in 0..16 { + tokio::select! { + biased; + result = &mut waiter => panic!("waiter completed before release: {result:?}"), + () = tokio::task::yield_now() => {} + } + } + drop(waiter); + fixture.release_lower_read(); + assert_eq!(initiating.await.unwrap().as_slice(), b"abcd"); + assert_eq!(fixture.lower_reads(), 1); + fixture.close().await; +} + +#[tokio::test] +async fn test_all_callers_dropped_detached_completion() { + let fixture = Fixture::new_with_blocked_lower_read().await; + let path = Path::new(SPLIT_PATH); + let mut initiating = Box::pin(fixture.storage.get_slice(path, 0..4)); + tokio::select! { + biased; + result = &mut initiating => panic!("fetch completed before release: {result:?}"), + () = fixture.wait_until_lower_read_started() => {} + } + drop(initiating); + fixture.release_lower_read(); + fixture.wait_until_lower_read_completed().await; + assert_eq!( + fixture + .storage + .get_slice(path, 0..4) + .await + .unwrap() + .as_slice(), + b"abcd" + ); + assert_eq!(fixture.lower_reads(), 1); + fixture.close().await; +} + +#[tokio::test] +async fn test_oversized_value_is_memory_only_and_returned() { + let payload = vec![7u8; 3 * 1024 * 1024]; + let fixture = Fixture::with_payload(&payload, true).await; + let path = Path::new(SPLIT_PATH); + let range = 0..payload.len(); + let bypasses_before = ADMISSION_MAX_ENTRY_SIZE.get(); + assert_eq!( + fixture + .storage + .get_slice(path, range.clone()) + .await + .unwrap() + .as_slice(), + payload.as_slice() + ); + assert_eq!( + fixture + .storage + .get_slice(path, range) + .await + .unwrap() + .as_slice(), + payload.as_slice() + ); + assert_eq!(fixture.lower_reads(), 1); + assert!(ADMISSION_MAX_ENTRY_SIZE.get() > bypasses_before); + fixture.close().await; +}