feat(log-ingestor): Make database connection pool size configurable (fixes #2039).

components/clp-py-utils/clp_py_utils/clp_config.py

@@ -97,6 +97,7 @@
 # Specific types
 # TODO: Replace this with pydantic_extra_types.domain.DomainStr.
 DomainStr = NonEmptyStr
+DatabaseConnectionPoolSize = Annotated[int, Field(gt=0, lt=2**32)]
 Port = Annotated[int, Field(gt=0, lt=2**16)]
 SerializablePath = Annotated[pathlib.Path, PlainSerializer(serialize_path)]
 ZstdCompressionLevel = Annotated[int, Field(ge=1, le=19)]
@@ -775,6 +776,7 @@ class ApiServer(BaseModel):
 class LogIngestor(BaseModel):
     host: DomainStr = "localhost"
     port: Port = 3002
+    database_connection_pool_size: DatabaseConnectionPoolSize = 100
     logging_level: LoggingLevelRust = "INFO"
 
 

components/clp-rust-utils/src/clp_config/package/config.rs

@@ -1,3 +1,5 @@
+use std::num::NonZeroU32;
+
 use serde::Deserialize;
 
 use crate::clp_config::{AwsAuthentication, S3Config};
@@ -227,6 +229,9 @@ impl Default for StreamOutputStorage {
     }
 }
 
+const DEFAULT_LOG_INGESTOR_DATABASE_CONNECTION_POOL_SIZE: NonZeroU32 =
+    NonZeroU32::new(100).unwrap();
+
 /// Mirror of `clp_py_utils.clp_config.LogIngestor`.
 ///
 /// # NOTE
@@ -237,6 +242,7 @@ impl Default for StreamOutputStorage {
 pub struct LogIngestor {
     pub host: String,
     pub port: u16,
+    pub database_connection_pool_size: NonZeroU32,
     pub logging_level: String,
 }
 
@@ -245,6 +251,7 @@ impl Default for LogIngestor {
         Self {
             host: "localhost".to_owned(),
             port: 3002,
+            database_connection_pool_size: DEFAULT_LOG_INGESTOR_DATABASE_CONNECTION_POOL_SIZE,
             logging_level: "INFO".to_owned(),
         }
     }
@@ -339,7 +346,25 @@ impl Default for Telemetry {
 
 #[cfg(test)]
 mod tests {
-    use super::LogsInput;
+    use super::{LogIngestor, LogsInput};
+
+    #[test]
+    fn deserialize_log_ingestor_database_connection_pool_size() {
+        let default_config = serde_json::from_str::<LogIngestor>("{}")
+            .expect("failed to deserialize default `LogIngestor` config");
+        assert_eq!(100, default_config.database_connection_pool_size.get());
+
+        let custom_config =
+            serde_json::from_str::<LogIngestor>(r#"{"database_connection_pool_size": 42}"#)
+                .expect("failed to deserialize custom `LogIngestor` config");
+        assert_eq!(42, custom_config.database_connection_pool_size.get());
+    }
+
+    #[test]
+    fn reject_zero_log_ingestor_database_connection_pool_size() {
+        let result = serde_json::from_str::<LogIngestor>(r#"{"database_connection_pool_size": 0}"#);
+        assert!(result.is_err());
+    }
 
     #[test]
     fn deserialize_logs_input_s3_config() {

components/log-ingestor/src/ingestion_job_manager.rs

@@ -73,10 +73,6 @@ impl IngestionJobManagerState {
     ///
     /// * [`anyhow::Error`] if the logs input type in the CLP configuration is unsupported.
     /// * Forwards [`ClpDbIngestionConnector::connect`]'s return values on failure.
-    ///
-    /// # Panics
-    ///
-    /// Panics if `clp_config.log_ingestor` is `None`.
     pub async fn from_config(
         clp_config: ClpConfig,
         clp_credentials: ClpCredentials,

components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs

@@ -1,5 +1,6 @@
 use std::time::Duration;
 
+use anyhow::Context;
 use async_trait::async_trait;
 use clp_rust_utils::{
     clp_config::{
@@ -178,6 +179,7 @@ impl ClpDbIngestionConnector {
     ///
     /// * Forwards [`clp_rust_utils::database::mysql::create_clp_db_mysql_pool`]'s return values on
     ///   failure.
+    /// * [`anyhow::Error`] if the log-ingestor configuration is missing.
     /// * Forwards [`Self::create_tables`]'s return values on failure.
     /// * Forwards [`Self::get_unfinished_compression_jobs`]'s return values on failure.
     /// * Forwards [`Self::load_ingestion_jobs`]'s return values on failure.
@@ -199,10 +201,16 @@ impl ClpDbIngestionConnector {
             }
         };
 
+        let database_connection_pool_size = clp_config
+            .log_ingestor
+            .as_ref()
+            .context("Invalid CLP config: log-ingestor is not configured")?
+            .database_connection_pool_size
+            .get();
         let mysql_pool = clp_rust_utils::database::mysql::create_clp_db_mysql_pool(
             &clp_config.database,
             &clp_credentials.database,
-            100,
+            database_connection_pool_size,
         )
         .await?;
 

components/package-template/src/etc/clp-config.template.json.yaml

@@ -22,6 +22,7 @@ telemetry:
 #log_ingestor:
 #  host: "localhost"
 #  port: 3002
+#  database_connection_pool_size: 100
 #  logging_level: "INFO"
 
 ## Location (e.g., directory) containing any logs you wish to compress. Must be reachable by all