diff --git a/Cargo.lock b/Cargo.lock index 640fcda85..ec1a8e21e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -942,6 +942,7 @@ checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" name = "clp-rust-utils" version = "0.13.1-dev" dependencies = [ + "anyhow", "aws-config", "aws-sdk-s3", "aws-sdk-sqs", diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 46ea77940..1b4f29b85 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/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" diff --git a/components/clp-rust-utils/Cargo.toml b/components/clp-rust-utils/Cargo.toml index fb7445c1e..f5387e1b2 100644 --- a/components/clp-rust-utils/Cargo.toml +++ b/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" diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs index 29863686b..39aec1e38 100644 --- a/components/clp-rust-utils/src/clp_config/package/config.rs +++ b/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::("{}")?; + assert_eq!(100, default_config.database_connection_pool_size.get()); + + let custom_config = + serde_json::from_str::(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::(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::(logs_input_config_json.to_string().as_str()) - .expect("failed to deserialize `LogsInput` from JSON"); + serde_json::from_str::(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::(logs_input_config_json.to_string().as_str()) - .expect("failed to deserialize `LogsInput` from JSON"); + serde_json::from_str::(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::(logs_input_config_json.to_string().as_str()) - .expect("failed to deserialize `LogsInput` from JSON"); + serde_json::from_str::(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(()) } } diff --git a/components/log-ingestor/src/ingestion_job_manager.rs b/components/log-ingestor/src/ingestion_job_manager.rs index 00639daf2..a3fe0ec52 100644 --- a/components/log-ingestor/src/ingestion_job_manager.rs +++ b/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, diff --git a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs index 7b65b5ae7..60bd7ad2e 100644 --- a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs +++ b/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(); diff --git a/components/package-template/src/etc/clp-config.template.json.yaml b/components/package-template/src/etc/clp-config.template.json.yaml index 63c8af433..64abed24f 100644 --- a/components/package-template/src/etc/clp-config.template.json.yaml +++ b/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 diff --git a/tools/deployment/package-helm/Chart.yaml b/tools/deployment/package-helm/Chart.yaml index 12f1fcac6..5e6fef4af 100644 --- a/tools/deployment/package-helm/Chart.yaml +++ b/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" diff --git a/tools/deployment/package-helm/templates/configmap.yaml b/tools/deployment/package-helm/templates/configmap.yaml index ee49584a0..1ea4bc4ee 100644 --- a/tools/deployment/package-helm/templates/configmap.yaml +++ b/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 diff --git a/tools/deployment/package-helm/values.yaml b/tools/deployment/package-helm/values.yaml index a52dd2172..7e86bcc58 100644 --- a/tools/deployment/package-helm/values.yaml +++ b/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"