diff --git a/docs/configuration/node-config.md b/docs/configuration/node-config.md index cbe3f8b0022..eba5411fe14 100644 --- a/docs/configuration/node-config.md +++ b/docs/configuration/node-config.md @@ -244,6 +244,7 @@ This section contains the configuration options for an indexer. The split store | `enable_otlp_endpoint` | If true, enables the OpenTelemetry exporter endpoint to ingest logs and traces via the OpenTelemetry Protocol (OTLP). | `false` | | `cpu_capacity` | Advisory parameter used by the control plane. The value can expressed be in threads (e.g. `2`) or in term of millicpus (`2000m`). The control plane will attempt to schedule indexing pipelines on the different nodes proportionally to the cpu capacity advertised by the indexer. It is NOT used as a limit. All pipelines will be scheduled regardless of whether the cluster has sufficient capacity or not. The control plane does not attempt to spread the work equally when the load is well below the `cpu_capacity`. Users who need a balanced load on all of their indexer nodes can set the `cpu_capacity` to an arbitrarily low value as long as they keep it proportional to the number of threads available. | `num threads available` | | `enable_cooperative_indexing` | Enable sharing resources more efficiently when the number of indexes actively written to is significantly higher than the number of cores but might decrease the overall indexing throughput. | `false` | +| `enable_spread_indexing_pipelines` | Enable spreading indexing pipelines in time to make sure they are not in sync. | `false` | Example: diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index c9daa723982..b52c4d22375 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -210,6 +210,10 @@ pub struct IndexerConfig { pub enable_otlp_endpoint: bool, #[serde(default = "IndexerConfig::default_enable_cooperative_indexing")] pub enable_cooperative_indexing: bool, + /// Spreads the indexing pipelines of a node uniformly in time, so that they don't all use + /// the same resources at the same time. Implied by `enable_cooperative_indexing`. + #[serde(default = "IndexerConfig::default_enable_spread_indexing_pipelines")] + pub enable_spread_indexing_pipelines: bool, #[serde(default = "IndexerConfig::default_cpu_capacity")] pub cpu_capacity: CpuCapacity, /// If true, run Parquet merges through the streaming column-major engine @@ -229,6 +233,10 @@ impl IndexerConfig { false } + fn default_enable_spread_indexing_pipelines() -> bool { + false + } + fn default_enable_otlp_endpoint() -> bool { #[cfg(any(test, feature = "testsuite"))] { @@ -269,6 +277,7 @@ impl IndexerConfig { use quickwit_proto::indexing::PIPELINE_FULL_CAPACITY; let indexer_config = IndexerConfig { enable_cooperative_indexing: false, + enable_spread_indexing_pipelines: false, enable_otlp_endpoint: true, split_store_max_num_bytes: ByteSize::mb(1), split_store_max_num_splits: 3, @@ -286,6 +295,7 @@ impl Default for IndexerConfig { fn default() -> Self { Self { enable_cooperative_indexing: Self::default_enable_cooperative_indexing(), + enable_spread_indexing_pipelines: Self::default_enable_spread_indexing_pipelines(), enable_otlp_endpoint: Self::default_enable_otlp_endpoint(), split_store_max_num_bytes: Self::default_split_store_max_num_bytes(), split_store_max_num_splits: Self::default_split_store_max_num_splits(), diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index f92e28446c0..599ab0b4fd2 100644 --- a/quickwit/quickwit-config/src/node_config/serialize.rs +++ b/quickwit/quickwit-config/src/node_config/serialize.rs @@ -850,6 +850,7 @@ mod tests { merge_concurrency: NonZeroUsize::new(2).unwrap(), cpu_capacity: IndexerConfig::default_cpu_capacity(), enable_cooperative_indexing: false, + enable_spread_indexing_pipelines: false, max_merge_write_throughput: Some(ByteSize::mb(100)), parquet_merge_use_streaming_engine: true, } diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index c8412ce9bdf..87970566a8c 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -51,7 +51,7 @@ use tracing::{Span, debug, info_span, warn}; use ulid::Ulid; use super::IndexSerializer; -use super::cooperative_indexing::{CooperativeIndexingCycle, CooperativeIndexingPeriod}; +use super::indexing_cycle::{IndexingCycle, IndexingPeriod}; use crate::docs_clustering::{DocIdClusterer, Fingerprinter}; use crate::metrics::SPLIT_BUILDERS; use crate::models::{ @@ -84,7 +84,8 @@ pub struct IndexerCounters { pub num_doc_batches_in_workbench: u64, /// Metrics describing the load and indexing performance of the - /// pipeline. This is only updated for cooperative indexers. + /// pipeline. This is only updated for cooperative indexers or + /// when indexing pipeline spreading is enabled. pub pipeline_metrics_opt: Option, } @@ -100,7 +101,7 @@ struct IndexerState { tokenizer_manager: TokenizerManager, max_num_partitions: NonZeroU32, index_settings: IndexSettings, - cooperative_indexing_opt: Option, + indexing_cycle_opt: Option, } impl IndexerState { @@ -199,15 +200,11 @@ impl IndexerState { workbench_id=%workbench_id, ); let indexing_span = info_span!(parent: batch_parent_span.id(), "indexer"); - let cooperative_indexing_period = - if let Some(cooperative_indexing) = &self.cooperative_indexing_opt { - Some( - ctx.protect_future(cooperative_indexing.cooperative_indexing_period()) - .await, - ) - } else { - None - }; + let indexing_period_opt = if let Some(indexing_cycle) = &self.indexing_cycle_opt { + Some(ctx.protect_future(indexing_cycle.indexing_period()).await) + } else { + None + }; let last_delete_opstamp_request = LastDeleteOpstampRequest { index_uid: Some(self.pipeline_id.index_uid.clone()), @@ -239,7 +236,7 @@ impl IndexerState { publish_lock, last_delete_opstamp, memory_usage: GaugeGuard::new(&IN_FLIGHT_INDEX_WRITER, 0.0), - cooperative_indexing_period, + indexing_period_opt, split_builders_guard, }; Ok(workbench) @@ -290,7 +287,7 @@ impl IndexerState { .get_or_create_workbench(indexing_workbench_opt, ctx) .await?; if publish_lock.is_dead() { - // Release indexing permit early. + // Release indexing permit early if there is one. indexing_workbench_opt.take(); return Ok(()); } @@ -366,7 +363,7 @@ struct IndexingWorkbench { // Number of bytes declared as used by tantivy. memory_usage: GaugeGuard, split_builders_guard: GaugeGuard, - cooperative_indexing_period: Option, + indexing_period_opt: Option, } pub struct Indexer { @@ -402,8 +399,8 @@ impl Actor for Indexer { } async fn initialize(&mut self, ctx: &ActorContext) -> Result<(), ActorExitStatus> { - if let Some(cooperative_indexing_cycle) = &self.indexer_state.cooperative_indexing_opt { - let initial_sleep_duration = cooperative_indexing_cycle.initial_sleep_duration(); + if let Some(indexing_cycle) = &self.indexer_state.indexing_cycle_opt { + let initial_sleep_duration = indexing_cycle.initial_sleep_duration(); ctx.pause(); ctx.schedule_self_msg(initial_sleep_duration, Command::Resume); } @@ -418,9 +415,7 @@ impl Actor for Indexer { return Ok(()); }; - let Some(cooperative_indexing_period) = - indexing_workbench.cooperative_indexing_period.take() - else { + let Some(indexing_period) = indexing_workbench.indexing_period_opt.take() else { return Ok(()); }; @@ -430,9 +425,9 @@ impl Actor for Indexer { .map(|split| split.split_attrs.uncompressed_docs_size_in_bytes) .sum::(); - // This also drops the indexing permit. + // This also drops the indexing permit, if any. let (sleep_duration, pipeline_metrics) = - cooperative_indexing_period.end_of_work(uncompressed_num_bytes); + indexing_period.end_of_work(uncompressed_num_bytes); self.counters.pipeline_metrics_opt = Some(pipeline_metrics); @@ -532,6 +527,7 @@ impl Indexer { indexing_directory: TempDirectory, indexing_settings: IndexingSettings, cooperative_indexing_permits_opt: Option>, + spread_indexing_pipelines: bool, index_serializer_mailbox: Mailbox, fingerprinter_opt: Option, ) -> Self { @@ -548,12 +544,12 @@ impl Indexer { // A configured fingerprinter supplies the mapping when the split is finalized. manual_doc_id_mapping: fingerprinter_opt.is_some(), }; - let cooperative_indexing_opt: Option = - cooperative_indexing_permits_opt.map(|cooperative_indexing_permits| { - CooperativeIndexingCycle::new( + let indexing_cycle_opt: Option = + (spread_indexing_pipelines || cooperative_indexing_permits_opt.is_some()).then(|| { + IndexingCycle::new( &pipeline_id, indexing_settings.commit_timeout(), - cooperative_indexing_permits, + cooperative_indexing_permits_opt, ) }); Self { @@ -569,7 +565,7 @@ impl Indexer { tokenizer_manager: tokenizer_manager.tantivy_manager().clone(), index_settings, max_num_partitions: doc_mapper.max_num_partitions(), - cooperative_indexing_opt, + indexing_cycle_opt, }, index_serializer_mailbox, indexing_workbench_opt: None, @@ -778,6 +774,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -919,6 +916,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -997,6 +995,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1081,6 +1080,7 @@ mod tests { indexing_directory, indexing_settings, Some(Arc::new(Semaphore::new(1))), + false, index_serializer_mailbox, None, ); @@ -1170,6 +1170,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1238,6 +1239,7 @@ mod tests { TempDirectory::for_test(), IndexingSettings::for_test(), None, + false, index_serializer_mailbox, Some(Fingerprinter::new( &serde_json::from_value::(serde_json::json!([ @@ -1358,6 +1360,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1457,6 +1460,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1530,6 +1534,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1604,6 +1609,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1670,6 +1676,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1740,6 +1747,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); diff --git a/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs b/quickwit/quickwit-indexing/src/actors/indexing_cycle.rs similarity index 74% rename from quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs rename to quickwit/quickwit-indexing/src/actors/indexing_cycle.rs index f39f97b39d5..fb7aef1d3bc 100644 --- a/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_cycle.rs @@ -27,20 +27,22 @@ const NUDGE_TOLERANCE: Duration = Duration::from_secs(5); // Origin of time. It is used to compute the phase of the pipeline. static ORIGIN_OF_TIME: LazyLock = LazyLock::new(Instant::now); -/// Cooperative indexing is a mechanism to deal with a large amount of pipelines. +/// The indexing cycle is a mechanism to deal with a large amount of pipelines. /// -/// Instead of having all pipelines index concurrently, cooperative indexing: -/// - have them take turn, making sure that at most only N pipelines are indexing at the same time. -/// This has the benefit is reducing RAM using (by having a limited number of `IndexWriter` at the -/// same time), reducing context switching. -/// - keeps the different pipelines work uniformously spread in time. If the system is not at -/// capacity, we prefer to have the indexing pipeline as desynchronized as possible to make sure -/// they don't all use the same resources (disk/cpu/network) at the same time. +/// Instead of having all pipelines index concurrently, the cycle keeps the different +/// pipelines' work uniformly spread in time. If the system is not +/// at capacity, we prefer to have the indexing pipelines as desynchronized as possible +/// to make sure they don't all use the same resources (disk/cpu/network) at the +/// same time. +/// +/// Optionally, it is possible to have them take turns, making sure that at most N pipelines are +/// indexing at the same time. This has the benefit of reducing RAM usage (by having a limited +/// number of `IndexWriter` at the same time), reducing context switching. /// /// It works by: -/// - a semaphore is used to restrict the number of pipelines indexing at the same time. -/// - in the indexer when `on_drain` is called, the indexer will cut a split and "go to sleep" for a -/// given amount of time. +/// - an optional semaphore is used to restrict the number of pipelines indexing at the same time. +/// - in the indexer when `on_drained_messages` is called, the indexer will cut a split and "go to +/// sleep" for a given amount of time. /// /// The key logic is in the computation of that sleep time. /// @@ -56,34 +58,34 @@ static ORIGIN_OF_TIME: LazyLock = LazyLock::new(Instant::now); /// /// Each period of this cycle is divided into three phases. /// - waking [t_wake..t_work_start) acquisition of the period guard (this is instantaneous) -/// acquisition of the semaphore +/// acquisition of the semaphore if any /// - working [t_work_start..t_work_end) /// - sleeping [t=t_work_end..t_sleep_end) /// -/// The idea is to first pick the sleep time to to create a cycle of period +/// The idea is to first pick the sleep time to create a cycle of period /// `commit_timeout`. -/// sleep_time := max(0, commit_timeout - (t_workend - t_wake)) +/// sleep_time := max(0, commit_timeout - (t_work_end - t_wake)) /// /// If the work phase is too long, the regular commit timeout mechanism -/// kicks in an the pipeline will create a split without waiting for the +/// kicks in and the pipeline will create a split without waiting for the /// mailbox to be drained. /// /// We then allow ourselves to tweak the sleep time one way or another by at -/// most two seconds to eventually nudge the system toward the desired phase. -pub(crate) struct CooperativeIndexingCycle { +/// most `NUDGE_TOLERANCE` to eventually nudge the system toward the desired phase. +pub(crate) struct IndexingCycle { target_phase: Duration, commit_timeout: Duration, - indexing_permits: Arc, + indexing_permits_opt: Option>, } -impl CooperativeIndexingCycle { - /// Creates a new cooperative indexing cycle object. +impl IndexingCycle { + /// Creates a new indexing cycle. /// `phase_id` is hashed to compute the target phase. pub fn new( phase_id: &(impl Hash + ?Sized), commit_timeout: Duration, - indexing_permits: Arc, - ) -> CooperativeIndexingCycle { + indexing_permits_opt: Option>, + ) -> IndexingCycle { assert!(commit_timeout.as_millis() > 0); let mut hasher = DefaultHasher::new(); phase_id.hash(&mut hasher); @@ -91,21 +93,21 @@ impl CooperativeIndexingCycle { Self::new_with_phase( Duration::from_millis(target_phase_millis), commit_timeout, - indexing_permits, + indexing_permits_opt, ) } fn new_with_phase( target_phase: Duration, commit_timeout: Duration, - indexing_permits: Arc, - ) -> CooperativeIndexingCycle { - // Force the initial of the origin of time. + indexing_permits_opt: Option>, + ) -> IndexingCycle { + // Force the initialization of the origin of time. let _t0 = *ORIGIN_OF_TIME; - CooperativeIndexingCycle { + IndexingCycle { target_phase, commit_timeout, - indexing_permits, + indexing_permits_opt, } } @@ -125,33 +127,37 @@ impl CooperativeIndexingCycle { Duration::from_millis(initial_sleep_millis) } - pub async fn cooperative_indexing_period(&self) -> CooperativeIndexingPeriod { + pub async fn indexing_period(&self) -> IndexingPeriod { let t_wake = Instant::now(); - let permit = Semaphore::acquire_owned(self.indexing_permits.clone()) - .await - .unwrap(); + let permit_opt = if let Some(permits) = self.indexing_permits_opt.clone() { + Some(permits.acquire_owned().await.unwrap()) + } else { + None + }; + let t_work_start = Instant::now(); - CooperativeIndexingPeriod { + IndexingPeriod { t_wake, t_work_start, commit_timeout: self.commit_timeout, target_phase: self.target_phase, - _permit: permit, + _permit_opt: permit_opt, } } } -pub(crate) struct CooperativeIndexingPeriod { - // measured right before the acquisition of the indexing semaphore +pub(crate) struct IndexingPeriod { + // measured right before the acquisition of the indexing semaphore, if any t_wake: Instant, - // measured after the acquisition of the semaphore. + // measured after the acquisition of the semaphore, if any. Equal to `t_wake` when the cycle + // has no semaphore. t_work_start: Instant, commit_timeout: Duration, target_phase: Duration, - _permit: OwnedSemaphorePermit, + _permit_opt: Option, } -impl CooperativeIndexingPeriod { +impl IndexingPeriod { fn compute_pipeline_metrics( &self, end: Instant, @@ -197,8 +203,8 @@ impl CooperativeIndexingPeriod { } } - /// This drops the indexing permit, allowing another indexer to start indexing. - /// This function also returns the amount of time to sleep until the next period. + /// If an indexing permit is being held, it is released here, allowing another indexer to start + /// indexing. This function also returns the amount of time to sleep until the next period. pub fn end_of_work(self, uncompressed_num_bytes: u64) -> (Duration, PipelineMetrics) { let end = Instant::now(); let sleep_duration = self.compute_sleep_duration(end); @@ -246,18 +252,13 @@ mod tests { for target_phase_secs in [0, 1, 2, 5, 10, 15, 20, 25, 29, 30, 1_000] { for start_time_secs in [0, 1, 2, 5, 10, 15, 20, 25, 29, 30] { let target_phase = Duration::from_secs(target_phase_secs); - let semaphore = Arc::new(Semaphore::new(1)); tokio::time::sleep(Duration::from_secs(start_time_secs)).await; - let cooperative_indexing = CooperativeIndexingCycle::new_with_phase( - target_phase, - Duration::from_secs(30), - semaphore.clone(), - ); - let initial_sleep_duration: Duration = - cooperative_indexing.initial_sleep_duration(); + let indexing_cycle = + IndexingCycle::new_with_phase(target_phase, Duration::from_secs(30), None); + let initial_sleep_duration: Duration = indexing_cycle.initial_sleep_duration(); tokio::time::sleep(initial_sleep_duration).await; - let target_phase_millis = cooperative_indexing.target_phase.as_millis() as i64; - let commit_timeout_ms = cooperative_indexing.commit_timeout.as_millis() as i64; + let target_phase_millis = indexing_cycle.target_phase.as_millis() as i64; + let commit_timeout_ms = indexing_cycle.commit_timeout.as_millis() as i64; let phase_millis = (t0.elapsed().as_millis() as i64 - target_phase_millis) % commit_timeout_ms; assert!(phase_millis >= -100, "{phase_millis}"); @@ -267,12 +268,10 @@ mod tests { } #[tokio::test] - async fn test_cooperative_indexing_simple() { + async fn test_indexing_cycle_simple() { tokio::time::pause(); - let semaphore = Arc::new(Semaphore::new(1)); - let cooperative_indexing = - CooperativeIndexingCycle::new("id", Duration::from_secs(30), semaphore.clone()); - let guard = cooperative_indexing.cooperative_indexing_period().await; + let indexing_cycle = IndexingCycle::new("id", Duration::from_secs(30), None); + let guard = indexing_cycle.indexing_period().await; tokio::time::advance(Duration::from_secs(10)).await; let (sleep_time, metrics) = guard.end_of_work(100_000_000); assert_approx_equal_sleep_time(sleep_time, Duration::from_secs(20)); @@ -294,11 +293,11 @@ mod tests { async fn test_cooperative_indexing_maximum_throughput() { tokio::time::pause(); let semaphore = Arc::new(Semaphore::new(1)); - let cooperative_indexing = - CooperativeIndexingCycle::new("id", Duration::from_secs(30), semaphore.clone()); + let indexing_cycle = + IndexingCycle::new("id", Duration::from_secs(30), Some(semaphore.clone())); let semaphore_guard = Semaphore::acquire_owned(semaphore).await; drop_after(semaphore_guard, Duration::from_secs(30)); - let cycle_guard = cooperative_indexing.cooperative_indexing_period().await; + let cycle_guard = indexing_cycle.indexing_period().await; tokio::time::advance(Duration::from_secs(15)).await; let (sleep_time, metrics) = cycle_guard.end_of_work(30_000_000); let expected_metrics = PipelineMetrics { @@ -313,11 +312,11 @@ mod tests { async fn test_cooperative_indexing_simple_contention() { tokio::time::pause(); let semaphore = Arc::new(Semaphore::new(1)); - let cooperative_indexing = - CooperativeIndexingCycle::new("id", Duration::from_secs(30), semaphore.clone()); + let indexing_cycle = + IndexingCycle::new("id", Duration::from_secs(30), Some(semaphore.clone())); let semaphore_guard = Semaphore::acquire_owned(semaphore).await; drop_after(semaphore_guard, Duration::from_secs(10)); - let cycle_guard = cooperative_indexing.cooperative_indexing_period().await; + let cycle_guard = indexing_cycle.indexing_period().await; tokio::time::advance(Duration::from_secs(10)).await; let (sleep_time, metrics) = cycle_guard.end_of_work(100_000_000); assert_approx_equal_sleep_time(sleep_time, Duration::from_secs(10)); @@ -329,27 +328,21 @@ mod tests { } #[tokio::test] - async fn test_cooperative_indexing_nudge_to_phase() { + async fn test_indexing_cycle_nudge_to_phase() { tokio::time::pause(); - let num_threads = 10; let num_pipelines = 100; let num_steps = 15; - let semaphore = Arc::new(Semaphore::new(num_threads)); let commit_timeout = Duration::from_secs(30); let t0 = Instant::now(); let mut handles = Vec::new(); for i in 0..num_pipelines { let target_phase = Duration::from_millis(commit_timeout.as_millis() as u64 * i / num_pipelines); - let cooperative_indexing = CooperativeIndexingCycle::new_with_phase( - target_phase, - commit_timeout, - semaphore.clone(), - ); + let indexing_cycle = IndexingCycle::new_with_phase(target_phase, commit_timeout, None); let join_handle = tokio::task::spawn(async move { let mut last_phase = 0; for _ in 0..num_steps { - let cycle_guard = cooperative_indexing.cooperative_indexing_period().await; + let cycle_guard = indexing_cycle.indexing_period().await; let work_time = Duration::from_millis(10); tokio::time::sleep(work_time).await; last_phase = diff --git a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs index b3d197bf000..9b06a738bb0 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs @@ -368,6 +368,7 @@ impl IndexingPipeline { self.params.indexing_directory.clone(), self.params.indexing_settings.clone(), self.params.cooperative_indexing_permits.clone(), + self.params.spread_indexing_pipelines, index_serializer_mailbox, self.params.fingerprinter_opt.clone(), ); @@ -553,6 +554,9 @@ pub struct IndexingPipelineParams { pub split_store: IndexingSplitStore, pub max_concurrent_split_uploads_index: usize, pub cooperative_indexing_permits: Option>, + /// Spreads the indexing pipelines of a node uniformly in time. Implied by cooperative + /// indexing. + pub spread_indexing_pipelines: bool, // Merge-related parameters pub merge_policy: Arc, @@ -697,6 +701,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox), event_broker: EventBroker::default(), params_fingerprint: 42u64, @@ -804,6 +809,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox), params_fingerprint: 42u64, event_broker: EventBroker::default(), @@ -934,6 +940,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox), event_broker: Default::default(), params_fingerprint: 42u64, @@ -1035,6 +1042,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox.clone()), event_broker: Default::default(), params_fingerprint: 42u64, @@ -1119,6 +1127,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: None, event_broker: Default::default(), params_fingerprint: 42u64, @@ -1273,6 +1282,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox), params_fingerprint: 42u64, event_broker: Default::default(), diff --git a/quickwit/quickwit-indexing/src/actors/indexing_service.rs b/quickwit/quickwit-indexing/src/actors/indexing_service.rs index 2bcb5131946..871253e06b5 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_service.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_service.rs @@ -123,6 +123,7 @@ pub struct IndexingService { #[cfg(feature = "metrics")] parquet_merge_pipeline_handles: HashMap, cooperative_indexing_permits: Option>, + spread_indexing_pipelines: bool, fingerprinter_opt: Option, merge_io_throughput_limiter_opt: Option, pub(crate) event_broker: EventBroker, @@ -166,6 +167,7 @@ impl IndexingService { } else { None }; + let spread_indexing_pipelines = indexer_config.enable_spread_indexing_pipelines; Ok(IndexingService { node_id, indexing_root_directory, @@ -189,6 +191,7 @@ impl IndexingService { fingerprinter_opt, merge_io_throughput_limiter_opt, cooperative_indexing_permits, + spread_indexing_pipelines, event_broker, }) } @@ -420,6 +423,7 @@ impl IndexingService { split_store, max_concurrent_split_uploads_index, cooperative_indexing_permits: self.cooperative_indexing_permits.clone(), + spread_indexing_pipelines: self.spread_indexing_pipelines, merge_policy, retention_policy, max_concurrent_split_uploads_merge, diff --git a/quickwit/quickwit-indexing/src/actors/mod.rs b/quickwit/quickwit-indexing/src/actors/mod.rs index fdab5df4bcc..4cc267438ad 100644 --- a/quickwit/quickwit-indexing/src/actors/mod.rs +++ b/quickwit/quickwit-indexing/src/actors/mod.rs @@ -12,10 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -mod cooperative_indexing; mod doc_processor; mod index_serializer; mod indexer; +mod indexing_cycle; mod indexing_pipeline; mod indexing_service; mod log_publisher_impl;