Go back to a drastic pruning strategy

Author: rdettai

quickwit/quickwit-lambda/src/indexer/environment.rs

@@ -38,3 +38,10 @@ pub static DISABLE_MERGE: Lazy<bool> =
 
 pub static DISABLE_JANITOR: Lazy<bool> =
     Lazy::new(|| var("QW_LAMBDA_DISABLE_JANITOR").is_ok_and(|v| v.as_str() == "true"));
+
+pub static MAX_CHECKPOINTS: Lazy<usize> = Lazy::new(|| {
+    var("QW_LAMBDA_MAX_CHECKPOINTS").map_or(100, |v| {
+        v.parse()
+            .expect("QW_LAMBDA_MAX_CHECKPOINTS must be a positive integer")
+    })
+});

quickwit/quickwit-lambda/src/indexer/ingest/helpers.rs

@@ -24,7 +24,6 @@ use std::path::{Path, PathBuf};
 use anyhow::{bail, Context};
 use chitchat::transport::ChannelTransport;
 use chitchat::FailureDetectorConfig;
-use chrono::Utc;
 use quickwit_actors::{ActorHandle, Mailbox, Universe};
 use quickwit_cluster::{Cluster, ClusterMember};
 use quickwit_common::pubsub::EventBroker;
@@ -49,20 +48,21 @@ use quickwit_metastore::{
 };
 use quickwit_proto::indexing::CpuCapacity;
 use quickwit_proto::metastore::{
-    AddSourceRequest, CreateIndexRequest, DeleteSourceRequest, IndexMetadataRequest,
-    MetastoreError, MetastoreService, MetastoreServiceClient,
+    CreateIndexRequest, DeleteSourceRequest, IndexMetadataRequest, MetastoreError,
+    MetastoreService, MetastoreServiceClient,
 };
 use quickwit_proto::types::{NodeId, PipelineUid};
 use quickwit_search::SearchJobPlacer;
 use quickwit_storage::StorageResolver;
 use quickwit_telemetry::payload::{QuickwitFeature, QuickwitTelemetryInfo, TelemetryEvent};
 use tracing::{debug, info, instrument};
 
-use super::source_id::{
-    create_lambda_source_id, filter_prunable_lambda_source_ids, is_lambda_source_id,
-};
 use crate::environment::INDEX_ID;
-use crate::indexer::environment::{DISABLE_JANITOR, DISABLE_MERGE, INDEX_CONFIG_URI};
+use crate::indexer::environment::{
+    DISABLE_JANITOR, DISABLE_MERGE, INDEX_CONFIG_URI, MAX_CHECKPOINTS,
+};
+
+const LAMBDA_SOURCE_ID: &str = "_ingest-lambda-source";
 
 /// The indexing service needs to update its cluster chitchat state so that the control plane is
 /// aware of the running tasks. We thus create a fake cluster to instantiate the indexing service
@@ -137,60 +137,22 @@ pub(super) async fn send_telemetry() {
     quickwit_telemetry::send_telemetry_event(TelemetryEvent::RunCommand).await;
 }
 
-/// Convert the incomming file path to a source config and save it to the metastore
-///
-/// If a Lambda file source already exists with the same path, format and transform, reuse it.
+/// Convert the incomming file path to a source config
 pub(super) async fn configure_source(
-    metastore: &mut MetastoreServiceClient,
     input_path: PathBuf,
     input_format: SourceInputFormat,
-    index_metadata: &IndexMetadata,
     vrl_script: Option<String>,
 ) -> anyhow::Result<SourceConfig> {
     let transform_config = vrl_script.map(|vrl_script| TransformConfig::new(vrl_script, None));
     let source_params = SourceParams::file(input_path.clone());
-
-    let existing_sources_for_config: Vec<_> = index_metadata
-        .sources
-        .iter()
-        .filter(|(src_id, src_config)| {
-            is_lambda_source_id(src_id)
-                && src_config.source_params == source_params
-                && src_config.input_format == input_format
-                && src_config.transform_config == transform_config
-        })
-        .map(|(src_id, _)| src_id)
-        .collect();
-
-    let source_id = match existing_sources_for_config.len() {
-        0 => create_lambda_source_id(Utc::now()),
-        1 => existing_sources_for_config[0].clone(),
-        n => bail!(
-            "Found {} existing Lambda sources for file {:?}, expected at most 1",
-            n,
-            input_path,
-        ),
-    };
-
-    let src_config = SourceConfig {
-        source_id,
+    Ok(SourceConfig {
+        source_id: LAMBDA_SOURCE_ID.to_owned(),
         num_pipelines: NonZeroUsize::new(1).expect("1 is always non-zero."),
         enabled: true,
         source_params,
         transform_config,
         input_format,
-    };
-
-    if existing_sources_for_config.is_empty() {
-        metastore
-            .add_source(AddSourceRequest {
-                index_uid: Some(index_metadata.index_uid.clone()),
-                source_config_json: serde_json::to_string(&src_config)?,
-            })
-            .await?;
-    }
-
-    Ok(src_config)
+    })
 }
 
 /// Check if the index exists, creating or overwriting it if necessary
@@ -321,26 +283,39 @@ pub(super) async fn spawn_pipelines(
     Ok((indexing_pipeline_handle, merge_pipeline_handle))
 }
 
-/// Delete old Lambda file sources
-pub(super) async fn prune_file_sources(
+/// Prune old Lambda file checkpoints if there are too many
+///
+/// Without pruning checkpoints accumulate indifinitely. This is particularly
+/// problematic when indexing a lot of small files, as the metastore will grow
+/// large even for a small index.
+///
+/// The current implementation just deletes all checkpoints if there are more
+/// than QW_LAMBDA_MAX_CHECKPOINTS. When this purging is performed, the Lambda
+/// indexer might ingest the same file again if it receives a duplicate
+/// notification.
+pub(super) async fn prune_lambda_source(
     metastore: &mut MetastoreServiceClient,
     index_metadata: IndexMetadata,
 ) -> anyhow::Result<()> {
-    let prunable_sources: Vec<_> =
-        filter_prunable_lambda_source_ids(index_metadata.sources.keys())?.collect();
-    info!(
-        existing = index_metadata.sources.len(),
-        prunable = prunable_sources.len(),
-        "prune file sources"
-    );
-    for src_id in prunable_sources {
-        metastore
-            .delete_source(DeleteSourceRequest {
-                index_uid: Some(index_metadata.index_uid.clone()),
-                source_id: src_id.clone(),
-            })
-            .await?;
+    let lambda_checkpoint_opt = index_metadata
+        .checkpoint
+        .source_checkpoint(LAMBDA_SOURCE_ID);
+
+    if let Some(lambda_checkpoint) = lambda_checkpoint_opt {
+        if lambda_checkpoint.num_partitions() > *MAX_CHECKPOINTS {
+            info!(
+                partitions = lambda_checkpoint.num_partitions(),
+                "prune Lambda checkpoints"
+            );
+            metastore
+                .delete_source(DeleteSourceRequest {
+                    index_uid: Some(index_metadata.index_uid.clone()),
+                    source_id: LAMBDA_SOURCE_ID.to_owned(),
+                })
+                .await?;
+        }
     }
+
     Ok(())
 }
 

quickwit/quickwit-lambda/src/indexer/ingest/mod.rs

@@ -18,7 +18,6 @@
 // along with this program. If not, see <http://www.gnu.org/licenses/>.
 
 mod helpers;
-mod source_id;
 
 use std::collections::HashSet;
 use std::path::PathBuf;
@@ -39,7 +38,7 @@ use quickwit_indexing::models::IndexingStatistics;
 use tracing::{debug, info};
 
 use crate::indexer::environment::{CONFIGURATION_TEMPLATE, DISABLE_JANITOR};
-use crate::indexer::ingest::helpers::{prune_file_sources, wait_for_merges};
+use crate::indexer::ingest::helpers::{prune_lambda_source, wait_for_merges};
 use crate::utils::load_node_config;
 
 #[derive(Debug, Eq, PartialEq)]
@@ -67,14 +66,8 @@ pub async fn ingest(args: IngestArgs) -> anyhow::Result<IndexingStatistics> {
     )
     .await?;
 
-    let source_config = configure_source(
-        &mut metastore,
-        args.input_path,
-        args.input_format,
-        &index_metadata,
-        args.vrl_script,
-    )
-    .await?;
+    let source_config =
+        configure_source(args.input_path, args.input_format, args.vrl_script).await?;
 
     let mut services = vec![QuickwitService::Indexer];
     if !*DISABLE_JANITOR {
@@ -99,7 +92,7 @@ pub async fn ingest(args: IngestArgs) -> anyhow::Result<IndexingStatistics> {
     let (indexing_pipeline_handle, merge_pipeline_handle) =
         spawn_pipelines(indexing_service_handle.mailbox(), source_config).await?;
 
-    prune_file_sources(&mut metastore, index_metadata).await?;
+    prune_lambda_source(&mut metastore, index_metadata).await?;
 
     debug!("wait for indexing to complete");
     let statistics = start_statistics_reporting_loop(indexing_pipeline_handle, false).await?;