diff --git a/quickwit/quickwit-storage/src/community_metrics.rs b/quickwit/quickwit-storage/src/community_metrics.rs new file mode 100644 index 00000000000..e4c92616e53 --- /dev/null +++ b/quickwit/quickwit-storage/src/community_metrics.rs @@ -0,0 +1,266 @@ +// 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. + +//! Temporary per-community object storage metrics. +//! +//! The community is extracted from the object key (`events-/.split`). +//! On top of plain counters, we expose the max 1-second value observed over the last +//! [`WINDOW_SECS`] seconds, because the average rate computed from counters over a 15s/30s +//! scrape interval hides the real peaks. + +use std::collections::{HashMap, VecDeque}; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use once_cell::sync::Lazy; +use quickwit_common::metrics::{ + IntCounter, IntCounterVec, IntGauge, IntGaugeVec, new_counter_vec, new_gauge_vec, +}; + +use crate::metrics_wrappers::ActionLabel; + +const WINDOW_SECS: usize = 30; +const UNKNOWN_COMMUNITY: &str = "unknown"; + +struct CommunityMetrics { + requests_total: IntCounterVec<2>, + bytes_total: IntCounterVec<2>, + peak_requests_per_sec: IntGaugeVec<2>, + peak_bytes_per_sec: IntGaugeVec<2>, +} + +static COMMUNITY_METRICS: Lazy = Lazy::new(|| CommunityMetrics { + requests_total: new_counter_vec( + "object_storage_community_requests_total", + "Number of requests (including retries) to the object store, by community and action.", + "storage", + &[], + ["community", "action"], + ), + bytes_total: new_counter_vec( + "object_storage_community_bytes_total", + "Number of bytes exchanged with the object store, by community and direction.", + "storage", + &[], + ["community", "direction"], + ), + peak_requests_per_sec: new_gauge_vec( + "object_storage_community_peak_requests_per_sec", + "Max number of requests to the object store within one second over the last 30 seconds, \ + by community and action.", + "storage", + &[], + ["community", "action"], + ), + peak_bytes_per_sec: new_gauge_vec( + "object_storage_community_peak_bytes_per_sec", + "Max number of bytes exchanged with the object store within one second over the last 30 \ + seconds, by community and direction.", + "storage", + &[], + ["community", "direction"], + ), +}); + +#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] +enum Kind { + Requests(&'static str), + Bytes(&'static str), +} + +struct Slot { + current_sec: u64, + window: VecDeque, + total_counter: IntCounter, + peak_gauge: IntGauge, +} + +impl Slot { + fn new(community: &str, kind: Kind) -> Self { + let (total_counter, peak_gauge) = match kind { + Kind::Requests(action) => ( + COMMUNITY_METRICS + .requests_total + .with_label_values([community, action]), + COMMUNITY_METRICS + .peak_requests_per_sec + .with_label_values([community, action]), + ), + Kind::Bytes(direction) => ( + COMMUNITY_METRICS + .bytes_total + .with_label_values([community, direction]), + COMMUNITY_METRICS + .peak_bytes_per_sec + .with_label_values([community, direction]), + ), + }; + Slot { + current_sec: 0, + window: VecDeque::with_capacity(WINDOW_SECS + 1), + total_counter, + peak_gauge, + } + } +} + +#[derive(Default)] +struct Slots { + per_community: HashMap>, +} + +impl Slots { + fn record(&mut self, community: &str, kind: Kind, value: u64) { + let community_slots = match self.per_community.get_mut(community) { + Some(community_slots) => community_slots, + None => self.per_community.entry(community.to_string()).or_default(), + }; + let slot = community_slots + .entry(kind) + .or_insert_with(|| Slot::new(community, kind)); + slot.current_sec += value; + slot.total_counter.inc_by(value); + } + + /// Closes the current one-second bucket of every slot and refreshes the peak gauges. + fn tick(&mut self) { + for slot in self + .per_community + .values_mut() + .flat_map(HashMap::values_mut) + { + slot.window.push_back(slot.current_sec); + if slot.window.len() > WINDOW_SECS { + slot.window.pop_front(); + } + slot.current_sec = 0; + let peak = slot.window.iter().copied().max().unwrap_or(0); + slot.peak_gauge.set(peak as i64); + } + } +} + +static SLOTS: Lazy> = Lazy::new(|| { + std::thread::Builder::new() + .name("community-metrics".to_string()) + .spawn(|| { + let mut next_tick = Instant::now(); + loop { + next_tick += Duration::from_secs(1); + std::thread::sleep(next_tick.saturating_duration_since(Instant::now())); + SLOTS.lock().unwrap().tick(); + } + }) + .expect("failed to spawn community metrics thread"); + Mutex::new(Slots::default()) +}); + +/// Extracts the community UUID from an object key such as +/// `events-/.split`. +fn extract_community(key: &str) -> &str { + let Some(parent) = key.rsplit('/').nth(1) else { + return UNKNOWN_COMMUNITY; + }; + let Some(community) = parent + .len() + .checked_sub(36) + .and_then(|start| parent.get(start..)) + else { + return UNKNOWN_COMMUNITY; + }; + let is_uuid = community.bytes().enumerate().all(|(i, b)| match i { + 8 | 13 | 18 | 23 => b == b'-', + _ => b.is_ascii_hexdigit(), + }); + if is_uuid { + community + } else { + UNKNOWN_COMMUNITY + } +} + +pub(crate) fn record_request(key: &str, action: ActionLabel) { + let community = extract_community(key); + SLOTS + .lock() + .unwrap() + .record(community, Kind::Requests(action.as_str()), 1); +} + +pub(crate) fn record_bytes(key: &str, direction: &'static str, num_bytes: u64) { + let community = extract_community(key); + SLOTS + .lock() + .unwrap() + .record(community, Kind::Bytes(direction), num_bytes); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_extract_community() { + assert_eq!( + extract_community( + "events-49cca94b-5ad3-4857-b644-1348a1ab5ea6/01M1PGRCR6JSW021GKK3Z4V5BW.split" + ), + "49cca94b-5ad3-4857-b644-1348a1ab5ea6" + ); + assert_eq!( + extract_community( + "indexes/events-49cca94b-5ad3-4857-b644-1348a1ab5ea6/01M1PGRCR6JSW021GKK3Z4V5BW.\ + split" + ), + "49cca94b-5ad3-4857-b644-1348a1ab5ea6" + ); + assert_eq!(extract_community("metastore.json"), UNKNOWN_COMMUNITY); + assert_eq!( + extract_community("my-index/01M1PGRCR6JSW021GKK3Z4V5BW.split"), + UNKNOWN_COMMUNITY + ); + assert_eq!( + extract_community("events-49cca94b_5ad3-4857-b644-1348a1ab5ea6/split"), + UNKNOWN_COMMUNITY + ); + assert_eq!(extract_community("é/split"), UNKNOWN_COMMUNITY); + } + + #[test] + fn test_slots_peak() { + let community = "test-community-peak"; + let kind = Kind::Requests("get_object"); + let mut slots = Slots::default(); + for _ in 0..5 { + slots.record(community, kind, 1); + } + slots.tick(); + let slot = &slots.per_community[community][&kind]; + assert_eq!(slot.peak_gauge.get(), 5); + assert_eq!(slot.total_counter.get(), 5); + + slots.record(community, kind, 2); + slots.tick(); + assert_eq!(slots.per_community[community][&kind].peak_gauge.get(), 5); + + for _ in 0..WINDOW_SECS - 1 { + slots.tick(); + } + // The 5 has left the window, the 2 is still in it. + assert_eq!(slots.per_community[community][&kind].peak_gauge.get(), 2); + slots.tick(); + assert_eq!(slots.per_community[community][&kind].peak_gauge.get(), 0); + assert_eq!(slots.per_community[community][&kind].total_counter.get(), 7); + } +} diff --git a/quickwit/quickwit-storage/src/lib.rs b/quickwit/quickwit-storage/src/lib.rs index 524374917e7..5e828e8d7e3 100644 --- a/quickwit/quickwit-storage/src/lib.rs +++ b/quickwit/quickwit-storage/src/lib.rs @@ -26,6 +26,7 @@ //! //! The `BundleStorage` bundles together multiple files into a single file. mod cache; +mod community_metrics; mod debouncer; mod file_descriptor_cache; mod metrics; diff --git a/quickwit/quickwit-storage/src/metrics_wrappers.rs b/quickwit/quickwit-storage/src/metrics_wrappers.rs index adbe63aa673..57ffb3f5914 100644 --- a/quickwit/quickwit-storage/src/metrics_wrappers.rs +++ b/quickwit/quickwit-storage/src/metrics_wrappers.rs @@ -42,7 +42,7 @@ pub enum ActionLabel { } impl ActionLabel { - fn as_str(&self) -> &'static str { + pub(crate) fn as_str(&self) -> &'static str { match self { ActionLabel::AbortMultipartUpload => "abort_multipart_upload", ActionLabel::CompleteMultipartUpload => "complete_multipart_upload", diff --git a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs index 649e601ff1f..640f44295e0 100644 --- a/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs +++ b/quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs @@ -59,7 +59,7 @@ use crate::stable_deref_bytes::into_owned_bytes; use crate::storage::SendableAsync; use crate::{ BulkDeleteError, DeleteFailure, OwnedBytes, Storage, StorageError, StorageErrorKind, - StorageResolverError, StorageResult, + StorageResolverError, StorageResult, community_metrics, }; const MAX_GET_OBJECT_CONNECTIONS_ENV_KEY: &str = "QW_S3_MAX_GET_OBJECT_CONNECTIONS"; @@ -401,6 +401,8 @@ impl S3CompatibleObjectStorage { }) => {} None => {} } + community_metrics::record_request(key, ActionLabel::PutObject); + community_metrics::record_bytes(key, "upload", len); req_builder .send() .with_count_and_upload_metrics(ActionLabel::PutObject, len) @@ -433,6 +435,7 @@ impl S3CompatibleObjectStorage { async fn create_multipart_upload(&self, key: &str) -> StorageResult { let upload_id = aws_retry(&self.retry_params, || async { + community_metrics::record_request(key, ActionLabel::CreateMultipartUpload); let mut req_builder = self .s3_client .create_multipart_upload() @@ -572,6 +575,8 @@ impl S3CompatibleObjectStorage { }) => {} None => {} } + community_metrics::record_request(key, ActionLabel::UploadPart); + community_metrics::record_bytes(key, "upload", part.len()); let upload_part_output = req_builder .send() .with_count_and_upload_metrics(ActionLabel::UploadPart, part.len()) @@ -647,6 +652,7 @@ impl S3CompatibleObjectStorage { .set_parts(Some(completed_parts)) .build(); aws_retry(&self.retry_params, || async { + community_metrics::record_request(key, ActionLabel::CompleteMultipartUpload); self.s3_client .complete_multipart_upload() .bucket(self.bucket.clone()) @@ -663,6 +669,7 @@ impl S3CompatibleObjectStorage { async fn abort_multipart_upload(&self, key: &str, upload_id: &str) -> StorageResult<()> { aws_retry(&self.retry_params, || async { + community_metrics::record_request(key, ActionLabel::AbortMultipartUpload); self.s3_client .abort_multipart_upload() .bucket(self.bucket.clone()) @@ -690,7 +697,7 @@ impl S3CompatibleObjectStorage { .s3_client .get_object() .bucket(self.bucket.clone()) - .key(key) + .key(&key) .set_range(range_str); match &self.encryption { Some(S3EncryptionConfig::SseC { @@ -706,10 +713,13 @@ impl S3CompatibleObjectStorage { None => {} } let connection_permit = acquire_get_object_connection_permit().await; + community_metrics::record_request(&key, ActionLabel::GetObject); let get_object_output = req_builder .send() .with_count_and_duration_metrics(ActionLabel::GetObject) .await?; + let content_length = get_object_output.content_length().unwrap_or(0).max(0) as u64; + community_metrics::record_bytes(&key, "download", content_length); Ok((get_object_output, connection_permit)) } @@ -916,6 +926,7 @@ impl Storage for S3CompatibleObjectStorage { let bucket = self.bucket.clone(); let key = self.key(path); let delete_res = aws_retry(&self.retry_params, || async { + community_metrics::record_request(&key, ActionLabel::DeleteObject); self.s3_client .delete_object() .bucket(&bucket) @@ -993,6 +1004,7 @@ impl Storage for S3CompatibleObjectStorage { let bucket = self.bucket.clone(); let key = self.key(path); let head_object_output = aws_retry(&self.retry_params, || async { + community_metrics::record_request(&key, ActionLabel::HeadObject); let mut req_builder = self.s3_client.head_object().bucket(&bucket).key(&key); match &self.encryption { Some(S3EncryptionConfig::SseC {