Generalize logging in compactor (#26398)

sjudd · Convex, Inc. · commit c553c3cf5883 · 2024-05-29T23:10:10.000Z
GitOrigin-RevId: 734bbdf093bf243ba9efaad842ea612e384dac83
diff --git a/crates/database/src/metrics.rs b/crates/database/src/metrics.rs
@@ -673,7 +673,7 @@ fn log_worker_status(is_working: bool, name: &'static str) {
 
 register_convex_histogram!(
     SEARCH_AND_VECTOR_BOOTSTRAP_SECONDS,
-    "Time taken to bootstrap search and vector indexes",
+    "Time taken to bootstrap text and vector indexes",
     &STATUS_LABEL
 );
 pub fn bootstrap_timer() -> StatusTimer {
@@ -682,18 +682,18 @@ pub fn bootstrap_timer() -> StatusTimer {
 
 register_convex_histogram!(
     SEARCH_AND_VECTOR_BOOTSTRAP_COMMITTER_UPDATE_SECONDS,
-    "Time to update search and vector index bootstrap in the committer"
+    "Time to update text and vector index bootstrap in the committer"
 );
 pub fn bootstrap_update_timer() -> Timer<VMHistogram> {
     Timer::new(&SEARCH_AND_VECTOR_BOOTSTRAP_COMMITTER_UPDATE_SECONDS)
 }
 register_convex_counter!(
     SEARCH_AND_VECTOR_BOOTSTRAP_COMMITTER_UPDATE_REVISIONS_TOTAL,
-    "Number of revisions loaded during search and vector bootstrap updates in the committer"
+    "Number of revisions loaded during text and vector bootstrap updates in the committer"
 );
 register_convex_counter!(
     SEARCH_AND_VECTOR_BOOTSTRAP_COMMITTER_UPDATE_REVISIONS_BYTES,
-    "Total size of revisions loaded during search and vector bootstrap updates in the committer"
+    "Total size of revisions loaded during text and vector bootstrap updates in the committer"
 );
 
 pub fn finish_bootstrap_update(num_revisions: usize, bytes: usize) {
@@ -709,11 +709,11 @@ pub fn finish_bootstrap_update(num_revisions: usize, bytes: usize) {
 
 register_convex_counter!(
     SEARCH_AND_VECTOR_BOOTSTRAP_REVISIONS_TOTAL,
-    "Number of revisions loaded during search and vector bootstrap"
+    "Number of revisions loaded during text and vector bootstrap"
 );
 register_convex_counter!(
     SEARCH_AND_VECTOR_BOOTSTRAP_REVISIONS_BYTES,
-    "Total size of revisions loaded during search and vector bootstrap"
+    "Total size of revisions loaded during text and vector bootstrap"
 );
 pub fn finish_bootstrap(num_revisions: usize, bytes: usize, timer: StatusTimer) {
     log_counter(
@@ -877,6 +877,76 @@ pub fn log_non_deleted_documents_per_search_index(count: u64, search_type: Searc
     );
 }
 
+const COMPACTION_REASON_LABEL: &str = "compaction_reason";
+
+pub enum CompactionReason {
+    Unknown,
+    SmallSegments,
+    LargeSegments,
+    Deletes,
+}
+
+impl CompactionReason {
+    fn metric_label(&self) -> StaticMetricLabel {
+        let label = match self {
+            CompactionReason::Unknown => "unknown",
+            CompactionReason::SmallSegments => "small",
+            CompactionReason::LargeSegments => "large",
+            CompactionReason::Deletes => "deletes",
+        };
+        StaticMetricLabel::new(COMPACTION_REASON_LABEL, label)
+    }
+}
+
+register_convex_histogram!(
+    COMPACTION_BUILD_ONE_SECONDS,
+    "Time to run a single vector/text index compaction",
+    &[STATUS_LABEL[0], COMPACTION_REASON_LABEL, SEARCH_TYPE_LABEL],
+);
+pub fn compaction_build_one_timer(search_type: SearchType) -> StatusTimer {
+    let mut timer = StatusTimer::new(&COMPACTION_BUILD_ONE_SECONDS);
+    timer.add_label(CompactionReason::Unknown.metric_label());
+    timer.add_label(search_type.tag());
+    timer
+}
+
+pub fn finish_compaction_timer(mut timer: StatusTimer, reason: CompactionReason) {
+    timer.replace_label(
+        CompactionReason::Unknown.metric_label(),
+        reason.metric_label(),
+    );
+    timer.finish();
+}
+
+register_convex_histogram!(
+    COMPACTION_COMPACTED_SEGMENTS_TOTAL,
+    "Total number of compacted segments",
+    &[SEARCH_TYPE_LABEL],
+);
+pub fn log_compaction_total_segments(total_segments: usize, search_type: SearchType) {
+    log_distribution_with_labels(
+        &COMPACTION_COMPACTED_SEGMENTS_TOTAL,
+        total_segments as f64,
+        vec![search_type.tag()],
+    );
+}
+
+register_convex_histogram!(
+    COMPACTION_COMPACTED_SEGMENT_NUM_DOCUMENTS_TOTAL,
+    "The number of documents in the newly generated compacted segment",
+    &[SEARCH_TYPE_LABEL],
+);
+pub fn log_compaction_compacted_segment_num_documents_total(
+    total_vectors: u64,
+    search_type: SearchType,
+) {
+    log_distribution_with_labels(
+        &COMPACTION_COMPACTED_SEGMENT_NUM_DOCUMENTS_TOTAL,
+        total_vectors as f64,
+        vec![search_type.tag()],
+    );
+}
+
 pub mod search {
 
     use metrics::{
@@ -941,7 +1011,7 @@ pub mod search {
 
 register_convex_histogram!(
     DATABASE_VECTOR_AND_SEARCH_BOOTSTRAP_SECONDS,
-    "Time to bootstrap vector and search indexes",
+    "Time to bootstrap vector and text indexes",
     &STATUS_LABEL
 );
 pub fn search_and_vector_bootstrap_timer() -> StatusTimer {
@@ -950,18 +1020,16 @@ pub fn search_and_vector_bootstrap_timer() -> StatusTimer {
 
 register_convex_counter!(
     SEARCH_AND_VECTOR_BOOTSTRAP_DOCUMENTS_SKIPPED_TOTAL,
-    "Number of documents skipped during vector and search index bootstrap",
+    "Number of documents skipped during vector and text index bootstrap",
 );
 pub fn log_document_skipped() {
     log_counter(&SEARCH_AND_VECTOR_BOOTSTRAP_DOCUMENTS_SKIPPED_TOTAL, 1);
 }
 
 pub mod vector {
     use metrics::{
-        log_distribution,
         register_convex_histogram,
         CancelableTimer,
-        StaticMetricLabel,
         StatusTimer,
         STATUS_LABEL,
     };
@@ -992,66 +1060,4 @@ pub mod vector {
     pub fn vector_search_with_retries_timer() -> CancelableTimer {
         CancelableTimer::new(&DATABASE_VECTOR_SEARCH_WITH_RETRIES_QUERY_SECONDS)
     }
-
-    const COMPACTION_REASON_LABEL: &str = "compaction_reason";
-
-    pub enum CompactionReason {
-        Unknown,
-        SmallSegments,
-        LargeSegments,
-        Deletes,
-    }
-
-    impl CompactionReason {
-        fn metric_label(&self) -> StaticMetricLabel {
-            let label = match self {
-                CompactionReason::Unknown => "unknown",
-                CompactionReason::SmallSegments => "small",
-                CompactionReason::LargeSegments => "large",
-                CompactionReason::Deletes => "deletes",
-            };
-            StaticMetricLabel::new(COMPACTION_REASON_LABEL, label)
-        }
-    }
-
-    register_convex_histogram!(
-        VECTOR_COMPACTION_BUILD_ONE_SECONDS,
-        "Time to run a single vector compaction",
-        &[STATUS_LABEL[0], COMPACTION_REASON_LABEL],
-    );
-    pub fn vector_compaction_build_one_timer() -> StatusTimer {
-        let mut timer = StatusTimer::new(&VECTOR_COMPACTION_BUILD_ONE_SECONDS);
-        timer.add_label(CompactionReason::Unknown.metric_label());
-        timer
-    }
-
-    pub fn finish_compaction_timer(mut timer: StatusTimer, reason: CompactionReason) {
-        timer.replace_label(
-            CompactionReason::Unknown.metric_label(),
-            reason.metric_label(),
-        );
-        timer.finish();
-    }
-
-    register_convex_histogram!(
-        VECTOR_COMPACTION_COMPACTED_SEGMENTS_TOTAL,
-        "Total number of compacted segments",
-    );
-    pub fn log_vector_compaction_total_segments(total_segments: usize) {
-        log_distribution(
-            &VECTOR_COMPACTION_COMPACTED_SEGMENTS_TOTAL,
-            total_segments as f64,
-        );
-    }
-
-    register_convex_histogram!(
-        VECTOR_COMPACTION_COMPACTED_SEGMENT_NUM_VECTORS_TOTAL,
-        "Size of the newly generated compacted segment",
-    );
-    pub fn log_vector_compaction_compacted_segment_num_vectors_total(total_vectors: u64) {
-        log_distribution(
-            &VECTOR_COMPACTION_COMPACTED_SEGMENT_NUM_VECTORS_TOTAL,
-            total_vectors as f64,
-        );
-    }
 }
diff --git a/crates/database/src/vector_index_worker/compactor.rs b/crates/database/src/vector_index_worker/compactor.rs
@@ -17,7 +17,10 @@ use common::{
 };
 use itertools::Itertools;
 use keybroker::Identity;
-use search::searcher::Searcher;
+use search::{
+    metrics::SearchType,
+    searcher::Searcher,
+};
 use storage::Storage;
 use value::ResolvedDocumentId;
 
@@ -35,11 +38,11 @@ use crate::{
         },
         writer::SearchIndexMetadataWriter,
     },
-    metrics::vector::{
+    metrics::{
+        compaction_build_one_timer,
         finish_compaction_timer,
-        log_vector_compaction_compacted_segment_num_vectors_total,
-        log_vector_compaction_total_segments,
-        vector_compaction_build_one_timer,
+        log_compaction_compacted_segment_num_documents_total,
+        log_compaction_total_segments,
         CompactionReason,
     },
     vector_index_worker::vector_meta::{
@@ -113,13 +116,17 @@ impl<RT: Runtime, T: SearchIndexConfigParser> VectorIndexCompactor<RT, T> {
         Ok(())
     }
 
+    fn search_type() -> SearchType {
+        <T::IndexType as SearchIndex>::search_type()
+    }
+
     pub(crate) async fn step(&self) -> anyhow::Result<(BTreeMap<TabletIndexName, u64>, Token)> {
         let mut metrics = BTreeMap::new();
 
         let (to_build, token) = self.needs_compaction().await?;
         let num_to_build = to_build.len();
         if num_to_build > 0 {
-            tracing::info!("{num_to_build} vector indexes to build");
+            tracing::info!("{num_to_build} {:?} indexes to build", Self::search_type());
         }
 
         for job in to_build {
@@ -129,7 +136,7 @@ impl<RT: Runtime, T: SearchIndexConfigParser> VectorIndexCompactor<RT, T> {
         }
 
         if num_to_build > 0 {
-            tracing::info!("built {num_to_build} vector indexes");
+            tracing::info!("built {num_to_build} {:?} indexes", Self::search_type());
         }
 
         Ok((metrics, token))
@@ -178,7 +185,10 @@ impl<RT: Runtime, T: SearchIndexConfigParser> VectorIndexCompactor<RT, T> {
                 _ => false,
             };
             if needs_compaction {
-                tracing::info!("Queueing vector index for compaction: {name:?}");
+                tracing::info!(
+                    "Queueing {:?} index for compaction: {name:?}",
+                    Self::search_type()
+                );
                 let job = CompactionJob {
                     index_id,
                     index_name: name.clone(),
@@ -192,26 +202,36 @@ impl<RT: Runtime, T: SearchIndexConfigParser> VectorIndexCompactor<RT, T> {
     }
 
     async fn build_one(&self, job: CompactionJob<T::IndexType>) -> anyhow::Result<u64> {
-        let timer = vector_compaction_build_one_timer();
+        let timer = compaction_build_one_timer(Self::search_type());
         let (segments, snapshot_ts) = match job.on_disk_state {
             SearchOnDiskState::Backfilling(BackfillState {
                 segments,
                 backfill_snapshot_ts,
                 ..
             }) => {
-                let ts = backfill_snapshot_ts
-                    .context("Trying to compact backfilling segments with no backfill timestamp")?;
+                let ts = backfill_snapshot_ts.with_context(|| {
+                    format!(
+                        "Trying to compact backfilling {:?} segments with no backfill timestamp",
+                        Self::search_type()
+                    )
+                })?;
                 (segments, ts)
             },
             SearchOnDiskState::Backfilled(snapshot)
             | SearchOnDiskState::SnapshottedAt(snapshot) => {
                 let segments = match snapshot.data {
                     SnapshotData::Unknown(_) => {
-                        anyhow::bail!("Trying to compact unknown vector snapshot");
+                        anyhow::bail!(
+                            "Trying to compact unknown {:?} snapshot",
+                            Self::search_type()
+                        );
                     },
                     SnapshotData::MultiSegment(segments) => segments,
                     SnapshotData::SingleSegment(_) => {
-                        anyhow::bail!("Trying to compact a single segment index!");
+                        anyhow::bail!(
+                            "Trying to compact a single segment {:?} index!",
+                            Self::search_type()
+                        );
                     },
                 };
                 (segments, snapshot.ts)
@@ -223,15 +243,15 @@ impl<RT: Runtime, T: SearchIndexConfigParser> VectorIndexCompactor<RT, T> {
         anyhow::ensure!(segments_to_compact.len() > 0);
 
         let total_compacted_segments = segments_to_compact.len();
-        log_vector_compaction_total_segments(total_compacted_segments);
+        log_compaction_total_segments(total_compacted_segments, Self::search_type());
 
         let new_segment = self
             .compact(&job.developer_config, &segments_to_compact)
             .await?;
         let stats = new_segment.statistics()?;
 
-        let total_vectors = stats.num_non_deleted_documents();
-        log_vector_compaction_compacted_segment_num_vectors_total(total_vectors);
+        let total_documents = stats.num_documents();
+        log_compaction_compacted_segment_num_documents_total(total_documents, Self::search_type());
 
         self.writer
             .commit_compaction(
@@ -268,11 +288,12 @@ impl<RT: Runtime, T: SearchIndexConfigParser> VectorIndexCompactor<RT, T> {
     ) -> anyhow::Result<String> {
         let stats = segment.statistics()?;
         Ok(format!(
-            "(id: {}, size: {}, vectors: {}, deletes: {})",
+            "(id: {}, size: {}, documents : {}, deletes: {}, type: {:?})",
             segment.id(),
             segment.total_size_bytes(developer_config)?,
             stats.num_documents(),
             stats.num_deleted_documents(),
+            Self::search_type(),
         ))
     }
 
@@ -366,7 +387,7 @@ impl<RT: Runtime, T: SearchIndexConfigParser> VectorIndexCompactor<RT, T> {
         }
 
         // Finally check to see if any individual segment has a large number of deleted
-        // vectors and if so compact just that segment.
+        // documents and if so compact just that segment.
         let compact_deletes = large_segments
             .into_iter()
             .try_find(|(segment, segment_size_bytes)| {
@@ -425,7 +446,7 @@ impl<RT: Runtime, T: SearchIndexConfigParser> VectorIndexCompactor<RT, T> {
             .try_fold(0u64, |sum, current| {
                 current.and_then(|current| {
                     sum.checked_add(current)
-                        .context("Summing vector sizes overflowed")
+                        .context("Summing document sizes overflowed")
                 })
             })?;
         anyhow::ensure!(
@@ -470,7 +491,6 @@ pub(crate) struct CompactionConfig {
     // only 2 of those can actually be compacted due to the max size restriction, then we don't
     // want to compact yet.
     pub(crate) min_compaction_segments: u64,
-    // 1.1 million vectors at the largest size
     pub(crate) max_segment_size_bytes: u64,
 }
 
diff --git a/crates/search/src/metrics.rs b/crates/search/src/metrics.rs
@@ -445,7 +445,7 @@ pub fn upload_archive_timer(search_file_type: SearchFileType) -> StatusTimer {
     timer
 }
 
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug)]
 pub enum SearchType {
     Vector,
     Text,

Original file line number	Diff line number	Diff line change
`@@ -445,7 +445,7 @@ pub fn upload_archive_timer(search_file_type: SearchFileType) -> StatusTimer {`
`445`	`445`	`timer`
`446`	`446`	`}`
`447`	`447`
`448`		`-#[derive(Clone, Copy)]`
	`448`	`+#[derive(Clone, Copy, Debug)]`
`449`	`449`	`pub enum SearchType {`
`450`	`450`	`Vector,`
`451`	`451`	`Text,`