Add shard scaling factor

Author: rdettai

quickwit/quickwit-common/src/shared_consts.rs

@@ -69,8 +69,14 @@ 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);
 
+/// Maximum factor that avoids oscillating between scale up and scale down
+pub const MAX_SHARD_SCALE_UP_FACTOR: f32 = 2.0;
+/// A high value to allow quick scale up by default
+pub const DEFAULT_SHARD_SCALE_UP_FACTOR: f32 = 1.5;
+const _: () = assert!(DEFAULT_SHARD_SCALE_UP_FACTOR < MAX_SHARD_SCALE_UP_FACTOR);
+
 // (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_scaling_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_scaling_factor: 1.01,
         }
     }
 }

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

@@ -25,7 +25,10 @@ use anyhow::{bail, ensure};
 use bytesize::ByteSize;
 use http::HeaderMap;
 use quickwit_common::net::HostAddr;
-use quickwit_common::shared_consts::{DEFAULT_SHARD_BURST_LIMIT, DEFAULT_SHARD_THROUGHPUT_LIMIT};
+use quickwit_common::shared_consts::{
+    DEFAULT_SHARD_BURST_LIMIT, DEFAULT_SHARD_SCALE_UP_FACTOR, DEFAULT_SHARD_THROUGHPUT_LIMIT,
+    MAX_SHARD_SCALE_UP_FACTOR,
+};
 use quickwit_common::uri::Uri;
 use quickwit_proto::indexing::CpuCapacity;
 use quickwit_proto::types::NodeId;
@@ -39,7 +42,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 +53,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 +86,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 +196,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 +222,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 +257,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,7 +341,7 @@ 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
@@ -347,10 +350,12 @@ pub struct IngestApiConfig {
     pub max_queue_disk_usage: ByteSize,
     replication_factor: usize,
     pub content_length_limit: ByteSize,
-    /// [hidden] Targeted throughput for each shard
+    /// (hidden) Targeted throughput for each shard
     pub shard_throughput_limit: ByteSize,
-    /// [hidden] Targeted throughput for each shard
+    /// (hidden) Targeted throughput for each shard
     pub shard_burst_limit: ByteSize,
+    /// (hidden) new_shard_count = ceil(old_shard_count * shard_scaling_factor)
+    pub shard_scale_up_factor: f32,
 }
 
 impl Default for IngestApiConfig {
@@ -362,6 +367,7 @@ impl Default for IngestApiConfig {
             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,
         }
     }
 }
@@ -422,11 +428,18 @@ impl IngestApiConfig {
             self.shard_burst_limit,
             estimated_persist_size,
         );
+        ensure!(
+            self.shard_scale_up_factor > 1.0
+                && self.shard_scale_up_factor <= MAX_SHARD_SCALE_UP_FACTOR,
+            "shard_scale_up_factor ({:?}) must be in the (1,{}) interval",
+            self.shard_scale_up_factor,
+            MAX_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_scaling_factor,
                 );
 
                 let readiness_tx = readiness_tx.clone();