feat: allow enabling debug logging in replicator (#433)

Author: imor

etl-api/src/configs/log.rs

@@ -0,0 +1,42 @@
+use std::fmt;
+
+use serde::{Deserialize, Serialize};
+use utoipa::ToSchema;
+
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema, Default)]
+pub enum LogLevel {
+    /// The "trace" level.
+    ///
+    /// Designates very low priority, often extremely verbose, information.
+    Trace,
+    /// The "debug" level.
+    ///
+    /// Designates lower priority information.
+    Debug,
+    /// The "info" level.
+    ///
+    /// Designates useful information.
+    #[default]
+    Info,
+    /// The "warn" level.
+    ///
+    /// Designates hazardous situations.
+    Warn,
+    /// The "error" level.
+    ///
+    /// Designates very serious errors.
+    Error,
+}
+
+impl fmt::Display for LogLevel {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let s = match self {
+            LogLevel::Trace => "trace",
+            LogLevel::Debug => "debug",
+            LogLevel::Info => "info",
+            LogLevel::Warn => "warn",
+            LogLevel::Error => "error",
+        };
+        write!(f, "{s}")
+    }
+}

etl-api/src/configs/mod.rs

@@ -1,5 +1,6 @@
 pub mod destination;
 pub mod encryption;
+pub mod log;
 pub mod pipeline;
 pub mod serde;
 pub mod source;

etl-api/src/configs/pipeline.rs

@@ -2,7 +2,7 @@ use etl_config::shared::{BatchConfig, PgConnectionConfig, PipelineConfig};
 use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 
-use crate::configs::store::Store;
+use crate::configs::{log::LogLevel, store::Store};
 
 const DEFAULT_BATCH_MAX_SIZE: usize = 100000;
 const DEFAULT_BATCH_MAX_FILL_MS: u64 = 10000;
@@ -41,6 +41,7 @@ pub struct FullApiPipelineConfig {
     #[schema(example = 4)]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub max_table_sync_workers: Option<u16>,
+    pub log_level: Option<LogLevel>,
 }
 
 impl From<StoredPipelineConfig> for FullApiPipelineConfig {
@@ -54,6 +55,7 @@ impl From<StoredPipelineConfig> for FullApiPipelineConfig {
             table_error_retry_delay_ms: Some(value.table_error_retry_delay_ms),
             table_error_retry_max_attempts: Some(value.table_error_retry_max_attempts),
             max_table_sync_workers: Some(value.max_table_sync_workers),
+            log_level: value.log_level,
         }
     }
 }
@@ -75,6 +77,8 @@ pub struct PartialApiPipelineConfig {
     #[schema(example = 4)]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub max_table_sync_workers: Option<u16>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub log_level: Option<LogLevel>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -85,6 +89,7 @@ pub struct StoredPipelineConfig {
     #[serde(default = "default_table_error_retry_max_attempts")]
     pub table_error_retry_max_attempts: u32,
     pub max_table_sync_workers: u16,
+    pub log_level: Option<LogLevel>,
 }
 
 impl StoredPipelineConfig {
@@ -129,6 +134,8 @@ impl StoredPipelineConfig {
         if let Some(value) = partial.max_table_sync_workers {
             self.max_table_sync_workers = value;
         }
+
+        self.log_level = partial.log_level
     }
 }
 
@@ -159,6 +166,7 @@ impl From<FullApiPipelineConfig> for StoredPipelineConfig {
             max_table_sync_workers: value
                 .max_table_sync_workers
                 .unwrap_or(DEFAULT_MAX_TABLE_SYNC_WORKERS),
+            log_level: value.log_level,
         }
     }
 }
@@ -179,6 +187,7 @@ mod tests {
             table_error_retry_delay_ms: 2000,
             table_error_retry_max_attempts: 7,
             max_table_sync_workers: 4,
+            log_level: None,
         };
 
         let json = serde_json::to_string(&config).unwrap();
@@ -208,6 +217,7 @@ mod tests {
             table_error_retry_delay_ms: None,
             table_error_retry_max_attempts: None,
             max_table_sync_workers: None,
+            log_level: Some(LogLevel::Debug),
         };
 
         let stored: StoredPipelineConfig = full_config.clone().into();
@@ -224,6 +234,7 @@ mod tests {
             table_error_retry_delay_ms: None,
             table_error_retry_max_attempts: None,
             max_table_sync_workers: None,
+            log_level: None,
         };
 
         let stored: StoredPipelineConfig = full_config.into();
@@ -255,6 +266,7 @@ mod tests {
             table_error_retry_delay_ms: 1000,
             table_error_retry_max_attempts: 3,
             max_table_sync_workers: 2,
+            log_level: None,
         };
 
         let partial = PartialApiPipelineConfig {
@@ -266,6 +278,7 @@ mod tests {
             table_error_retry_delay_ms: Some(5000),
             table_error_retry_max_attempts: Some(9),
             max_table_sync_workers: None,
+            log_level: None,
         };
 
         stored.merge(partial);

etl-api/src/k8s/base.rs

@@ -3,7 +3,7 @@ use etl_config::Environment;
 use k8s_openapi::api::core::v1::ConfigMap;
 use thiserror::Error;
 
-use crate::configs::destination::StoredDestinationConfig;
+use crate::configs::{destination::StoredDestinationConfig, log::LogLevel};
 
 /// Errors emitted by the Kubernetes integration.
 ///
@@ -143,6 +143,7 @@ pub trait K8sClient: Send + Sync {
         replicator_image: &str,
         environment: Environment,
         destination_type: DestinationType,
+        log_level: LogLevel,
     ) -> Result<(), K8sError>;
 
     /// Deletes the replicator [`StatefulSet`] if it exists.

etl-api/src/k8s/core.rs

@@ -3,6 +3,7 @@ use etl_config::shared::{ReplicatorConfigWithoutSecrets, SupabaseConfigWithoutSe
 use secrecy::ExposeSecret;
 
 use crate::configs::destination::{StoredDestinationConfig, StoredIcebergConfig};
+use crate::configs::log::LogLevel;
 use crate::configs::pipeline::StoredPipelineConfig;
 use crate::configs::source::StoredSourceConfig;
 use crate::db::destinations::Destination;
@@ -72,6 +73,8 @@ pub async fn create_or_update_pipeline_resources_in_k8s(
         api_url: supabase_api_url.map(|url| url.to_owned()),
     };
 
+    let log_level = pipeline.config.log_level.clone().unwrap_or_default();
+
     let replicator_config = build_replicator_config_without_secrets(
         k8s_client,
         // We are safe to perform this conversion, since the i64 -> u64 conversion performs wrap
@@ -93,6 +96,7 @@ pub async fn create_or_update_pipeline_resources_in_k8s(
         image.name,
         environment,
         destination_type,
+        log_level,
     )
     .await?;
 
@@ -287,9 +291,16 @@ async fn create_or_update_replicator_stateful_set(
     replicator_image: String,
     environment: Environment,
     destination_type: DestinationType,
+    log_level: LogLevel,
 ) -> Result<(), PipelineError> {
     k8s_client
-        .create_or_update_stateful_set(prefix, &replicator_image, environment, destination_type)
+        .create_or_update_stateful_set(
+            prefix,
+            &replicator_image,
+            environment,
+            destination_type,
+            log_level,
+        )
         .await?;
 
     Ok(())

etl-api/src/k8s/http.rs

@@ -1,3 +1,4 @@
+use crate::configs::log::LogLevel;
 use crate::k8s::DestinationType;
 use crate::k8s::{K8sClient, K8sError, PodPhase};
 use async_trait::async_trait;
@@ -329,6 +330,7 @@ impl K8sClient for HttpK8sClient {
         replicator_image: &str,
         environment: Environment,
         destination_type: DestinationType,
+        log_level: LogLevel,
     ) -> Result<(), K8sError> {
         debug!("patching stateful set");
 
@@ -341,6 +343,7 @@ impl K8sClient for HttpK8sClient {
             &environment,
             replicator_image,
             destination_type,
+            log_level,
         );
 
         let node_selector = create_node_selector_json(&environment);
@@ -567,6 +570,7 @@ fn create_container_environment_json(
     environment: &Environment,
     replicator_image: &str,
     destination_type: DestinationType,
+    log_level: LogLevel,
 ) -> Vec<serde_json::Value> {
     let mut container_environment = vec![
         json!({
@@ -578,6 +582,10 @@ fn create_container_environment_json(
             //TODO: set APP_VERSION to proper version instead of the replicator image name
             "value": replicator_image
         }),
+        json!({
+            "name": "RUST_LOG",
+            "value": log_level.to_string()
+        }),
     ];
 
     match environment {
@@ -1100,6 +1108,7 @@ mod tests {
             &environment,
             replicator_image,
             DestinationType::BigQuery,
+            LogLevel::Info,
         );
         assert_json_snapshot!(container_environment);
 
@@ -1109,6 +1118,7 @@ mod tests {
             &environment,
             replicator_image,
             DestinationType::BigQuery,
+            LogLevel::Info,
         );
         assert_json_snapshot!(container_environment);
 
@@ -1118,6 +1128,7 @@ mod tests {
             &environment,
             replicator_image,
             DestinationType::BigQuery,
+            LogLevel::Info,
         );
         assert_json_snapshot!(container_environment);
     }
@@ -1132,6 +1143,7 @@ mod tests {
             &Environment::Dev,
             replicator_image,
             DestinationType::Iceberg,
+            LogLevel::Info,
         );
         assert_json_snapshot!(container_environment);
 
@@ -1140,6 +1152,7 @@ mod tests {
             &Environment::Staging,
             replicator_image,
             DestinationType::Iceberg,
+            LogLevel::Info,
         );
         assert_json_snapshot!(container_environment);
 
@@ -1148,6 +1161,7 @@ mod tests {
             &Environment::Prod,
             replicator_image,
             DestinationType::Iceberg,
+            LogLevel::Info,
         );
         assert_json_snapshot!(container_environment);
     }
@@ -1231,6 +1245,7 @@ mod tests {
             &environment,
             replicator_image,
             DestinationType::BigQuery,
+            LogLevel::Info,
         );
 
         let node_selector = create_node_selector_json(&environment);
@@ -1261,6 +1276,7 @@ mod tests {
             &environment,
             replicator_image,
             DestinationType::BigQuery,
+            LogLevel::Info,
         );
 
         let node_selector = create_node_selector_json(&environment);
@@ -1291,6 +1307,7 @@ mod tests {
             &environment,
             replicator_image,
             DestinationType::BigQuery,
+            LogLevel::Info,
         );
 
         let node_selector = create_node_selector_json(&environment);
@@ -1328,6 +1345,7 @@ mod tests {
             &environment,
             replicator_image,
             DestinationType::Iceberg,
+            LogLevel::Info,
         );
 
         let node_selector = create_node_selector_json(&environment);
@@ -1358,6 +1376,7 @@ mod tests {
             &environment,
             replicator_image,
             DestinationType::Iceberg,
+            LogLevel::Info,
         );
 
         let node_selector = create_node_selector_json(&environment);
@@ -1388,6 +1407,7 @@ mod tests {
             &environment,
             replicator_image,
             DestinationType::Iceberg,
+            LogLevel::Info,
         );
 
         let node_selector = create_node_selector_json(&environment);

etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_bq_container_environment-2.snap

@@ -11,6 +11,10 @@ expression: container_environment
     "name": "APP_VERSION",
     "value": "ramsup/etl-replicator:2a41356af735f891de37d71c0e1a62864fe4630e"
   },
+  {
+    "name": "RUST_LOG",
+    "value": "info"
+  },
   {
     "name": "APP_SENTRY__DSN",
     "valueFrom": {

etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_bq_container_environment-3.snap

@@ -11,6 +11,10 @@ expression: container_environment
     "name": "APP_VERSION",
     "value": "ramsup/etl-replicator:2a41356af735f891de37d71c0e1a62864fe4630e"
   },
+  {
+    "name": "RUST_LOG",
+    "value": "info"
+  },
   {
     "name": "APP_SENTRY__DSN",
     "valueFrom": {

etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_bq_container_environment.snap

@@ -11,6 +11,10 @@ expression: container_environment
     "name": "APP_VERSION",
     "value": "ramsup/etl-replicator:2a41356af735f891de37d71c0e1a62864fe4630e"
   },
+  {
+    "name": "RUST_LOG",
+    "value": "info"
+  },
   {
     "name": "APP_PIPELINE__PG_CONNECTION__PASSWORD",
     "valueFrom": {

etl-api/src/k8s/snapshots/etl_api__k8s__http__tests__create_bq_replicator_stateful_set_json-2.snap

@@ -38,6 +38,10 @@ expression: stateful_set_json
                 "name": "APP_VERSION",
                 "value": "ramsup/etl-replicator:2a41356af735f891de37d71c0e1a62864fe4630e"
               },
+              {
+                "name": "RUST_LOG",
+                "value": "info"
+              },
               {
                 "name": "APP_SENTRY__DSN",
                 "valueFrom": {