Increase burst limit

Author: rdettai

quickwit/quickwit-common/src/shared_consts.rs

@@ -66,7 +66,11 @@ 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);
+///
+pub const DEFAULT_SHARD_BURST_LIMIT: ByteSize = ByteSize::mib(50);
 
 // (Just a reexport).
 pub use bytesize::MIB;

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

@@ -25,7 +25,7 @@ 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_THROUGHPUT_LIMIT};
 use quickwit_common::uri::Uri;
 use quickwit_proto::indexing::CpuCapacity;
 use quickwit_proto::types::NodeId;
@@ -323,11 +323,16 @@ impl SearcherConfig {
 #[derive(Clone, Debug, Eq, 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] Targeted throughput for each shard
+    pub shard_burst_limit: ByteSize,
 }
 
 impl Default for IngestApiConfig {
@@ -338,6 +343,7 @@ 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,
         }
     }
 }
@@ -389,6 +395,15 @@ impl IngestApiConfig {
             "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,
+        );
         Ok(())
     }
 }

quickwit/quickwit-serve/src/lib.rs

@@ -851,7 +851,6 @@ async fn setup_ingest_v2(
 ) -> anyhow::Result<(IngestRouter, IngestRouterServiceClient, Option<Ingester>)> {
     // Instantiate ingest router.
     let self_node_id: NodeId = cluster.self_node_id().into();
-    let content_length_limit = node_config.ingest_api_config.content_length_limit;
     let replication_factor = node_config
         .ingest_api_config
         .replication_factor()
@@ -873,15 +872,10 @@ async fn setup_ingest_v2(
         .stack_layer(INGEST_GRPC_SERVER_METRICS_LAYER.clone())
         .build(ingest_router.clone());
 
-    // We compute the burst limit as something a bit larger than the content length limit, because
-    // we actually rewrite the `\n-delimited format into a tiny bit larger buffer, where the
-    // line length is prefixed.
-    let burst_limit = (content_length_limit.as_u64() * 3 / 2).clamp(10_000_000, 200_000_000);
-
     let rate_limit =
         ConstantRate::bytes_per_sec(node_config.ingest_api_config.shard_throughput_limit);
     let rate_limiter_settings = RateLimiterSettings {
-        burst_limit,
+        burst_limit: node_config.ingest_api_config.shard_burst_limit.as_u64(),
         rate_limit,
         // Refill every 100ms.
         refill_period: Duration::from_millis(100),