This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch connectors_test
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/connectors_test by this push:
new 971d8e0b Cleanup common config loader, improve configs for mcp and
connectors
971d8e0b is described below
commit 971d8e0bb93cb334d55b2ef1208a6b90b2ec5d02
Author: spetz <[email protected]>
AuthorDate: Mon Sep 22 21:45:16 2025 +0200
Cleanup common config loader, improve configs for mcp and connectors
---
Cargo.lock | 4 +-
core/ai/mcp/Cargo.toml | 2 +-
core/ai/mcp/README.md | 4 +-
core/ai/mcp/config.toml | 8 +-
core/ai/mcp/src/api.rs | 31 ++---
core/ai/mcp/src/configs.rs | 116 ++++++++++++++----
core/ai/mcp/src/error.rs | 2 -
core/ai/mcp/src/main.rs | 26 ++--
core/ai/mcp/src/stream.rs | 11 +-
core/common/src/configs/mod.rs | 28 ++---
core/connectors/README.md | 4 +-
core/connectors/runtime/Cargo.toml | 2 +-
core/connectors/runtime/README.md | 4 +-
core/connectors/runtime/config.toml | 4 +-
core/connectors/runtime/example_config.toml | 4 +-
core/connectors/runtime/src/api/auth.rs | 9 +-
core/connectors/runtime/src/api/config.rs | 62 +++++++++-
core/connectors/runtime/src/api/mod.rs | 31 ++---
core/connectors/runtime/src/configs.rs | 165 ++++++++++++++++++++------
core/connectors/runtime/src/context.rs | 15 ++-
core/connectors/runtime/src/main.rs | 4 +-
core/connectors/runtime/src/stream.rs | 11 +-
core/integration/src/test_server.rs | 2 +-
core/integration/tests/config_provider/mod.rs | 2 +-
core/server/src/configs/server.rs | 10 +-
25 files changed, 361 insertions(+), 200 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 644474fe..97037297 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3997,7 +3997,7 @@ dependencies = [
[[package]]
name = "iggy-connectors"
-version = "0.1.0"
+version = "0.1.1"
dependencies = [
"axum 0.8.4",
"axum-server",
@@ -4028,7 +4028,7 @@ dependencies = [
[[package]]
name = "iggy-mcp"
-version = "0.1.0"
+version = "0.1.1"
dependencies = [
"axum 0.8.4",
"axum-server",
diff --git a/core/ai/mcp/Cargo.toml b/core/ai/mcp/Cargo.toml
index 68e733c1..8e26507a 100644
--- a/core/ai/mcp/Cargo.toml
+++ b/core/ai/mcp/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-mcp"
-version = "0.1.0"
+version = "0.1.1"
description = "MCP Server for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/ai/mcp/README.md b/core/ai/mcp/README.md
index c16b515a..999ac7e8 100644
--- a/core/ai/mcp/README.md
+++ b/core/ai/mcp/README.md
@@ -33,8 +33,8 @@ allow_private_network = false
[http.tls] # Optional TLS configuration for HTTP API
enabled = false
-cert = "core/certs/iggy_cert.pem"
-key = "core/certs/iggy_key.pem"
+cert_file = "core/certs/iggy_cert.pem"
+key_file = "core/certs/iggy_key.pem"
[permissions]
create = true
diff --git a/core/ai/mcp/config.toml b/core/ai/mcp/config.toml
index 357502cd..41f5b572 100644
--- a/core/ai/mcp/config.toml
+++ b/core/ai/mcp/config.toml
@@ -32,15 +32,15 @@ allow_private_network = false
[http.tls] # Optional TLS configuration for HTTP API
enabled = false
-cert = "core/certs/iggy_cert.pem"
-key = "core/certs/iggy_key.pem"
+cert_file = "core/certs/iggy_cert.pem"
+key_file = "core/certs/iggy_key.pem"
[iggy]
address = "localhost:8090"
username = "iggy"
password = "iggy"
-# token = "secret" # Personal Access Token (PAT) can be used instead of
username and password
-# consumer = "iggy-mcp" # Optional consumer name
+token = "" # Personal Access Token (PAT) can be used instead of username and
password
+consumer = "iggy-mcp" # Optional consumer name
[permissions]
create = true
diff --git a/core/ai/mcp/src/api.rs b/core/ai/mcp/src/api.rs
index e37e2320..aae86612 100644
--- a/core/ai/mcp/src/api.rs
+++ b/core/ai/mcp/src/api.rs
@@ -18,7 +18,7 @@
use crate::{
Permissions,
- configs::{HttpApiConfig, configure_cors},
+ configs::{HttpConfig, configure_cors},
error::McpRuntimeError,
service::IggyService,
};
@@ -36,7 +36,7 @@ use tokio::spawn;
use tracing::{error, info};
pub async fn init(
- config: HttpApiConfig,
+ config: HttpConfig,
iggy_client: Arc<IggyClient>,
iggy_consumer: Arc<Consumer>,
permissions: Permissions,
@@ -71,19 +71,11 @@ pub async fn init(
)
.nest_service(&config.path, service);
- if let Some(cors) = &config.cors
- && cors.enabled
- {
- app = app.layer(configure_cors(cors));
+ if config.cors.enabled {
+ app = app.layer(configure_cors(&config.cors));
}
- let tls_enabled = config
- .tls
- .as_ref()
- .map(|tls| tls.enabled)
- .unwrap_or_default();
-
- if !tls_enabled {
+ if !config.tls.enabled {
let listener = tokio::net::TcpListener::bind(&config.address)
.await
.map_err(|error| {
@@ -110,12 +102,15 @@ pub async fn init(
return Ok(());
}
- let tls = config.tls.as_ref().expect("TLS configuration is required");
- let tls_config = RustlsConfig::from_pem_file(PathBuf::from(&tls.cert),
PathBuf::from(&tls.key))
- .await
- .unwrap();
+ let tls_config = RustlsConfig::from_pem_file(
+ PathBuf::from(&config.tls.cert_file),
+ PathBuf::from(&config.tls.key_file),
+ )
+ .await
+ .expect("Failed to load TLS certificate or key file");
- let listener = std::net::TcpListener::bind(&config.address).unwrap();
+ let listener =
+ std::net::TcpListener::bind(&config.address).expect("Failed to bind
TCP listener");
let address = listener
.local_addr()
.expect("Failed to get local address for HTTPS / TLS server");
diff --git a/core/ai/mcp/src/configs.rs b/core/ai/mcp/src/configs.rs
index b1666309..6dbf5c72 100644
--- a/core/ai/mcp/src/configs.rs
+++ b/core/ai/mcp/src/configs.rs
@@ -16,6 +16,8 @@
* under the License.
*/
+use std::fmt::Formatter;
+
use axum::http::Method;
use figment::{
Metadata, Profile, Provider,
@@ -31,7 +33,7 @@ use tower_http::cors::{AllowOrigin, CorsLayer};
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(default)]
pub struct McpServerConfig {
- pub http: Option<HttpApiConfig>,
+ pub http: HttpConfig,
pub iggy: IggyConfig,
pub permissions: PermissionsConfig,
pub transport: McpTransport,
@@ -40,18 +42,10 @@ pub struct McpServerConfig {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IggyConfig {
pub address: String,
- pub username: Option<String>,
- pub password: Option<String>,
- pub token: Option<String>,
- pub consumer: Option<String>,
-}
-
-#[derive(Debug, Clone, Deserialize, Serialize)]
-pub struct HttpApiConfig {
- pub address: String,
- pub path: String,
- pub cors: Option<HttpCorsConfig>,
- pub tls: Option<HttpTlsConfig>,
+ pub username: String,
+ pub password: String,
+ pub token: String,
+ pub consumer: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
@@ -62,11 +56,19 @@ pub struct PermissionsConfig {
pub delete: bool,
}
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct HttpConfig {
+ pub address: String,
+ pub path: String,
+ pub cors: HttpCorsConfig,
+ pub tls: HttpTlsConfig,
+}
+
#[derive(Debug, Default, Deserialize, Serialize, Clone)]
pub struct HttpTlsConfig {
pub enabled: bool,
- pub cert: String,
- pub key: String,
+ pub cert_file: String,
+ pub key_file: String,
}
#[derive(Debug, Default, Deserialize, Serialize, Clone)]
@@ -94,7 +96,7 @@ pub enum McpTransport {
impl Default for McpServerConfig {
fn default() -> Self {
Self {
- http: Some(HttpApiConfig::default()),
+ http: HttpConfig::default(),
iggy: IggyConfig::default(),
permissions: PermissionsConfig::default(),
transport: McpTransport::Http,
@@ -106,21 +108,21 @@ impl Default for IggyConfig {
fn default() -> Self {
Self {
address: "localhost:8090".to_owned(),
- username: Some(DEFAULT_ROOT_USERNAME.to_owned()),
- password: Some(DEFAULT_ROOT_PASSWORD.to_owned()),
- token: None,
- consumer: None,
+ username: DEFAULT_ROOT_USERNAME.to_owned(),
+ password: DEFAULT_ROOT_PASSWORD.to_owned(),
+ token: "".to_owned(),
+ consumer: "iggy-mcp".to_owned(),
}
}
}
-impl Default for HttpApiConfig {
+impl Default for HttpConfig {
fn default() -> Self {
Self {
address: "localhost:8082".to_owned(),
path: "/mcp".to_owned(),
- tls: None,
- cors: None,
+ cors: HttpCorsConfig::default(),
+ tls: HttpTlsConfig::default(),
}
}
}
@@ -184,16 +186,80 @@ pub fn configure_cors(config: &HttpCorsConfig) ->
CorsLayer {
.allow_private_network(config.allow_private_network)
}
+impl std::fmt::Display for IggyConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ address: {}, username: {}, password: {}, token: {}, consumer:
{} }}",
+ self.address,
+ self.username,
+ if !self.password.is_empty() {
+ "****"
+ } else {
+ ""
+ },
+ if !self.token.is_empty() { "****" } else { "" },
+ self.consumer
+ )
+ }
+}
+
+impl std::fmt::Display for PermissionsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ create: {}, read: {}, update: {}, delete: {} }}",
+ self.create, self.read, self.update, self.delete
+ )
+ }
+}
+
impl std::fmt::Display for McpServerConfig {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "McpServerConfig {{ http: {:?}, iggy: {:?}, permissions: {:?},
transport: {:?} }}",
+ "{{ http: {}, iggy: {}, permissions: {:?}, transport: {} }}",
self.http, self.iggy, self.permissions, self.transport
)
}
}
+impl std::fmt::Display for HttpConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ address: {}, path: {}, cors: {}, tls: {} }}",
+ self.address, self.path, self.cors, self.tls
+ )
+ }
+}
+
+impl std::fmt::Display for HttpTlsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, cert_file: {}, key_file: {} }}",
+ self.enabled, self.cert_file, self.key_file
+ )
+ }
+}
+
+impl std::fmt::Display for HttpCorsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, allowed_methods: {:?}, allowed_origins: {:?},
allowed_headers: {:?}, exposed_headers: {:?}, allow_credentials: {},
allow_private_network: {} }}",
+ self.enabled,
+ self.allowed_methods,
+ self.allowed_origins,
+ self.allowed_headers,
+ self.exposed_headers,
+ self.allow_credentials,
+ self.allow_private_network
+ )
+ }
+}
+
impl McpServerConfig {
pub fn config_provider(path: String) ->
FileConfigProvider<McpServerEnvProvider> {
let default_config =
Toml::string(include_str!("../../../ai/mcp/config.toml"));
diff --git a/core/ai/mcp/src/error.rs b/core/ai/mcp/src/error.rs
index 0344cfac..efdcb606 100644
--- a/core/ai/mcp/src/error.rs
+++ b/core/ai/mcp/src/error.rs
@@ -22,8 +22,6 @@ use thiserror::Error;
pub enum McpRuntimeError {
#[error("Failed to create service")]
FailedToCreateService,
- #[error("Missing configuration")]
- MissingConfig,
#[error("Failed to start HTTP server")]
FailedToStartHttpServer,
#[error("Iggy client error")]
diff --git a/core/ai/mcp/src/main.rs b/core/ai/mcp/src/main.rs
index 0b5792ed..b5cfb428 100644
--- a/core/ai/mcp/src/main.rs
+++ b/core/ai/mcp/src/main.rs
@@ -77,11 +77,14 @@ async fn main() -> Result<(), McpRuntimeError> {
info!("Starting Iggy MCP Server, transport: {transport}...");
- let consumer_id = Identifier::from_str_value(
- config.iggy.consumer.as_deref().unwrap_or("iggy-mcp"),
- )
- .map_err(|error| {
- error!("Failed to create Iggy consumer ID: {:?}", error);
+ let consumer = if config.iggy.consumer.is_empty() {
+ "iggy-mcp"
+ } else {
+ config.iggy.consumer.as_str()
+ };
+
+ let consumer_id = Identifier::from_str_value(consumer).map_err(|error| {
+ error!("Failed to create Iggy consumer ID: {consumer}. {error}",);
McpRuntimeError::FailedToCreateConsumerId
})?;
let iggy_consumer = Arc::new(iggy::prelude::Consumer::new(consumer_id));
@@ -98,8 +101,8 @@ async fn main() -> Result<(), McpRuntimeError> {
let Ok(service) = IggyService::new(iggy_client, iggy_consumer,
permissions)
.serve(stdio())
.await
- .inspect_err(|e| {
- error!("Serving error: {:?}", e);
+ .inspect_err(|error| {
+ error!("Serving error. {error}");
})
else {
error!("Failed to create service");
@@ -107,15 +110,10 @@ async fn main() -> Result<(), McpRuntimeError> {
};
if let Err(error) = service.waiting().await {
- error!("waiting error: {:?}", error);
+ error!("Waiting for service error. {error}");
}
} else {
- let Some(http_config) = config.http else {
- error!("HTTP API configuration not found");
- return Err(McpRuntimeError::MissingConfig);
- };
-
- api::init(http_config, iggy_client, iggy_consumer, permissions).await?;
+ api::init(config.http, iggy_client, iggy_consumer, permissions).await?;
}
#[cfg(unix)]
diff --git a/core/ai/mcp/src/stream.rs b/core/ai/mcp/src/stream.rs
index da47faeb..4b56df03 100644
--- a/core/ai/mcp/src/stream.rs
+++ b/core/ai/mcp/src/stream.rs
@@ -25,24 +25,17 @@ pub async fn init(config: IggyConfig) -> Result<IggyClient,
McpRuntimeError> {
let password = config.password;
let token = config.token;
- let connection_string = if let Some(token) = token {
- if token.is_empty() {
- error!("Iggy token cannot be empty (if username and password are
not provided)");
- return Err(McpRuntimeError::MissingIggyCredentials);
- }
-
+ let connection_string = if !token.is_empty() {
let redacted_token = token.chars().take(3).collect::<String>();
info!("Using token: {redacted_token}*** for Iggy authentication");
format!("iggy://{token}@{address}")
} else {
info!("Using username and password for Iggy authentication");
- let username =
username.ok_or(McpRuntimeError::MissingIggyCredentials)?;
if username.is_empty() {
- error!("Iggy password cannot be empty (if token is not provided)");
+ error!("Iggy username cannot be empty (if token is not provided)");
return Err(McpRuntimeError::MissingIggyCredentials);
}
- let password =
password.ok_or(McpRuntimeError::MissingIggyCredentials)?;
if password.is_empty() {
error!("Iggy password cannot be empty (if token is not provided)");
return Err(McpRuntimeError::MissingIggyCredentials);
diff --git a/core/common/src/configs/mod.rs b/core/common/src/configs/mod.rs
index cf9159ba..0d0071c2 100644
--- a/core/common/src/configs/mod.rs
+++ b/core/common/src/configs/mod.rs
@@ -34,28 +34,22 @@ 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};
-// Constants for configuration handling
const SECRET_MASK: &str = "******";
const ARRAY_SEPARATOR: char = '_';
const PATH_SEPARATOR: &str = ".";
+const DISPLAY_CONFIG_ENV: &str = "IGGY_DISPLAY_CONFIG";
-/// Type alias for configuration profiles mapping
type ProfileMap = FigmentMap<Profile, Dict>;
/// Type alias for configuration that can be serialized, deserialized, has
defaults, and can be displayed
pub trait ConfigurationType: Serialize + DeserializeOwned + Default + Display
{}
impl<T: Serialize + DeserializeOwned + Default + Display> ConfigurationType
for T {}
-/// Errors that can occur during configuration loading and processing
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ConfigurationError {
- /// Failed to load configuration from file or environment
CannotLoadConfiguration,
- /// Failed to serialize default configuration to TOML
DefaultSerializationFailed,
- /// Failed to parse default configuration TOML
DefaultParsingFailed,
- /// Environment variable parsing failed
EnvironmentVariableParsingFailed,
}
@@ -76,9 +70,7 @@ impl Display for ConfigurationError {
impl std::error::Error for ConfigurationError {}
-/// Trait for configuration providers that can load configuration of a
specific type
pub trait ConfigProvider<T: ConfigurationType> {
- /// Load configuration asynchronously
fn load_config(self) -> impl Future<Output = Result<T,
ConfigurationError>>;
}
@@ -96,8 +88,8 @@ impl<P: Provider> FileConfigProvider<P> {
/// # Arguments
/// * `file_path` - Path to the configuration file
/// * `env_provider` - Environment variable provider
- /// * `default_config` - Optional default configuration data
/// * `display_config` - Whether to display the loaded configuration
+ /// * `default_config` - Optional default configuration data
pub fn new(
file_path: String,
env_provider: P,
@@ -113,15 +105,6 @@ impl<P: Provider> FileConfigProvider<P> {
}
}
-/// Custom environment variable provider with advanced override capabilities
-///
-/// This provider supports:
-/// - Prefix-based filtering of environment variables
-/// - Secret key masking for security
-/// - Complex path resolution for nested structures
-/// - Array index overrides
-/// - HashMap field overrides
-/// - JSON value field handling
#[derive(Debug, Clone)]
pub struct CustomEnvProvider<T: ConfigurationType> {
prefix: String,
@@ -745,7 +728,7 @@ impl<T: ConfigurationType, P: Provider + Clone>
ConfigProvider<T> for FileConfig
async fn load_config(self) -> Result<T, ConfigurationError> {
println!("Loading config from path: '{}'...", self.file_path);
- // Start with the default configuration
+ // Start with the default configuration if provided
let mut config_builder = Figment::new();
if let Some(default) = self.default_config {
config_builder = config_builder.merge(default);
@@ -773,7 +756,10 @@ impl<T: ConfigurationType, P: Provider + Clone>
ConfigProvider<T> for FileConfig
match config_result {
Ok(config) => {
println!("Config loaded successfully.");
- if self.display_config {
+ 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}");
}
Ok(config)
diff --git a/core/connectors/README.md b/core/connectors/README.md
index 0dc256c0..a7532d5d 100644
--- a/core/connectors/README.md
+++ b/core/connectors/README.md
@@ -19,11 +19,11 @@ The [docker
image](https://hub.docker.com/r/apache/iggy-connect) is available, a
## Quick Start
-1. Build the project in release mode, and make sure that the plugins specified
in `core/connectors/config.toml` under `path` are available. You can use either
of `toml`, `json` or `yaml` formats for the configuration file.
+1. Build the project in release mode (or debug, and update the connectors
paths in the config accordingly), and make sure that the plugins specified in
`core/connectors/example_config.toml` under `path` are available. You can use
either of `toml`, `json` or `yaml` formats for the configuration file.
2. Run `docker compose up -d` from `/examples/rust/src/sink-data-producer`
which will start the Quickwit server to be used by an example sink connector.
At this point, you can access the Quickwit UI at
[http://localhost:7280](http://localhost:7280) - check this dashboard again
later on, after the `events` index will be created.
-3. Set environment variable
`IGGY_CONNECTORS_CONFIG_PATH=core/connectors/runtime/config` (adjust the path
as needed) pointing to the runtime configuration file.
+3. Set environment variable
`IGGY_CONNECTORS_CONFIG_PATH=core/connectors/runtime/example_config.toml`
(adjust the path as needed) pointing to the runtime configuration file.
4. Start the Iggy server and invoke the following commands via Iggy CLI to
create the example streams and topics used by the sample connectors.
diff --git a/core/connectors/runtime/Cargo.toml
b/core/connectors/runtime/Cargo.toml
index 759f12a2..8987cea7 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-connectors"
-version = "0.1.0"
+version = "0.1.1"
description = "Connectors runtime for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/runtime/README.md
b/core/connectors/runtime/README.md
index 34019292..9e5471a1 100644
--- a/core/connectors/runtime/README.md
+++ b/core/connectors/runtime/README.md
@@ -21,7 +21,7 @@ The minimal viable configuration requires at least the Iggy
credentials, to crea
address = "localhost:8090"
username = "iggy"
password = "iggy"
-# token = "secret" # Personal Access Token (PAT) can be used instead of
username and password
+token = "" # Personal Access Token (PAT) can be used instead of username and
password
[state]
path = "local_state"
@@ -39,7 +39,7 @@ Connector runtime has an optional HTTP API that can be
enabled by setting the `e
[http] # Optional HTTP API configuration
enabled = true
address = "127.0.0.1:8081"
-# api_key = "secret" # Optional API key for authentication to be passed as
`api-key` header
+api_key = "" # Optional API key for authentication to be passed as `api-key`
header
[http.cors] # Optional CORS configuration for HTTP API
enabled = false
diff --git a/core/connectors/runtime/config.toml
b/core/connectors/runtime/config.toml
index 3446708a..84801034 100644
--- a/core/connectors/runtime/config.toml
+++ b/core/connectors/runtime/config.toml
@@ -18,7 +18,7 @@
[http] # Optional HTTP API configuration
enabled = true
address = "127.0.0.1:8081"
-# api_key = "secret" # Optional API key for authentication to be passed as
`api-key` header
+api_key = "" # Optional API key for authentication to be passed as `api-key`
header
[http.cors] # Optional CORS configuration for HTTP API
enabled = false
@@ -38,7 +38,7 @@ key_file = "core/certs/iggy_key.pem"
address = "localhost:8090"
username = "iggy"
password = "iggy"
-# token = "secret" # Personal Access Token (PAT) can be used instead of
username and password
+token = "" # Personal Access Token (PAT) can be used instead of username and
password
[state]
path = "local_state"
diff --git a/core/connectors/runtime/example_config.toml
b/core/connectors/runtime/example_config.toml
index 3e00608c..da999c37 100644
--- a/core/connectors/runtime/example_config.toml
+++ b/core/connectors/runtime/example_config.toml
@@ -18,7 +18,7 @@
[http] # Optional HTTP API configuration
enabled = true
address = "127.0.0.1:8081"
-# api_key = "secret" # Optional API key for authentication to be passed as
`api-key` header
+api_key = "" # Optional API key for authentication to be passed as `api-key`
header
[http.cors] # Optional CORS configuration for HTTP API
enabled = false
@@ -38,7 +38,7 @@ key_file = "core/certs/iggy_key.pem"
address = "localhost:8090"
username = "iggy"
password = "iggy"
-# token = "secret" # Personal Access Token (PAT) can be used instead of
username and password
+token = "" # Personal Access Token (PAT) can be used instead of username and
password
[state]
path = "local_state"
diff --git a/core/connectors/runtime/src/api/auth.rs
b/core/connectors/runtime/src/api/auth.rs
index 8fb84a3d..6f4439af 100644
--- a/core/connectors/runtime/src/api/auth.rs
+++ b/core/connectors/runtime/src/api/auth.rs
@@ -27,7 +27,6 @@ use axum::{
use std::sync::Arc;
const API_KEY_HEADER: &str = "api-key";
-
const PUBLIC_PATHS: &[&str] = &["/", "/health"];
pub async fn resolve_api_key(
@@ -39,14 +38,10 @@ pub async fn resolve_api_key(
return Ok(next.run(request).await);
}
- let Some(expected_api_key) = context.api_key.as_ref() else {
+ if context.api_key.is_empty() {
return Ok(next.run(request).await);
};
- if expected_api_key.is_empty() {
- return Ok(next.run(request).await);
- }
-
let Some(api_key) = request
.headers()
.get(API_KEY_HEADER)
@@ -55,7 +50,7 @@ pub async fn resolve_api_key(
return Err(StatusCode::UNAUTHORIZED);
};
- if api_key != expected_api_key {
+ if api_key != context.api_key {
return Err(StatusCode::UNAUTHORIZED);
}
diff --git a/core/connectors/runtime/src/api/config.rs
b/core/connectors/runtime/src/api/config.rs
index 2422b62f..b2563efe 100644
--- a/core/connectors/runtime/src/api/config.rs
+++ b/core/connectors/runtime/src/api/config.rs
@@ -16,6 +16,8 @@
* under the License.
*/
+use std::fmt::Formatter;
+
use crate::{configs::ConfigFormat, error::RuntimeError};
use axum::http::{HeaderValue, Method};
use serde::{Deserialize, Serialize};
@@ -28,15 +30,15 @@ pub const TOML_HEADER: HeaderValue =
HeaderValue::from_static("application/toml"
pub const TEXT_HEADER: HeaderValue = HeaderValue::from_static("text/plain");
#[derive(Debug, Clone, Deserialize, Serialize)]
-pub struct HttpApiConfig {
+pub struct HttpConfig {
pub enabled: bool,
pub address: String,
- pub api_key: Option<String>,
- pub cors: Option<HttpCorsConfig>,
- pub tls: Option<HttpTlsConfig>,
+ pub api_key: String,
+ pub cors: HttpCorsConfig,
+ pub tls: HttpTlsConfig,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Default, Deserialize, Serialize, Clone)]
pub struct HttpCorsConfig {
pub enabled: bool,
pub allowed_methods: Vec<String>,
@@ -47,7 +49,7 @@ pub struct HttpCorsConfig {
pub allow_private_network: bool,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Default, Deserialize, Serialize, Clone)]
pub struct HttpTlsConfig {
pub enabled: bool,
pub cert_file: String,
@@ -129,3 +131,51 @@ pub fn configure_cors(config: &HttpCorsConfig) ->
CorsLayer {
.allow_credentials(config.allow_credentials)
.allow_private_network(config.allow_private_network)
}
+
+impl Default for HttpConfig {
+ fn default() -> Self {
+ Self {
+ enabled: true,
+ address: "localhost:8081".to_owned(),
+ api_key: "".to_owned(),
+ cors: HttpCorsConfig::default(),
+ tls: HttpTlsConfig::default(),
+ }
+ }
+}
+
+impl std::fmt::Display for HttpConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ address: {}, api_key: {}, cors: {}, tls: {} }}",
+ self.address, self.api_key, self.cors, self.tls
+ )
+ }
+}
+
+impl std::fmt::Display for HttpTlsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, cert_file: {}, key_file: {} }}",
+ self.enabled, self.cert_file, self.key_file
+ )
+ }
+}
+
+impl std::fmt::Display for HttpCorsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, allowed_methods: {:?}, allowed_origins: {:?},
allowed_headers: {:?}, exposed_headers: {:?}, allow_credentials: {},
allow_private_network: {} }}",
+ self.enabled,
+ self.allowed_methods,
+ self.allowed_origins,
+ self.allowed_headers,
+ self.exposed_headers,
+ self.allow_credentials,
+ self.allow_private_network
+ )
+ }
+}
diff --git a/core/connectors/runtime/src/api/mod.rs
b/core/connectors/runtime/src/api/mod.rs
index 6cb09dbd..54cae35b 100644
--- a/core/connectors/runtime/src/api/mod.rs
+++ b/core/connectors/runtime/src/api/mod.rs
@@ -20,7 +20,7 @@ use crate::context::RuntimeContext;
use auth::resolve_api_key;
use axum::{Json, Router, middleware, routing::get};
use axum_server::tls_rustls::RustlsConfig;
-use config::{HttpApiConfig, configure_cors};
+use config::{HttpConfig, configure_cors};
use std::{net::SocketAddr, path::PathBuf, sync::Arc};
use tokio::spawn;
use tracing::{error, info};
@@ -34,7 +34,7 @@ mod source;
const NAME: &str = env!("CARGO_PKG_NAME");
-pub async fn init(config: &HttpApiConfig, context: Arc<RuntimeContext>) {
+pub async fn init(config: &HttpConfig, context: Arc<RuntimeContext>) {
if !config.enabled {
info!("{NAME} HTTP API is disabled");
return;
@@ -54,18 +54,11 @@ pub async fn init(config: &HttpApiConfig, context:
Arc<RuntimeContext>) {
resolve_api_key,
));
- if let Some(cors) = &config.cors
- && cors.enabled
- {
- app = app.layer(configure_cors(cors));
+ if config.cors.enabled {
+ app = app.layer(configure_cors(&config.cors));
}
- let tls_enabled = config
- .tls
- .as_ref()
- .map(|tls| tls.enabled)
- .unwrap_or_default();
- if !tls_enabled {
+ if !config.tls.enabled {
let listener = tokio::net::TcpListener::bind(&config.address)
.await
.unwrap_or_else(|_| panic!("Failed to bind to HTTP address {}",
config.address));
@@ -86,13 +79,15 @@ pub async fn init(config: &HttpApiConfig, context:
Arc<RuntimeContext>) {
return;
}
- let tls = config.tls.as_ref().expect("TLS configuration is required");
- let tls_config =
- RustlsConfig::from_pem_file(PathBuf::from(&tls.cert_file),
PathBuf::from(&tls.key_file))
- .await
- .unwrap();
+ let tls_config = RustlsConfig::from_pem_file(
+ PathBuf::from(&config.tls.cert_file),
+ PathBuf::from(&config.tls.key_file),
+ )
+ .await
+ .expect("Failed to load TLS certificate or key file");
- let listener = std::net::TcpListener::bind(&config.address).unwrap();
+ let listener =
+ std::net::TcpListener::bind(&config.address).expect("Failed to bind
TCP listener");
let address = listener
.local_addr()
.expect("Failed to get local address for HTTPS / TLS server");
diff --git a/core/connectors/runtime/src/configs.rs
b/core/connectors/runtime/src/configs.rs
index 12504a69..32b71707 100644
--- a/core/connectors/runtime/src/configs.rs
+++ b/core/connectors/runtime/src/configs.rs
@@ -16,7 +16,7 @@
* under the License.
*/
-use crate::api::config::HttpApiConfig;
+use crate::api::config::HttpConfig;
use figment::{
Metadata, Profile, Provider,
providers::{Format, Toml},
@@ -26,7 +26,7 @@ use iggy::prelude::{DEFAULT_ROOT_PASSWORD,
DEFAULT_ROOT_USERNAME};
use iggy_common::{CustomEnvProvider, FileConfigProvider};
use iggy_connector_sdk::{Schema, transforms::TransformType};
use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
+use std::{collections::HashMap, fmt::Formatter};
use strum::Display;
#[derive(
@@ -47,8 +47,8 @@ pub enum ConfigFormat {
#[derive(Debug, Default, Clone, Deserialize, Serialize)]
#[serde(default)]
-pub struct RuntimeConfig {
- pub http: HttpApiConfig,
+pub struct ConnectorsConfig {
+ pub http: HttpConfig,
pub iggy: IggyConfig,
pub sinks: HashMap<String, SinkConfig>,
pub sources: HashMap<String, SourceConfig>,
@@ -58,20 +58,9 @@ pub struct RuntimeConfig {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IggyConfig {
pub address: String,
- pub username: Option<String>,
- pub password: Option<String>,
- pub token: Option<String>,
-}
-
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
-pub struct SinkConfig {
- pub enabled: bool,
- pub name: String,
- pub path: String,
- pub transforms: Option<TransformsConfig>,
- pub streams: Vec<StreamConsumerConfig>,
- pub config_format: Option<ConfigFormat>,
- pub config: Option<serde_json::Value>,
+ pub username: String,
+ pub password: String,
+ pub token: String,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
@@ -93,6 +82,17 @@ pub struct StreamProducerConfig {
pub linger_time: Option<String>,
}
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct SinkConfig {
+ pub enabled: bool,
+ pub name: String,
+ pub path: String,
+ pub transforms: Option<TransformsConfig>,
+ pub streams: Vec<StreamConsumerConfig>,
+ pub config_format: Option<ConfigFormat>,
+ pub config: Option<serde_json::Value>,
+}
+
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct SourceConfig {
pub enabled: bool,
@@ -120,16 +120,117 @@ pub struct StateConfig {
pub path: String,
}
-impl std::fmt::Display for RuntimeConfig {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+impl std::fmt::Display for StateConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(f, "{{ path: {} }}", self.path)
+ }
+}
+
+impl std::fmt::Display for ConnectorsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "RuntimeConfig {{ http: {:?}, iggy: {:?}, sinks: {:?}, sources:
{:?}, state: {:?} }}",
+ "{{ http: {}, iggy: {}, sinks: {:?}, sources: {:?}, state: {:} }}",
self.http, self.iggy, self.sinks, self.sources, self.state
)
}
}
+impl std::fmt::Display for IggyConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ address: {}, username: {}, password: {}, token: {} }}",
+ self.address,
+ self.username,
+ if !self.password.is_empty() {
+ "****"
+ } else {
+ ""
+ },
+ if !self.token.is_empty() { "****" } else { "" },
+ )
+ }
+}
+
+impl std::fmt::Display for StreamConsumerConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ stream: {}, topics: {}, schema: {:?}, batch_length: {:?},
poll_interval: {:?}, consumer_group: {:?} }}",
+ self.stream,
+ self.topics
+ .iter()
+ .map(|s| s.as_str())
+ .collect::<Vec<&str>>()
+ .join(", "),
+ self.schema,
+ self.batch_length,
+ self.poll_interval,
+ self.consumer_group
+ )
+ }
+}
+
+impl std::fmt::Display for StreamProducerConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ stream: {}, topic: {}, schema: {:?}, batch_length: {:?},
linger_time: {:?} }}",
+ self.stream, self.topic, self.schema, self.batch_length,
self.linger_time
+ )
+ }
+}
+
+impl std::fmt::Display for SinkConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, name: {}, path: {}, transforms: {:?}, streams:
[{}], config_format: {:?} }}",
+ self.enabled,
+ self.name,
+ self.path,
+ self.transforms,
+ self.streams
+ .iter()
+ .map(|s| s.to_string())
+ .collect::<Vec<String>>()
+ .join(", "),
+ self.config_format,
+ )
+ }
+}
+
+impl std::fmt::Display for SourceConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, name: {}, path: {}, transforms: {:?}, streams:
[{}], config_format: {:?} }}",
+ self.enabled,
+ self.name,
+ self.path,
+ self.transforms,
+ self.streams
+ .iter()
+ .map(|s| s.to_string())
+ .collect::<Vec<String>>()
+ .join(", "),
+ self.config_format,
+ )
+ }
+}
+
+impl std::fmt::Display for TransformsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ let transforms: Vec<String> = self
+ .transforms
+ .iter()
+ .map(|(k, v)| format!("{}: {}", k, v))
+ .collect();
+ write!(f, "{{ {} }}", transforms.join(", "))
+ }
+}
+
impl Default for StateConfig {
fn default() -> Self {
Self {
@@ -142,26 +243,14 @@ impl Default for IggyConfig {
fn default() -> Self {
Self {
address: "localhost:8090".to_owned(),
- username: Some(DEFAULT_ROOT_USERNAME.to_owned()),
- password: Some(DEFAULT_ROOT_PASSWORD.to_owned()),
- token: None,
- }
- }
-}
-
-impl Default for HttpApiConfig {
- fn default() -> Self {
- Self {
- enabled: true,
- address: "localhost:8081".to_owned(),
- cors: None,
- api_key: None,
- tls: None,
+ username: DEFAULT_ROOT_USERNAME.to_owned(),
+ password: DEFAULT_ROOT_PASSWORD.to_owned(),
+ token: "".to_owned(),
}
}
}
-impl RuntimeConfig {
+impl ConnectorsConfig {
pub fn config_provider(path: String) ->
FileConfigProvider<ConnectorsEnvProvider> {
let default_config =
Toml::string(include_str!("../../../connectors/runtime/config.toml"));
FileConfigProvider::new(
@@ -175,7 +264,7 @@ impl RuntimeConfig {
#[derive(Debug, Clone)]
pub struct ConnectorsEnvProvider {
- provider: CustomEnvProvider<RuntimeConfig>,
+ provider: CustomEnvProvider<ConnectorsConfig>,
}
impl Default for ConnectorsEnvProvider {
diff --git a/core/connectors/runtime/src/context.rs
b/core/connectors/runtime/src/context.rs
index d7dea9f8..1a9d6a2f 100644
--- a/core/connectors/runtime/src/context.rs
+++ b/core/connectors/runtime/src/context.rs
@@ -18,7 +18,7 @@
use crate::{
SinkConnectorWrapper, SourceConnectorWrapper,
- configs::RuntimeConfig,
+ configs::ConnectorsConfig,
manager::{
sink::{SinkDetails, SinkInfo, SinkManager},
source::{SourceDetails, SourceInfo, SourceManager},
@@ -29,22 +29,25 @@ use tracing::error;
pub struct RuntimeContext {
pub sinks: SinkManager,
pub sources: SourceManager,
- pub api_key: Option<String>,
+ pub api_key: String,
}
pub fn init(
- config: &RuntimeConfig,
+ config: &ConnectorsConfig,
sink_wrappers: &[SinkConnectorWrapper],
source_wrappers: &[SourceConnectorWrapper],
) -> RuntimeContext {
RuntimeContext {
sinks: SinkManager::new(map_sinks(config, sink_wrappers)),
sources: SourceManager::new(map_sources(config, source_wrappers)),
- api_key: config.http.api_key.clone(),
+ api_key: config.http.api_key.to_owned(),
}
}
-fn map_sinks(config: &RuntimeConfig, sink_wrappers: &[SinkConnectorWrapper])
-> Vec<SinkDetails> {
+fn map_sinks(
+ config: &ConnectorsConfig,
+ sink_wrappers: &[SinkConnectorWrapper],
+) -> Vec<SinkDetails> {
let mut sinks = vec![];
for sink_wrapper in sink_wrappers.iter() {
for sink_plugin in sink_wrapper.plugins.iter() {
@@ -73,7 +76,7 @@ fn map_sinks(config: &RuntimeConfig, sink_wrappers:
&[SinkConnectorWrapper]) ->
}
fn map_sources(
- config: &RuntimeConfig,
+ config: &ConnectorsConfig,
source_wrappers: &[SourceConnectorWrapper],
) -> Vec<SourceDetails> {
let mut sources = vec![];
diff --git a/core/connectors/runtime/src/main.rs
b/core/connectors/runtime/src/main.rs
index 1baa40eb..319b96b9 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -16,7 +16,7 @@
* under the License.
*/
-use configs::{ConfigFormat, RuntimeConfig};
+use configs::{ConfigFormat, ConnectorsConfig};
use dlopen2::wrapper::{Container, WrapperApi};
use dotenvy::dotenv;
use error::RuntimeError;
@@ -112,7 +112,7 @@ async fn main() -> Result<(), RuntimeError> {
env::var("IGGY_CONNECTORS_CONFIG_PATH").unwrap_or_else(|_|
DEFAULT_CONFIG_PATH.to_string());
info!("Starting Iggy Connectors Runtime, loading configuration from:
{config_path}...");
- let config: RuntimeConfig = RuntimeConfig::config_provider(config_path)
+ let config: ConnectorsConfig =
ConnectorsConfig::config_provider(config_path)
.load_config()
.await
.expect("Failed to load configuration");
diff --git a/core/connectors/runtime/src/stream.rs
b/core/connectors/runtime/src/stream.rs
index 3f62cf22..51943ae1 100644
--- a/core/connectors/runtime/src/stream.rs
+++ b/core/connectors/runtime/src/stream.rs
@@ -38,24 +38,17 @@ async fn create_client(config: &IggyConfig) ->
Result<IggyClient, RuntimeError>
let password = config.password.to_owned();
let token = config.token.to_owned();
- let connection_string = if let Some(token) = token {
- if token.is_empty() {
- error!("Iggy token cannot be empty (if username and password are
not provided)");
- return Err(RuntimeError::MissingIggyCredentials);
- }
-
+ let connection_string = if !token.is_empty() {
let redacted_token = token.chars().take(3).collect::<String>();
info!("Using token: {redacted_token}*** for Iggy authentication");
format!("iggy://{token}@{address}")
} else {
info!("Using username and password for Iggy authentication");
- let username = username.ok_or(RuntimeError::MissingIggyCredentials)?;
if username.is_empty() {
- error!("Iggy password cannot be empty (if token is not provided)");
+ error!("Iggy username cannot be empty (if token is not provided)");
return Err(RuntimeError::MissingIggyCredentials);
}
- let password = password.ok_or(RuntimeError::MissingIggyCredentials)?;
if password.is_empty() {
error!("Iggy password cannot be empty (if token is not provided)");
return Err(RuntimeError::MissingIggyCredentials);
diff --git a/core/integration/src/test_server.rs
b/core/integration/src/test_server.rs
index f09c9369..372a4413 100644
--- a/core/integration/src/test_server.rs
+++ b/core/integration/src/test_server.rs
@@ -325,7 +325,7 @@ impl TestServer {
sleep(Duration::from_millis(SLEEP_INTERVAL_MS));
continue;
}
- match ServerConfig::config_provider(config_path.clone())
+ match ServerConfig::file_config_provider(config_path.clone())
.load_config()
.await
{
diff --git a/core/integration/tests/config_provider/mod.rs
b/core/integration/tests/config_provider/mod.rs
index 79de9689..1252c2c5 100644
--- a/core/integration/tests/config_provider/mod.rs
+++ b/core/integration/tests/config_provider/mod.rs
@@ -324,5 +324,5 @@ async fn validate_cluster_contiguous_array_override() {
fn get_file_config_provider() -> FileConfigProvider<ServerEnvProvider> {
let config_path = get_root_path().join("../configs/server.toml");
- ServerConfig::config_provider(config_path.as_path().display().to_string())
+
ServerConfig::file_config_provider(config_path.as_path().display().to_string())
}
diff --git a/core/server/src/configs/server.rs
b/core/server/src/configs/server.rs
index f0d347ea..c51e6e02 100644
--- a/core/server/src/configs/server.rs
+++ b/core/server/src/configs/server.rs
@@ -195,9 +195,9 @@ pub fn resolve(config_provider_type: &str) ->
Result<ConfigProviderKind, ConfigE
DEFAULT_CONFIG_PROVIDER => {
let path =
env::var("IGGY_CONFIG_PATH").unwrap_or_else(|_|
DEFAULT_CONFIG_PATH.to_string());
- Ok(ConfigProviderKind::File(ServerConfig::config_provider(
- path,
- )))
+ Ok(ConfigProviderKind::File(
+ ServerConfig::file_config_provider(path),
+ ))
}
_ => Err(ConfigError::InvalidConfigurationProvider {
provider_type: config_provider_type.to_string(),
@@ -221,7 +221,7 @@ impl ConfigProviderKind {
}
impl ServerConfig {
- pub async fn load(config_provider: ConfigProviderKind) ->
Result<ServerConfig, ConfigError> {
+ pub async fn load(config_provider: ConfigProviderKind) -> Result<Self,
ConfigError> {
let server_config = config_provider
.load_config()
.await
@@ -234,7 +234,7 @@ impl ServerConfig {
Ok(server_config)
}
- pub fn config_provider(path: String) ->
FileConfigProvider<ServerEnvProvider> {
+ pub fn file_config_provider(path: String) ->
FileConfigProvider<ServerEnvProvider> {
let default_config =
Toml::string(include_str!("../../../configs/server.toml"));
FileConfigProvider::new(
path,
