From fe6f19dfc25fe14d6db97e882f1ea24770e90c25 Mon Sep 17 00:00:00 2001 From: Arnaud Robert Date: Fri, 31 Jul 2026 16:35:56 +0200 Subject: [PATCH 1/3] indexing: label published split docs by publication type --- .../src/actors/log_publisher_impl.rs | 60 ++++++++++++++++++- quickwit/quickwit-indexing/src/metrics.rs | 24 ++++++-- 2 files changed, 76 insertions(+), 8 deletions(-) diff --git a/quickwit/quickwit-indexing/src/actors/log_publisher_impl.rs b/quickwit/quickwit-indexing/src/actors/log_publisher_impl.rs index 469e1c67254..db3943be536 100644 --- a/quickwit/quickwit-indexing/src/actors/log_publisher_impl.rs +++ b/quickwit/quickwit-indexing/src/actors/log_publisher_impl.rs @@ -116,8 +116,13 @@ impl Handler for Publisher { } return Err(publish_error); } + let publication_type = if replaced_split_ids.is_empty() { + "initial" + } else { + "replacement" + }; for split in &new_splits { - record_published_split(&index_id, split); + record_published_split(&index_id, split, publication_type); } let num_docs: usize = new_splits.iter().map(|split| split.num_docs).sum(); // `footer_offsets.end` is the on-disk size of the split file in bytes. @@ -161,6 +166,7 @@ mod tests { IndexCheckpointDelta, PartitionId, SourceCheckpoint, SourceCheckpointDelta, }; use quickwit_metastore::{PublishSplitsRequestExt, SplitMetadata}; + use quickwit_metrics::{counter, label_values}; use quickwit_proto::metastore::{ EmptyResponse, MetastoreError, MetastoreServiceClient, MockMetastoreService, }; @@ -169,13 +175,34 @@ mod tests { use super::PUBLISHER_NAME; use crate::actors::publisher::Publisher; + use crate::metrics::{PUBLISHED_SPLIT_DOCS, PUBLISHED_SPLIT_DOCS_TOTAL}; use crate::models::{PublishLock, SharedPublishToken, SplitsUpdate}; use crate::source::SuggestTruncate; + fn published_split_docs( + index_id: &str, + source_id: &str, + merge_ops: usize, + publication_type: &'static str, + ) -> u64 { + let index = quickwit_common::metrics::index_label(index_id); + let labels = label_values!( + PUBLISHED_SPLIT_DOCS => + index.to_string(), + source_id.to_string(), + merge_ops.to_string(), + publication_type + ); + counter!(parent: PUBLISHED_SPLIT_DOCS_TOTAL, labels: [labels]).get() + } + #[tokio::test] async fn test_publisher_publish_operation() { let universe = Universe::with_accelerated_time(); - let ref_index_uid: IndexUid = IndexUid::for_test("index", 1); + let ref_index_uid: IndexUid = IndexUid::for_test("publisher-metric-initial", 1); + let initial_before = published_split_docs(&ref_index_uid.index_id, "source", 0, "initial"); + let replacement_before = + published_split_docs(&ref_index_uid.index_id, "source", 0, "replacement"); let mut mock_metastore = MockMetastoreService::new(); let ref_index_uid_clone = ref_index_uid.clone(); mock_metastore @@ -213,6 +240,8 @@ mod tests { index_uid: ref_index_uid.clone(), new_splits: vec![SplitMetadata { split_id: "split".into(), + source_id: "source".into(), + num_docs: 7, ..Default::default() }], replaced_split_ids: Vec::new(), @@ -230,6 +259,15 @@ mod tests { let publisher_observation = publisher_handle.process_pending_and_observe().await.state; assert_eq!(publisher_observation.num_published_splits, 1); + assert_eq!( + published_split_docs(&ref_index_uid.index_id, "source", 0, "initial") - initial_before, + 7 + ); + assert_eq!( + published_split_docs(&ref_index_uid.index_id, "source", 0, "replacement") + - replacement_before, + 0 + ); let suggest_truncate_checkpoints: Vec = source_inbox .drain_for_test_typed::() @@ -328,7 +366,10 @@ mod tests { async fn test_publisher_replace_operation() { let universe = Universe::with_accelerated_time(); let mut mock_metastore = MockMetastoreService::new(); - let ref_index_uid: IndexUid = IndexUid::for_test("index", 1); + let ref_index_uid: IndexUid = IndexUid::for_test("publisher-metric-replacement", 1); + let initial_before = published_split_docs(&ref_index_uid.index_id, "source", 0, "initial"); + let replacement_before = + published_split_docs(&ref_index_uid.index_id, "source", 0, "replacement"); let ref_index_uid_clone = ref_index_uid.clone(); mock_metastore .expect_publish_splits() @@ -357,6 +398,10 @@ mod tests { index_uid: ref_index_uid.clone(), new_splits: vec![SplitMetadata { split_id: "split3".into(), + source_id: "source".into(), + num_docs: 11, + // Delete-and-merge rewrites can preserve zero merge operations. + num_merge_ops: 0, ..Default::default() }], replaced_split_ids: vec![SplitId::from("split1"), SplitId::from("split2")], @@ -370,6 +415,15 @@ mod tests { let publisher_observation = publisher_handle.process_pending_and_observe().await.state; assert_eq!(publisher_observation.num_published_splits, 0); assert_eq!(publisher_observation.num_replace_operations, 1); + assert_eq!( + published_split_docs(&ref_index_uid.index_id, "source", 0, "replacement") + - replacement_before, + 11 + ); + assert_eq!( + published_split_docs(&ref_index_uid.index_id, "source", 0, "initial") - initial_before, + 0 + ); use crate::models::NewSplits; let merge_planner_msgs = merge_planner_inbox.drain_for_test_typed::(); diff --git a/quickwit/quickwit-indexing/src/metrics.rs b/quickwit/quickwit-indexing/src/metrics.rs index b7431c6d032..4208d25b536 100644 --- a/quickwit/quickwit-indexing/src/metrics.rs +++ b/quickwit/quickwit-indexing/src/metrics.rs @@ -23,6 +23,8 @@ pub(crate) const ACTOR_NAME: LabelNames<1> = label_names!("actor_name"); pub(crate) const COMPONENT: LabelNames<1> = label_names!("component"); pub(crate) const INDEX_SOURCE: LabelNames<2> = label_names!("index", "source"); pub(crate) const PUBLISHED_SPLIT: LabelNames<3> = label_names!("index", "source", "merge_ops"); +pub(crate) const PUBLISHED_SPLIT_DOCS: LabelNames<4> = + label_names!("index", "source", "merge_ops", "publication_type"); pub(crate) static PROCESSED_DOCS_TOTAL: LazyCounter = lazy_counter!( name: "processed_docs_total", @@ -51,7 +53,7 @@ pub(crate) static PUBLISHED_SPLIT_BYTES_TOTAL: LazyCounter = lazy_counter!( pub(crate) static PUBLISHED_SPLIT_DOCS_TOTAL: LazyCounter = lazy_counter!( name: "published_split_docs_total", - description: "Documents in successfully published splits.", + description: "Documents in successfully published splits by publication type.", subsystem: "indexing", ); @@ -76,9 +78,14 @@ pub(crate) static PUBLISHED_SPLIT_SIZE_BYTES: LazyHistogram = lazy_histogram!( /// Records one split after the metastore has successfully published it. /// -/// All metrics deliberately use the same labels so ratios computed over a time -/// window describe the same set of splits. -pub(crate) fn record_published_split(index_id: &str, split: &SplitMetadata) { +/// The document counter additionally distinguishes initial publications from +/// replacements, so consumers can exclude documents that were republished by +/// merges or delete operations. +pub(crate) fn record_published_split( + index_id: &str, + split: &SplitMetadata, + publication_type: &'static str, +) { let index = quickwit_common::metrics::index_label(index_id); let labels = label_values!( PUBLISHED_SPLIT => @@ -86,11 +93,18 @@ pub(crate) fn record_published_split(index_id: &str, split: &SplitMetadata) { split.source_id.to_string(), split.num_merge_ops.to_string() ); + let docs_labels = label_values!( + PUBLISHED_SPLIT_DOCS => + index.to_string(), + split.source_id.to_string(), + split.num_merge_ops.to_string(), + publication_type + ); let split_size_bytes = split.footer_offsets.end; counter!(parent: PUBLISHED_SPLIT_BYTES_TOTAL, labels: [labels.clone()]) .inc_by(split_size_bytes); - counter!(parent: PUBLISHED_SPLIT_DOCS_TOTAL, labels: [labels.clone()]) + counter!(parent: PUBLISHED_SPLIT_DOCS_TOTAL, labels: [docs_labels]) .inc_by(split.num_docs as u64); counter!(parent: PUBLISHED_SPLITS_TOTAL, labels: [labels.clone()]).inc(); counter!(parent: PUBLISHED_SPLIT_UNCOMPRESSED_BYTES_TOTAL, labels: [labels.clone()]) From c1c32944c6076639f6d9c9b7e4e5ede3ace3bc43 Mon Sep 17 00:00:00 2001 From: Arnaud Robert Date: Fri, 31 Jul 2026 16:53:55 +0200 Subject: [PATCH 2/3] indexing: label all published split metrics --- .../src/actors/log_publisher_impl.rs | 4 ++-- quickwit/quickwit-indexing/src/metrics.rs | 24 +++++++------------ 2 files changed, 10 insertions(+), 18 deletions(-) diff --git a/quickwit/quickwit-indexing/src/actors/log_publisher_impl.rs b/quickwit/quickwit-indexing/src/actors/log_publisher_impl.rs index db3943be536..0a9507519a6 100644 --- a/quickwit/quickwit-indexing/src/actors/log_publisher_impl.rs +++ b/quickwit/quickwit-indexing/src/actors/log_publisher_impl.rs @@ -175,7 +175,7 @@ mod tests { use super::PUBLISHER_NAME; use crate::actors::publisher::Publisher; - use crate::metrics::{PUBLISHED_SPLIT_DOCS, PUBLISHED_SPLIT_DOCS_TOTAL}; + use crate::metrics::{PUBLISHED_SPLIT, PUBLISHED_SPLIT_DOCS_TOTAL}; use crate::models::{PublishLock, SharedPublishToken, SplitsUpdate}; use crate::source::SuggestTruncate; @@ -187,7 +187,7 @@ mod tests { ) -> u64 { let index = quickwit_common::metrics::index_label(index_id); let labels = label_values!( - PUBLISHED_SPLIT_DOCS => + PUBLISHED_SPLIT => index.to_string(), source_id.to_string(), merge_ops.to_string(), diff --git a/quickwit/quickwit-indexing/src/metrics.rs b/quickwit/quickwit-indexing/src/metrics.rs index 4208d25b536..e76953446c2 100644 --- a/quickwit/quickwit-indexing/src/metrics.rs +++ b/quickwit/quickwit-indexing/src/metrics.rs @@ -22,8 +22,7 @@ use quickwit_metrics::{ pub(crate) const ACTOR_NAME: LabelNames<1> = label_names!("actor_name"); pub(crate) const COMPONENT: LabelNames<1> = label_names!("component"); pub(crate) const INDEX_SOURCE: LabelNames<2> = label_names!("index", "source"); -pub(crate) const PUBLISHED_SPLIT: LabelNames<3> = label_names!("index", "source", "merge_ops"); -pub(crate) const PUBLISHED_SPLIT_DOCS: LabelNames<4> = +pub(crate) const PUBLISHED_SPLIT: LabelNames<4> = label_names!("index", "source", "merge_ops", "publication_type"); pub(crate) static PROCESSED_DOCS_TOTAL: LazyCounter = lazy_counter!( @@ -47,7 +46,7 @@ pub(crate) static DOCS_SORT_GROUP_SIZE: LazyHistogram = lazy_histogram!( pub(crate) static PUBLISHED_SPLIT_BYTES_TOTAL: LazyCounter = lazy_counter!( name: "published_split_bytes_total", - description: "Compressed bytes in successfully published splits.", + description: "Compressed bytes in successfully published splits by publication type.", subsystem: "indexing", ); @@ -59,28 +58,27 @@ pub(crate) static PUBLISHED_SPLIT_DOCS_TOTAL: LazyCounter = lazy_counter!( pub(crate) static PUBLISHED_SPLITS_TOTAL: LazyCounter = lazy_counter!( name: "published_splits_total", - description: "Number of successfully published splits.", + description: "Number of successfully published splits by publication type.", subsystem: "indexing", ); pub(crate) static PUBLISHED_SPLIT_UNCOMPRESSED_BYTES_TOTAL: LazyCounter = lazy_counter!( name: "published_split_uncompressed_bytes_total", - description: "Uncompressed document bytes in successfully published splits.", + description: "Uncompressed document bytes in successfully published splits by publication type.", subsystem: "indexing", ); pub(crate) static PUBLISHED_SPLIT_SIZE_BYTES: LazyHistogram = lazy_histogram!( name: "published_split_size_bytes", - description: "Compressed size in bytes of successfully published splits.", + description: "Compressed size in bytes of successfully published splits by publication type.", subsystem: "indexing", buckets: exponential_buckets(1_000_000.0, 2.0, 14).unwrap(), ); /// Records one split after the metastore has successfully published it. /// -/// The document counter additionally distinguishes initial publications from -/// replacements, so consumers can exclude documents that were republished by -/// merges or delete operations. +/// All metrics distinguish initial publications from replacements, so consumers +/// can exclude documents that were republished by merges or delete operations. pub(crate) fn record_published_split( index_id: &str, split: &SplitMetadata, @@ -91,12 +89,6 @@ pub(crate) fn record_published_split( PUBLISHED_SPLIT => index.to_string(), split.source_id.to_string(), - split.num_merge_ops.to_string() - ); - let docs_labels = label_values!( - PUBLISHED_SPLIT_DOCS => - index.to_string(), - split.source_id.to_string(), split.num_merge_ops.to_string(), publication_type ); @@ -104,7 +96,7 @@ pub(crate) fn record_published_split( counter!(parent: PUBLISHED_SPLIT_BYTES_TOTAL, labels: [labels.clone()]) .inc_by(split_size_bytes); - counter!(parent: PUBLISHED_SPLIT_DOCS_TOTAL, labels: [docs_labels]) + counter!(parent: PUBLISHED_SPLIT_DOCS_TOTAL, labels: [labels.clone()]) .inc_by(split.num_docs as u64); counter!(parent: PUBLISHED_SPLITS_TOTAL, labels: [labels.clone()]).inc(); counter!(parent: PUBLISHED_SPLIT_UNCOMPRESSED_BYTES_TOTAL, labels: [labels.clone()]) From 722bbdb0e8269d93894a80719e77b21c8057f8d8 Mon Sep 17 00:00:00 2001 From: Arnaud Robert Date: Fri, 31 Jul 2026 17:33:34 +0200 Subject: [PATCH 3/3] [Nit] Comments --- quickwit/quickwit-indexing/src/metrics.rs | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/quickwit/quickwit-indexing/src/metrics.rs b/quickwit/quickwit-indexing/src/metrics.rs index e76953446c2..6d2ca572f4f 100644 --- a/quickwit/quickwit-indexing/src/metrics.rs +++ b/quickwit/quickwit-indexing/src/metrics.rs @@ -46,39 +46,39 @@ pub(crate) static DOCS_SORT_GROUP_SIZE: LazyHistogram = lazy_histogram!( pub(crate) static PUBLISHED_SPLIT_BYTES_TOTAL: LazyCounter = lazy_counter!( name: "published_split_bytes_total", - description: "Compressed bytes in successfully published splits by publication type.", + description: "Compressed bytes in successfully published splits.", subsystem: "indexing", ); pub(crate) static PUBLISHED_SPLIT_DOCS_TOTAL: LazyCounter = lazy_counter!( name: "published_split_docs_total", - description: "Documents in successfully published splits by publication type.", + description: "Documents in successfully published splits.", subsystem: "indexing", ); pub(crate) static PUBLISHED_SPLITS_TOTAL: LazyCounter = lazy_counter!( name: "published_splits_total", - description: "Number of successfully published splits by publication type.", + description: "Number of successfully published splits.", subsystem: "indexing", ); pub(crate) static PUBLISHED_SPLIT_UNCOMPRESSED_BYTES_TOTAL: LazyCounter = lazy_counter!( name: "published_split_uncompressed_bytes_total", - description: "Uncompressed document bytes in successfully published splits by publication type.", + description: "Uncompressed document bytes in successfully published splits.", subsystem: "indexing", ); pub(crate) static PUBLISHED_SPLIT_SIZE_BYTES: LazyHistogram = lazy_histogram!( name: "published_split_size_bytes", - description: "Compressed size in bytes of successfully published splits by publication type.", + description: "Compressed size in bytes of successfully published splits.", subsystem: "indexing", buckets: exponential_buckets(1_000_000.0, 2.0, 14).unwrap(), ); /// Records one split after the metastore has successfully published it. /// -/// All metrics distinguish initial publications from replacements, so consumers -/// can exclude documents that were republished by merges or delete operations. +/// All metrics deliberately use the same labels so ratios computed over a time +/// window describe the same set of splits. pub(crate) fn record_published_split( index_id: &str, split: &SplitMetadata,