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/Cargo.toml

@@ -27,5 +27,6 @@ tracing-subscriber = { version = "0.3.22", features = ["json", "env-filter", "fm
 utoipa = { version = "5.4.0" }
 
 [dev-dependencies]
+anyhow = "1.0.100"
 hex = "0.4.3"
 serde_json = "1.0.149"

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};
@@ -237,6 +239,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 +248,8 @@ impl Default for LogIngestor {
         Self {
             host: "localhost".to_owned(),
             port: 3002,
+            database_connection_pool_size: NonZeroU32::new(100)
+                .expect("default database connection pool size must be nonzero"),
             logging_level: "INFO".to_owned(),
         }
     }
@@ -339,10 +344,31 @@ impl Default for Telemetry {
 
 #[cfg(test)]
 mod tests {
-    use super::LogsInput;
+    use super::{LogIngestor, LogsInput};
+
+    #[test]
+    fn deserialize_log_ingestor_database_connection_pool_size() -> anyhow::Result<()> {
+        let default_config = serde_json::from_str::<LogIngestor>("{}")?;
+        assert_eq!(100, default_config.database_connection_pool_size.get());
+
+        let custom_config =
+            serde_json::from_str::<LogIngestor>(r#"{"database_connection_pool_size": 42}"#)?;
+        assert_eq!(42, custom_config.database_connection_pool_size.get());
+        Ok(())
+    }
+
+    #[test]
+    fn reject_zero_log_ingestor_database_connection_pool_size() -> anyhow::Result<()> {
+        let result = serde_json::from_str::<LogIngestor>(r#"{"database_connection_pool_size": 0}"#);
+        anyhow::ensure!(
+            result.is_err(),
+            "zero database connection pool size was accepted"
+        );
+        Ok(())
+    }
 
     #[test]
-    fn deserialize_logs_input_s3_config() {
+    fn deserialize_logs_input_s3_config() -> anyhow::Result<()> {
         const ACCESS_KEY_ID: &str = "YSCOPE";
         const SECRET_ACCESS_KEY: &str = "IamSecret";
         let logs_input_config_json = serde_json::json!({
@@ -357,8 +383,7 @@ mod tests {
         });
 
         let deserialized =
-            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())
-                .expect("failed to deserialize `LogsInput` from JSON");
+            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())?;
 
         match deserialized {
             LogsInput::S3 { config } => match config.aws_authentication {
@@ -367,15 +392,16 @@ mod tests {
                     assert_eq!(credentials.secret_access_key, SECRET_ACCESS_KEY);
                 }
                 crate::clp_config::AwsAuthentication::Default => {
-                    panic!("Expected credentials, got `default`")
+                    panic!("expected credentials, got `default`")
                 }
             },
-            LogsInput::Fs { .. } => panic!("Expected S3"),
+            LogsInput::Fs { .. } => panic!("expected S3"),
         }
+        Ok(())
     }
 
     #[test]
-    fn deserialize_logs_input_s3_default_config() {
+    fn deserialize_logs_input_s3_default_config() -> anyhow::Result<()> {
         let logs_input_config_json = serde_json::json!({
             "type": "s3",
             "aws_authentication": {
@@ -384,8 +410,7 @@ mod tests {
         });
 
         let deserialized =
-            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())
-                .expect("failed to deserialize `LogsInput` from JSON");
+            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())?;
 
         match deserialized {
             LogsInput::S3 { config } => {
@@ -394,12 +419,13 @@ mod tests {
                     crate::clp_config::AwsAuthentication::Default
                 );
             }
-            LogsInput::Fs { .. } => panic!("Expected S3"),
+            LogsInput::Fs { .. } => panic!("expected S3"),
         }
+        Ok(())
     }
 
     #[test]
-    fn deserialize_logs_input_fs_config() {
+    fn deserialize_logs_input_fs_config() -> anyhow::Result<()> {
         const DIRECTORY: &str = "/var/logs";
 
         let logs_input_config_json = serde_json::json!({
@@ -408,14 +434,14 @@ mod tests {
         });
 
         let deserialized =
-            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())
-                .expect("failed to deserialize `LogsInput` from JSON");
+            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())?;
 
         match deserialized {
             LogsInput::Fs { config } => {
                 assert_eq!(config.directory, DIRECTORY);
             }
-            LogsInput::S3 { .. } => panic!("Expected Fs"),
+            LogsInput::S3 { .. } => panic!("expected Fs"),
         }
+        Ok(())
     }
 }

components/log-ingestor/src/ingestion_job_manager.rs

@@ -25,22 +25,22 @@ use crate::{
 /// Errors for ingestion job manager operations.
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
-    #[error("Log ingestor internal error: {0}")]
+    #[error("log ingestor internal error: {0}")]
     InternalError(#[from] anyhow::Error),
 
-    #[error("Ingestion job not found: {0}")]
+    #[error("ingestion job not found: {0}")]
     JobNotFound(IngestionJobId),
 
-    #[error("Prefix conflict: {0}")]
+    #[error("prefix conflict: {0}")]
     PrefixConflict(String),
 
-    #[error("Custom endpoint URL not supported: {0}")]
+    #[error("custom endpoint URL not supported: {0}")]
     CustomEndpointUrlNotSupported(String),
 
-    #[error("Invalid job config: {0}")]
+    #[error("invalid job config: {0}")]
     InvalidConfig(#[from] ConfigError),
 
-    #[error("A region code must be specified when using the default AWS endpoint")]
+    #[error("a region code must be specified when using the default AWS endpoint")]
     MissingRegionCode,
 }
 
@@ -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::{
@@ -176,6 +177,7 @@ impl ClpDbIngestionConnector {
     ///
     /// Returns an error if:
     ///
+    /// * [`anyhow::Error`] if the log-ingestor configuration is missing.
     /// * Forwards [`clp_rust_utils::database::mysql::create_clp_db_mysql_pool`]'s return values on
     ///   failure.
     /// * Forwards [`Self::create_tables`]'s return values on failure.
@@ -193,16 +195,22 @@ impl ClpDbIngestionConnector {
             LogsInput::S3 { config } => config.aws_authentication,
             LogsInput::Fs { .. } => {
                 panic!(
-                    "Invalid CLP config: Unsupported logs input type. The current implementation \
-                     only supports S3 input."
+                    "invalid CLP config: unsupported logs input type; the current implementation \
+                     only supports S3 input"
                 );
             }
         };
 
+        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?;
 
@@ -363,7 +371,7 @@ impl ClpDbIngestionConnector {
                     },
                     compression_job_id,
                     num_object_metadata_submitted: usize::try_from(num_submitted)
-                        .expect("Number of files submitted is not `usize` compatible"),
+                        .expect("number of files submitted is not `usize` compatible"),
                 },
             )
             .collect();

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

tools/deployment/package-helm/Chart.yaml

@@ -1,6 +1,6 @@
 apiVersion: "v2"
 name: "clp"
-version: "0.4.1-dev.2"
+version: "0.4.1-dev.3"
 description: "A Helm chart for CLP's (Compressed Log Processor) package deployment"
 type: "application"
 appVersion: "0.13.1-dev"

tools/deployment/package-helm/templates/configmap.yaml

@@ -121,6 +121,7 @@ data:
     {{- end }}{{/* with .Values.clpConfig.logs_input */}}
     {{- with .Values.clpConfig.log_ingestor }}
     log_ingestor:
+      database_connection_pool_size: {{ .database_connection_pool_size | int }}
       host: "localhost"
       logging_level: {{ .logging_level | quote }}
       port: 3002

tools/deployment/package-helm/values.yaml

@@ -284,6 +284,7 @@ clpConfig:
 
   # log-ingestor config. Currently, the config is applicable only if `logs_input.type` is "s3".
   log_ingestor:
+    database_connection_pool_size: 100
     port: 30302
     logging_level: "INFO"