Merge branch 'master' into message_bus_final

Author: numinnex

Cargo.lock

@@ -34,6 +34,7 @@ use figment::{
 use serde::{Serialize, de::DeserializeOwned};
 use std::{env, fmt::Display, future::Future, marker::PhantomData, path::Path};
 use toml::{Value as TomlValue, map::Map as TomlMap};
+use tracing::{error, info, warn};
 
 const SECRET_MASK: &str = "******";
 const ARRAY_SEPARATOR: char = '_';
@@ -162,7 +163,7 @@ impl<T: ConfigurationType> CustomEnvProvider<T> {
                 value = SECRET_MASK.to_string();
             }
 
-            println!("{env_key} value changed to: {value} from environment variable");
+            info!("{env_key} value changed to: {value} from environment variable");
             Self::insert_environment_override(&source_dict, &mut target_dict, keys, env_var_value);
         }
 
@@ -752,28 +753,30 @@ fn file_exists<P: AsRef<Path>>(path: P) -> bool {
 
 impl<T: ConfigurationType, P: Provider + Clone> ConfigProvider<T> for FileConfigProvider<P> {
     async fn load_config(&self) -> Result<T, ConfigurationError> {
-        println!("Loading config from path: '{}'...", self.file_path);
+        info!("Loading config from path: '{}'...", self.file_path);
 
         // Start with the default configuration if provided
         let mut config_builder = Figment::new();
         let has_default = self.default_config.is_some();
         if let Some(default) = &self.default_config {
             config_builder = config_builder.merge(default);
         } else {
-            println!("No default configuration provided.");
+            warn!("No default configuration provided.");
         }
 
         // If the config file exists, merge it into the configuration
         if file_exists(&self.file_path) {
-            println!("Found configuration file at path: '{}'.", self.file_path);
+            info!("Found configuration file at path: '{}'.", self.file_path);
             config_builder = config_builder.merge(Toml::file(&self.file_path));
         } else {
-            println!(
+            warn!(
                 "Configuration file not found at path: '{}'.",
                 self.file_path
             );
             if has_default {
-                println!("Using default configuration as no config file was found.");
+                info!(
+                    "Using default configuration embedded into server, as no config file was found."
+                );
             }
         }
 
@@ -785,17 +788,17 @@ impl<T: ConfigurationType, P: Provider + Clone> ConfigProvider<T> for FileConfig
 
         match config_result {
             Ok(config) => {
-                println!("Config loaded successfully.");
+                info!("Config loaded successfully.");
                 let display_config = env::var(DISPLAY_CONFIG_ENV)
                     .map(|val| val == "1" || val.to_lowercase() == "true")
                     .unwrap_or(self.display_config);
                 if display_config {
-                    println!("Using Config: {config}");
+                    info!("Using Config: {config}");
                 }
                 Ok(config)
             }
             Err(e) => {
-                println!("Failed to load config: {e}");
+                error!("Failed to load config: {e}");
                 Err(ConfigurationError::CannotLoadConfiguration)
             }
         }

core/common/src/configs/mod.rs

@@ -340,9 +340,16 @@ path = "runtime"
 # Path for storing log files.
 path = "logs"
 
-# Level of logging detail. Options: "debug", "info", "warn", "error".
+# Log filtering directive using the same syntax as the RUST_LOG environment variable.
+# Supports simple levels ("trace", "debug", "info", "warn", "error", "off" or "none")
+# as well as complex directives like "warn,server=debug,iggy=trace".
+# Note: RUST_LOG environment variable always takes precedence over this setting.
 level = "info"
 
+# Whether to write logs to file. When false, logs are only written to stdout.
+# When enabled, logs are stored in {system.path}/{system.logging.path} (default: local_data/logs).
+file_enabled = true
+
 # Maximum size of the log files before rotation.
 max_size = "512 MB"
 

core/configs/server.toml

@@ -47,6 +47,8 @@ mimalloc = { workspace = true }
 once_cell = { workspace = true }
 postcard = { workspace = true }
 reqwest = { workspace = true }
+reqwest-middleware = { workspace = true }
+reqwest-retry = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 serde_with = { workspace = true }

core/connectors/runtime/Cargo.toml

@@ -65,6 +65,13 @@ timeout = "10s"
 [connectors.request_headers]
 api-key = "your-api-key"
 
+[connectors.retry]
+enabled = true
+max_attempts = 3
+initial_backoff = "1 s"
+max_backoff = "30 s"
+backoff_multiplier = 2
+
 [connectors.url_templates]
 # Optional: Customize URL templates for specific operations
 # If not specified, default RESTful URL patterns are used

core/connectors/runtime/README.md

@@ -242,6 +242,7 @@ pub async fn create_connectors_config_provider(
                 &config.request_headers,
                 &config.url_templates,
                 &config.response,
+                &config.retry,
             )?;
             Ok(Box::new(provider))
         }

core/connectors/runtime/src/configs/connectors.rs

@@ -26,10 +26,12 @@ use crate::configs::connectors::{
     ConnectorConfigVersions, ConnectorsConfig, ConnectorsConfigProvider, CreateSinkConfig,
     CreateSourceConfig, SinkConfig, SourceConfig,
 };
-use crate::configs::runtime::ResponseConfig;
+use crate::configs::runtime::{ResponseConfig, RetryConfig};
 use crate::error::RuntimeError;
 use async_trait::async_trait;
 use reqwest;
+use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
+use reqwest_retry::{Jitter, RetryTransientMiddleware, policies::ExponentialBackoff};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::str::FromStr;
@@ -43,7 +45,7 @@ struct SetActiveVersionRequest {
 pub struct HttpConnectorsConfigProvider {
     url_builder: UrlBuilder,
     response_extractor: ResponseExtractor,
-    client: reqwest::Client,
+    client: ClientWithMiddleware,
 }
 
 impl HttpConnectorsConfigProvider {
@@ -53,6 +55,7 @@ impl HttpConnectorsConfigProvider {
         request_headers: &HashMap<String, String>,
         url_templates: &HashMap<String, String>,
         response_config: &ResponseConfig,
+        retry_config: &RetryConfig,
     ) -> Result<Self, RuntimeError> {
         let mut headers = reqwest::header::HeaderMap::new();
         for (key, value) in request_headers {
@@ -75,13 +78,35 @@ impl HttpConnectorsConfigProvider {
                 RuntimeError::InvalidConfiguration(format!("Failed to build HTTP client: {err}"))
             })?;
 
+        let mut client_with_middleware = ClientBuilder::new(client);
+
+        if retry_config.enabled {
+            tracing::trace!("Apply retry config: {:?}", retry_config);
+
+            let retry_policy = ExponentialBackoff::builder()
+                .retry_bounds(
+                    retry_config.initial_backoff.get_duration(),
+                    retry_config.max_backoff.get_duration(),
+                )
+                .base(retry_config.backoff_multiplier)
+                .jitter(Jitter::Bounded)
+                .build_with_max_retries(retry_config.max_attempts);
+
+            let retry_transient_middleware =
+                RetryTransientMiddleware::new_with_policy(retry_policy);
+
+            client_with_middleware = client_with_middleware.with(retry_transient_middleware);
+        }
+
+        let final_client = client_with_middleware.build();
+
         let url_builder = UrlBuilder::new(base_url, url_templates);
         let response_extractor = ResponseExtractor::new(response_config);
 
         Ok(Self {
             url_builder,
             response_extractor,
-            client,
+            client: final_client,
         })
     }
 

core/connectors/runtime/src/configs/connectors/http_provider.rs

@@ -53,6 +53,44 @@ pub struct IggyTlsConfig {
     pub domain: Option<String>,
 }
 
+#[serde_as]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RetryConfig {
+    pub enabled: bool,
+    pub max_attempts: u32,
+    #[serde_as(as = "DisplayFromStr")]
+    pub initial_backoff: IggyDuration,
+    #[serde_as(as = "DisplayFromStr")]
+    pub max_backoff: IggyDuration,
+    pub backoff_multiplier: u32,
+}
+
+impl Default for RetryConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            max_attempts: 3,
+            initial_backoff: IggyDuration::new_from_secs(1),
+            max_backoff: IggyDuration::new_from_secs(30),
+            backoff_multiplier: 2,
+        }
+    }
+}
+
+impl Display for RetryConfig {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{{ enabled: {}, max_attempts: {}, initial_backoff: {}, max_backoff: {}, backoff_multiplier: {} }}",
+            self.enabled,
+            self.max_attempts,
+            self.initial_backoff,
+            self.max_backoff,
+            self.backoff_multiplier
+        )
+    }
+}
+
 #[derive(Debug, Default, Clone, Deserialize, Serialize)]
 #[serde(default)]
 pub struct LocalConnectorsConfig {
@@ -72,18 +110,21 @@ pub struct HttpConnectorsConfig {
     pub url_templates: HashMap<String, String>,
     #[serde(default)]
     pub response: ResponseConfig,
+    #[serde(default)]
+    pub retry: RetryConfig,
 }
 
 impl Display for HttpConnectorsConfig {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "{{ type: \"http\", base_url: {:?}, request_headers: {:?}, timeout: {}, url_templates: {:?}, response: {:?} }}",
+            "{{ type: \"http\", base_url: {:?}, request_headers: {:?}, timeout: {}, url_templates: {:?}, response: {:?}, retry: {} }}",
             self.base_url,
             self.request_headers.keys(),
             self.timeout,
             self.url_templates,
-            self.response
+            self.response,
+            self.retry
         )
     }
 }
@@ -99,6 +140,7 @@ pub struct ResponseConfig {
     pub error_path: Option<String>,
 }
 
+#[allow(clippy::large_enum_variant)]
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(tag = "config_type", rename_all = "lowercase")]
 pub enum ConnectorsConfig {

core/connectors/runtime/src/configs/runtime.rs

@@ -399,6 +399,7 @@ impl Default for LoggingConfig {
         LoggingConfig {
             path: SERVER_CONFIG.system.logging.path.parse().unwrap(),
             level: SERVER_CONFIG.system.logging.level.parse().unwrap(),
+            file_enabled: SERVER_CONFIG.system.logging.file_enabled,
             max_size: SERVER_CONFIG.system.logging.max_size.parse().unwrap(),
             retention: SERVER_CONFIG.system.logging.retention.parse().unwrap(),
             sysinfo_print_interval: SERVER_CONFIG

core/server/src/configs/defaults.rs

@@ -248,9 +248,10 @@ impl Display for LoggingConfig {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "{{ path: {}, level: {}, max_size: {}, retention: {} }}",
+            "{{ path: {}, level: {}, file_enabled: {}, max_size: {}, retention: {} }}",
             self.path,
             self.level,
+            self.file_enabled,
             self.max_size.as_human_string_with_zero_as_unlimited(),
             self.retention
         )