Explicit scale up field names and comments

Author: rdettai

quickwit/quickwit-common/src/shared_consts.rs

@@ -67,9 +67,9 @@ pub const INGESTER_PRIMARY_SHARDS_PREFIX: &str = "ingester.primary_shards:";
 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
+/// 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
+/// 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.

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

@@ -24,7 +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,
+    pub shard_scale_up_factor: f32,
 }
 
 impl ClusterConfig {
@@ -36,7 +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,
+            shard_scale_up_factor: 1.01,
         }
     }
 }

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

@@ -351,9 +351,10 @@ pub struct IngestApiConfig {
     pub content_length_limit: ByteSize,
     /// (hidden) Targeted throughput for each shard
     pub shard_throughput_limit: ByteSize,
-    /// (hidden) Targeted throughput for each shard
+    /// (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_scaling_factor)
+    /// (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.
@@ -412,27 +413,27 @@ 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 ({:?})",
+            "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",
+            "shard_scale_up_factor ({}) must be greater than 1",
             self.shard_scale_up_factor,
         );
         Ok(())

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

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

quickwit/quickwit-control-plane/src/ingest/ingest_controller.rs

@@ -285,7 +285,7 @@ impl IngestController {
         ingester_pool: IngesterPool,
         replication_factor: usize,
         max_shard_ingestion_throughput_mib_per_sec: f32,
-        shard_scaling_factor: f32,
+        shard_scale_up_factor: f32,
     ) -> Self {
         IngestController {
             metastore,
@@ -295,7 +295,7 @@ impl IngestController {
             stats: IngestControllerStats::default(),
             scaling_arbiter: ScalingArbiter::with_max_shard_ingestion_throughput_mib_per_sec(
                 max_shard_ingestion_throughput_mib_per_sec,
-                shard_scaling_factor,
+                shard_scale_up_factor,
             ),
         }
     }
@@ -673,17 +673,17 @@ impl IngestController {
         shard_stats: ShardStats,
         model: &mut ControlPlaneModel,
         progress: &Progress,
-        shards_to_create: usize,
+        num_shards_to_open: usize,
     ) -> MetastoreResult<()> {
         if !model
-            .acquire_scaling_permits(&source_uid, ScalingMode::Up(shards_to_create))
+            .acquire_scaling_permits(&source_uid, ScalingMode::Up(num_shards_to_open))
             .unwrap_or(false)
         {
             return Ok(());
         }
-        let new_num_open_shards = shard_stats.num_open_shards + shards_to_create;
+        let new_num_open_shards = shard_stats.num_open_shards + num_shards_to_open;
         let new_shards_per_source: HashMap<SourceUid, usize> =
-            HashMap::from_iter([(source_uid.clone(), shards_to_create)]);
+            HashMap::from_iter([(source_uid.clone(), num_shards_to_open)]);
         let successful_source_uids_res = self
             .try_open_shards(new_shards_per_source, model, &Default::default(), progress)
             .await;
@@ -695,7 +695,7 @@ impl IngestController {
                 if successful_source_uids.is_empty() {
                     // We did not manage to create the shard.
                     // We can release our permit.
-                    model.release_scaling_permits(&source_uid, ScalingMode::Up(shards_to_create));
+                    model.release_scaling_permits(&source_uid, ScalingMode::Up(num_shards_to_open));
                     warn!(
                         index_uid=%source_uid.index_uid,
                         source_id=%source_uid.source_id,
@@ -719,7 +719,7 @@ impl IngestController {
                     source_id=%source_uid.source_id,
                     "scaling up number of shards to {new_num_open_shards} failed: {metastore_error:?}"
                 );
-                model.release_scaling_permits(&source_uid, ScalingMode::Up(shards_to_create));
+                model.release_scaling_permits(&source_uid, ScalingMode::Up(num_shards_to_open));
                 Err(metastore_error)
             }
         }

quickwit/quickwit-control-plane/src/ingest/scaling_arbiter.rs

@@ -30,22 +30,22 @@ pub(crate) struct ScalingArbiter {
     // after scaling up, and double check that it is above the long term threshold.
     scale_up_shards_short_term_threshold_mib_per_sec: f32,
     scale_up_shards_long_term_threshold_mib_per_sec: f32,
-    // The max increase factor of the number of shards in one scaling operation
-    shard_scaling_factor: f32,
+    // The max increase factor of the number of shards in one scale up operation
+    shard_scale_up_factor: f32,
 }
 
 impl ScalingArbiter {
     pub fn with_max_shard_ingestion_throughput_mib_per_sec(
         max_shard_throughput_mib_per_sec: f32,
-        shard_scaling_factor: f32,
+        shard_scale_up_factor: f32,
     ) -> ScalingArbiter {
         ScalingArbiter {
             scale_up_shards_short_term_threshold_mib_per_sec: max_shard_throughput_mib_per_sec
                 * 0.8f32,
             scale_up_shards_long_term_threshold_mib_per_sec: max_shard_throughput_mib_per_sec
                 * 0.3f32,
             scale_down_shards_threshold_mib_per_sec: max_shard_throughput_mib_per_sec * 0.2f32,
-            shard_scaling_factor,
+            shard_scale_up_factor,
         }
     }
 
@@ -65,7 +65,7 @@ impl ScalingArbiter {
 
             // compute the next number of shards we should have according the scaling factor
             let target_number_shards =
-                (shard_stats.num_open_shards as f32 * self.shard_scaling_factor).ceil() as usize;
+                (shard_stats.num_open_shards as f32 * self.shard_scale_up_factor).ceil() as usize;
 
             let new_number_shards = max_number_shards.min(target_number_shards);
 

quickwit/quickwit-control-plane/src/model/shard_table.rs

@@ -25,7 +25,7 @@ use quickwit_proto::ingest::{Shard, ShardState};
 use quickwit_proto::types::{IndexUid, NodeId, ShardId, SourceId, SourceUid};
 use tracing::{error, info, warn};
 
-/// Limits the number of shards that can be opened for scaling up a source to 5 per minute.
+/// Limits the number of scale up operations that can happen to a source to 5 per minute.
 const SCALING_UP_RATE_LIMITER_SETTINGS: RateLimiterSettings = RateLimiterSettings {
     burst_limit: 5,
     rate_limit: ConstantRate::new(5, Duration::from_secs(60)),

quickwit/quickwit-serve/src/lib.rs

@@ -1058,7 +1058,7 @@ async fn setup_control_plane(
         default_index_root_uri,
         replication_factor,
         shard_throughput_limit: ingest_api_config.shard_throughput_limit,
-        shard_scaling_factor: ingest_api_config.shard_scale_up_factor,
+        shard_scale_up_factor: ingest_api_config.shard_scale_up_factor,
     };
     let (control_plane_mailbox, _control_plane_handle, mut readiness_rx) = ControlPlane::spawn(
         universe,