diff --git a/config/quickwit.yaml b/config/quickwit.yaml index 997c6bf7df7..110d5bc542d 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 @@ -213,19 +215,18 @@ indexer: # # 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: /var/cache/quickwit/split-range-v1 -# disk_capacity: 300G -# memory_capacity: 1G -# buffer_pool_size: 512M -# submit_queue_size_threshold: 1G +# path: /quickwit/qwdata/split-range-v1 +# disk_capacity: 1500G +# memory_capacity: 15G # memory_eviction_policy: s3-fifo -# write_policy: write-on-eviction # compression: lz4 # recover_mode: quiet -# block_size: 16M -# max_entry_size: 15M -# flushers: 4 -# reclaimers: 4 +# block_size: 64M +# max_entry_size: 60M +# flushers: 24 +# buffer_pool_size: 2G +# submit_queue_size_threshold: 3G +# reclaimers: 8 # -------------------------------- Jaeger settings -------------------------------- jaeger: diff --git a/docs/configuration/node-config.md b/docs/configuration/node-config.md index cbe3f8b0022..f3da405da2b 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,25 @@ 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. | | Example: @@ -336,6 +356,19 @@ 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 ``` ## 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 79a70cb7db0..8731ca5a860 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -9588,6 +9588,7 @@ dependencies = [ "lru 0.18.0", "md5", "metrics", + "metrics-util", "mini-moka", "mixtrics", "mockall", diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index 0793bf442fb..9fcfc0b156d 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -506,11 +506,12 @@ 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()}>")] @@ -698,6 +699,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) @@ -759,7 +767,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)), @@ -793,6 +801,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 @@ -1429,6 +1451,80 @@ mod tests { #[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] diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index 14542a0c174..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)), diff --git a/quickwit/quickwit-lambda-server/src/context.rs b/quickwit/quickwit-lambda-server/src/context.rs index e8faad760a5..579d01a6036 100644 --- a/quickwit/quickwit-lambda-server/src/context.rs +++ b/quickwit/quickwit-lambda-server/src/context.rs @@ -65,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/src/service.rs b/quickwit/quickwit-search/src/service.rs index 518b330583a..2647f7e6edb 100644 --- a/quickwit/quickwit-search/src/service.rs +++ b/quickwit/quickwit-search/src/service.rs @@ -474,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); @@ -507,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-storage/Cargo.toml b/quickwit/quickwit-storage/Cargo.toml index c6b640da215..52231bd2970 100644 --- a/quickwit/quickwit-storage/Cargo.toml +++ b/quickwit/quickwit-storage/Cargo.toml @@ -65,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 @@ -107,8 +108,3 @@ integration-testsuite = [ "dep:reqwest", ] testsuite = ["mockall"] - -[package.metadata.cargo-machete] -# Declared here so the Foyer pin lands with licenses. Follow-up split-range -# cache PRs use these crates; remove the ignore when they do. -ignored = ["metrics", "mixtrics"] 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 index c122f186c5e..0e7dbca5735 100644 --- a/quickwit/quickwit-storage/src/split_range_cache/mod.rs +++ b/quickwit/quickwit-storage/src/split_range_cache/mod.rs @@ -13,6 +13,7 @@ // limitations under the License. mod key; +mod metrics; mod storage; #[cfg(test)] mod tests; @@ -61,6 +62,7 @@ impl FoyerSplitRangeCache { 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) diff --git a/quickwit/quickwit-storage/src/split_range_cache/storage.rs b/quickwit/quickwit-storage/src/split_range_cache/storage.rs index 2b2c9d85227..80d6fd3821e 100644 --- a/quickwit/quickwit-storage/src/split_range_cache/storage.rs +++ b/quickwit/quickwit-storage/src/split_range_cache/storage.rs @@ -26,6 +26,7 @@ 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; @@ -85,13 +86,17 @@ impl FoyerSplitRangeCache { 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 admission_bypass_reason(key_size, &bytes, max_entry_size, block_size).is_some() { + 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>(( @@ -105,8 +110,18 @@ impl FoyerSplitRangeCache { }) .await { - Ok(entry) => Ok(entry.value().clone()), + 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 { @@ -200,7 +215,10 @@ impl Storage for FoyerSplitRangeStorage { match fetch_result { Ok(bytes) => Ok(into_owned_bytes(bytes)), Err(CacheFetchError::Lower(storage_error)) => Err(storage_error), - Err(CacheFetchError::Foyer) => self.inner.get_slice(path, byte_range).await, + Err(CacheFetchError::Foyer) => { + record_fail_open(); + self.inner.get_slice(path, byte_range).await + } } } diff --git a/quickwit/quickwit-storage/src/split_range_cache/tests.rs b/quickwit/quickwit-storage/src/split_range_cache/tests.rs index d1454e51cf8..d9800f2034b 100644 --- a/quickwit/quickwit-storage/src/split_range_cache/tests.rs +++ b/quickwit/quickwit-storage/src/split_range_cache/tests.rs @@ -25,6 +25,7 @@ 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::{ @@ -232,6 +233,8 @@ async fn test_empty_range_and_exact_hit_behavior() { .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 @@ -241,6 +244,7 @@ async fn test_empty_range_and_exact_hit_behavior() { .as_slice(), b"bcd" ); + assert!(REQUESTS_MISS.get() > misses_before); assert_eq!( fixture .storage @@ -250,6 +254,7 @@ async fn test_empty_range_and_exact_hit_behavior() { .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!( @@ -279,6 +284,7 @@ async fn test_identical_concurrent_misses_fetch_once() { #[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 @@ -287,6 +293,7 @@ async fn test_remote_error_is_not_cached_or_rewritten() { .unwrap_err(); assert_eq!(error.kind(), StorageErrorKind::NotFound); } + assert!(REQUESTS_ERROR.get() >= errors_before + 2); assert_eq!(fixture.lower_reads(), 2); fixture.close().await; } @@ -405,6 +412,7 @@ async fn test_oversized_value_is_memory_only_and_returned() { 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 @@ -424,5 +432,6 @@ async fn test_oversized_value_is_memory_only_and_returned() { payload.as_slice() ); assert_eq!(fixture.lower_reads(), 1); + assert!(ADMISSION_MAX_ENTRY_SIZE.get() > bypasses_before); fixture.close().await; }