Add CLI support for parquet GC, IndexService method, and OTEL metrics retention config

Author: mattmkim

quickwit/quickwit-cli/src/tool.rs

@@ -660,51 +660,101 @@ pub async fn garbage_collect_index_cli(args: GarbageCollectIndexArgs) -> anyhow:
         get_resolvers(&config.storage_configs, &config.metastore_configs);
     let metastore = metastore_resolver.resolve(&config.metastore_uri).await?;
     let mut index_service = IndexService::new(metastore, storage_resolver);
+
+    if quickwit_common::is_metrics_index(&args.index_id) {
+        let removal_info = index_service
+            .garbage_collect_parquet_index(&args.index_id, args.grace_period, args.dry_run)
+            .await?;
+        return print_parquet_gc_result(args.dry_run, removal_info);
+    }
+
     let removal_info = index_service
         .garbage_collect_index(&args.index_id, args.grace_period, args.dry_run)
         .await?;
+    print_tantivy_gc_result(args.dry_run, removal_info)
+}
+
+fn print_tantivy_gc_result(
+    dry_run: bool,
+    removal_info: quickwit_index_management::SplitRemovalInfo,
+) -> anyhow::Result<()> {
     if removal_info.removed_split_entries.is_empty() && removal_info.failed_splits.is_empty() {
         println!("No dangling files to garbage collect.");
         return Ok(());
     }
 
-    if args.dry_run {
+    if dry_run {
         println!("The following files will be garbage collected.");
-        for split_info in removal_info.removed_split_entries {
-            println!(" - {}", split_info.file_name.display());
+        for entry in &removal_info.removed_split_entries {
+            println!(" - {}", entry.file_name.display());
         }
         return Ok(());
     }
 
     if !removal_info.failed_splits.is_empty() {
         println!("The following splits were attempted to be removed, but failed.");
-        for split_info in &removal_info.failed_splits {
-            println!(" - {}", split_info.split_id);
+        for split in &removal_info.failed_splits {
+            println!(" - {}", split.split_id);
         }
-        println!(
-            "{} Splits were unable to be removed.",
-            removal_info.failed_splits.len()
-        );
+        println!("{} Splits were unable to be removed.", removal_info.failed_splits.len());
     }
 
     let deleted_bytes: u64 = removal_info
         .removed_split_entries
         .iter()
-        .map(|split_info| split_info.file_size_bytes.as_u64())
+        .map(|s| s.file_size_bytes.as_u64())
         .sum();
-    println!(
-        "{}MB of storage garbage collected.",
-        deleted_bytes / 1_000_000
-    );
+    println!("{}MB of storage garbage collected.", deleted_bytes / 1_000_000);
 
     if removal_info.failed_splits.is_empty() {
+        println!("{} Index successfully garbage collected.", "✔".color(GREEN_COLOR));
+    } else if removal_info.removed_split_entries.is_empty() {
+        println!("{} Failed to garbage collect index.", "✘".color(RED_COLOR));
+    } else {
         println!(
-            "{} Index successfully garbage collected.",
-            "✔".color(GREEN_COLOR)
+            "{} Index partially garbage collected.",
+            "✘".color(RED_COLOR)
         );
-    } else if removal_info.removed_split_entries.is_empty()
-        && !removal_info.failed_splits.is_empty()
+    }
+
+    Ok(())
+}
+
+fn print_parquet_gc_result(
+    dry_run: bool,
+    removal_info: quickwit_index_management::ParquetSplitRemovalInfo,
+) -> anyhow::Result<()> {
+    if removal_info.removed_parquet_splits_entries.is_empty()
+        && removal_info.failed_parquet_splits.is_empty()
     {
+        println!("No dangling files to garbage collect.");
+        return Ok(());
+    }
+
+    if dry_run {
+        println!("The following files will be garbage collected.");
+        for entry in &removal_info.removed_parquet_splits_entries {
+            println!(" - {}.parquet", entry.split_id);
+        }
+        return Ok(());
+    }
+
+    if !removal_info.failed_parquet_splits.is_empty() {
+        println!("The following splits were attempted to be removed, but failed.");
+        for split in &removal_info.failed_parquet_splits {
+            println!(" - {}", split.split_id);
+        }
+        println!(
+            "{} Splits were unable to be removed.",
+            removal_info.failed_parquet_splits.len()
+        );
+    }
+
+    println!("{}MB of storage garbage collected.", removal_info.removed_bytes() / 1_000_000);
+
+    if removal_info.failed_parquet_splits.is_empty() {
+        println!("{} Index successfully garbage collected.", "✔".color(GREEN_COLOR));
+    } else if removal_info.removed_parquet_splits_entries.is_empty() {
         println!("{} Failed to garbage collect index.", "✘".color(RED_COLOR));
     } else {
         println!(

quickwit/quickwit-index-management/src/index.rs

@@ -45,6 +45,7 @@ use crate::garbage_collection::{
     DeleteSplitsError, SplitRemovalInfo, delete_splits_from_storage_and_metastore,
     run_garbage_collect,
 };
+use crate::parquet_garbage_collection::{ParquetSplitRemovalInfo, run_parquet_garbage_collect};
 
 #[derive(Error, Debug)]
 pub enum IndexServiceError {
@@ -405,6 +406,47 @@ impl IndexService {
         Ok(deleted_entries)
     }
 
+    /// Detect all dangling parquet splits and associated files from a metrics index
+    /// and removes them.
+    ///
+    /// * `index_id` - The target metrics index Id.
+    /// * `grace_period` - Threshold period after which a staged split can be garbage collected.
+    /// * `dry_run` - Should this only return a list of affected files without performing deletion.
+    pub async fn garbage_collect_parquet_index(
+        &mut self,
+        index_id: &str,
+        grace_period: Duration,
+        dry_run: bool,
+    ) -> anyhow::Result<ParquetSplitRemovalInfo> {
+        let index_metadata_request = IndexMetadataRequest::for_index_id(index_id.to_string());
+        let index_metadata = self
+            .metastore
+            .index_metadata(index_metadata_request)
+            .await?
+            .deserialize_index_metadata()?;
+        let index_uid = index_metadata.index_uid.clone();
+        let index_config = index_metadata.into_index_config();
+        let storage = self
+            .storage_resolver
+            .resolve(&index_config.index_uri)
+            .await?;
+
+        let deleted_entries = run_parquet_garbage_collect(
+            HashMap::from([(index_uid, storage)]),
+            self.metastore.clone(),
+            grace_period,
+            // deletion_grace_period of zero, so that a cli call directly deletes splits after
+            // marking to be deleted.
+            Duration::ZERO,
+            dry_run,
+            None,
+            None,
+        )
+        .await?;
+
+        Ok(deleted_entries)
+    }
+
     /// Clears the index by applying the following actions:
     /// - mark all splits for deletion in the metastore.
     /// - delete the files of all splits marked for deletion using garbage collection.

quickwit/quickwit-index-management/src/lib.rs

@@ -16,7 +16,7 @@ mod garbage_collection;
 mod index;
 mod parquet_garbage_collection;
 
-pub use garbage_collection::{GcMetrics, run_garbage_collect};
+pub use garbage_collection::{GcMetrics, SplitRemovalInfo, run_garbage_collect};
 pub use index::{IndexService, IndexServiceError, clear_cache_directory, validate_storage_uri};
 pub use parquet_garbage_collection::{
     ParquetSplitInfo, ParquetSplitRemovalInfo, run_parquet_garbage_collect,

quickwit/quickwit-opentelemetry/src/otlp/otel_metrics.rs

@@ -47,17 +47,29 @@ pub const OTEL_METRICS_INDEX_ID: &str = "otel-metrics-v0_9";
 /// directly and queries via DataFusion. The doc mapping is unused; this config
 /// exists only so the index can be registered in the metastore for source
 /// assignment and lifecycle management.
+/// 
+/// TODO: As a temporary hack, we are including a timestamp_field, so that 
+/// we can pass the retention policy validation. 
 const OTEL_METRICS_INDEX_CONFIG: &str = r#"
 version: 0.8
 
 index_id: ${INDEX_ID}
 
 doc_mapping:
   mode: dynamic
+  field_mappings:
+    - name: timestamp
+      type: datetime
+      fast: true
+  timestamp_field: timestamp
 
 indexing_settings:
   commit_timeout_secs: 15
 
+retention:
+  period: 30 days
+  schedule: hourly
+
 search_settings:
   default_search_fields: []
 "#;
@@ -542,6 +554,12 @@ mod tests {
         let index_config =
             OtlpGrpcMetricsService::index_config(&Uri::for_test("ram:///indexes")).unwrap();
         assert_eq!(index_config.index_id, OTEL_METRICS_INDEX_ID);
+        let retention = index_config.retention_policy_opt.expect("retention policy should be set");
+        assert_eq!(
+            retention.retention_period().unwrap(),
+            std::time::Duration::from_secs(30 * 24 * 3600)
+        );
+        assert_eq!(retention.evaluation_schedule, "hourly");
     }
 
     #[tokio::test]