Faster shard scaling

quickwit/quickwit-common/src/shared_consts.rs

@@ -66,7 +66,14 @@ pub const INGESTER_PRIMARY_SHARDS_PREFIX: &str = "ingester.primary_shards:";
 /// File name for the encoded list of fields in the split
 pub const SPLIT_FIELDS_FILE_NAME: &str = "split_fields";
 
+/// More or less the indexing throughput of a core
+/// i.e. PIPELINE_THROUGHPUT / PIPELINE_FULL_CAPACITY
 pub const DEFAULT_SHARD_THROUGHPUT_LIMIT: ByteSize = ByteSize::mib(5);
+/// Large enough to absorb small bursts but should remain defensive against unbalanced shards.
+pub const DEFAULT_SHARD_BURST_LIMIT: ByteSize = ByteSize::mib(50);
+
+/// A compromise between "exponential" scale up and moderate shard count increase.
+pub const DEFAULT_SHARD_SCALE_UP_FACTOR: f32 = 1.5;
 
 // (Just a reexport).
 pub use bytesize::MIB;

quickwit/quickwit-config/src/cluster_config/mod.rs

@@ -24,6 +24,7 @@ pub struct ClusterConfig {
     pub default_index_root_uri: Uri,
     pub replication_factor: usize,
     pub shard_throughput_limit: ByteSize,
+    pub shard_scale_up_factor: f32,
 }
 
 impl ClusterConfig {
@@ -35,6 +36,7 @@ impl ClusterConfig {
             default_index_root_uri: Uri::for_test("ram:///indexes"),
             replication_factor: 1,
             shard_throughput_limit: quickwit_common::shared_consts::DEFAULT_SHARD_THROUGHPUT_LIMIT,
+            shard_scale_up_factor: 1.01,
         }
     }
 }

quickwit/quickwit-config/src/node_config/mod.rs

@@ -25,7 +25,9 @@ use anyhow::{bail, ensure};
 use bytesize::ByteSize;
 use http::HeaderMap;
 use quickwit_common::net::HostAddr;
-use quickwit_common::shared_consts::DEFAULT_SHARD_THROUGHPUT_LIMIT;
+use quickwit_common::shared_consts::{
+    DEFAULT_SHARD_BURST_LIMIT, DEFAULT_SHARD_SCALE_UP_FACTOR, DEFAULT_SHARD_THROUGHPUT_LIMIT,
+};
 use quickwit_common::uri::Uri;
 use quickwit_proto::indexing::CpuCapacity;
 use quickwit_proto::types::NodeId;
@@ -39,7 +41,7 @@ use crate::{ConfigFormat, MetastoreConfigs};
 
 pub const DEFAULT_QW_CONFIG_PATH: &str = "config/quickwit.yaml";
 
-#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct RestConfig {
     pub listen_addr: SocketAddr,
@@ -50,7 +52,7 @@ pub struct RestConfig {
     pub tls: Option<TlsConfig>,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct GrpcConfig {
     #[serde(default = "GrpcConfig::default_max_message_size")]
@@ -83,7 +85,7 @@ impl Default for GrpcConfig {
     }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct TlsConfig {
     pub cert_path: String,
@@ -193,7 +195,7 @@ impl Default for IndexerConfig {
     }
 }
 
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct SplitCacheLimits {
     pub max_num_bytes: ByteSize,
@@ -219,7 +221,7 @@ impl SplitCacheLimits {
     }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields, default)]
 pub struct SearcherConfig {
     pub aggregation_memory_limit: ByteSize,
@@ -254,7 +256,7 @@ pub struct SearcherConfig {
 /// This policy is inspired by this guidance. It does not track instanteneous throughput, but
 /// computes an overall timeout using the following formula:
 /// `timeout_offset + num_bytes_get_request / min_throughtput`
-#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct StorageTimeoutPolicy {
     pub min_throughtput_bytes_per_secs: u64,
     pub timeout_millis: u64,
@@ -338,14 +340,25 @@ impl SearcherConfig {
     }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields, default)]
 pub struct IngestApiConfig {
+    /// Maximum memory space taken by the ingest WAL
     pub max_queue_memory_usage: ByteSize,
+    /// Maximum disk space taken by the ingest WAL
     pub max_queue_disk_usage: ByteSize,
     replication_factor: usize,
     pub content_length_limit: ByteSize,
+    /// (hidden) Targeted throughput for each shard
     pub shard_throughput_limit: ByteSize,
+    /// (hidden) Maximum accumulated throughput capacity for underutilized
+    /// shards, allowing the throughput limit to be temporarily exceeded
+    pub shard_burst_limit: ByteSize,
+    /// (hidden) new_shard_count = ceil(old_shard_count * shard_scale_up_factor)
+    ///
+    /// Setting this too high will be cancelled out by the arbiter that prevents
+    /// creating too many shards at once.
+    pub shard_scale_up_factor: f32,
 }
 
 impl Default for IngestApiConfig {
@@ -356,6 +369,8 @@ impl Default for IngestApiConfig {
             replication_factor: 1,
             content_length_limit: ByteSize::mib(10),
             shard_throughput_limit: DEFAULT_SHARD_THROUGHPUT_LIMIT,
+            shard_burst_limit: DEFAULT_SHARD_BURST_LIMIT,
+            shard_scale_up_factor: DEFAULT_SHARD_SCALE_UP_FACTOR,
         }
     }
 }
@@ -398,20 +413,34 @@ impl IngestApiConfig {
             self.max_queue_memory_usage
         );
         info!(
-            "ingestion shard throughput limit: {:?}",
+            "ingestion shard throughput limit: {}",
             self.shard_throughput_limit
         );
         ensure!(
             self.shard_throughput_limit >= ByteSize::mib(1)
                 && self.shard_throughput_limit <= ByteSize::mib(20),
-            "shard_throughput_limit ({:?}) must be within 1mb and 20mb",
+            "shard_throughput_limit ({}) must be within 1mb and 20mb",
             self.shard_throughput_limit
         );
+        // The newline delimited format is persisted as something a bit larger
+        // (lines prefixed with their length)
+        let estimated_persist_size = ByteSize::b(3 * self.content_length_limit.as_u64() / 2);
+        ensure!(
+            self.shard_burst_limit >= estimated_persist_size,
+            "shard_burst_limit ({}) must be at least 1.5*content_length_limit ({})",
+            self.shard_burst_limit,
+            estimated_persist_size,
+        );
+        ensure!(
+            self.shard_scale_up_factor > 1.0,
+            "shard_scale_up_factor ({}) must be greater than 1",
+            self.shard_scale_up_factor,
+        );
         Ok(())
     }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct JaegerConfig {
     /// Enables the gRPC endpoint that allows the Jaeger Query Service to connect and retrieve

quickwit/quickwit-control-plane/src/control_plane.rs

@@ -167,6 +167,7 @@ impl ControlPlane {
                     ingester_pool.clone(),
                     replication_factor,
                     shard_throughput_limit_mib,
+                    cluster_config.shard_scale_up_factor,
                 );
 
                 let readiness_tx = readiness_tx.clone();