You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ma...@apache.org on 2025/11/07 12:34:50 UTC
(iggy) branch connectors_config updated (67f028b42 -> 280145c56)
This is an automated email from the ASF dual-hosted git repository.
maciej pushed a change to branch connectors_config
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 67f028b42 refactor(connectors): split connectors configs from the runtime config
new 280145c56 refactor(connectors): split connectors configs from the runtime config
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (67f028b42)
\
N -- N -- N refs/heads/connectors_config (280145c56)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/connectors/runtime/src/main.rs | 7 +++----
core/connectors/runtime/src/sink.rs | 2 +-
core/connectors/runtime/src/source.rs | 4 +---
3 files changed, 5 insertions(+), 8 deletions(-)
(iggy) 01/01: refactor(connectors): split connectors configs from the runtime config
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
maciej pushed a commit to branch connectors_config
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 280145c561336834fc23669cd122ef9be9539fc2
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Fri Nov 7 12:32:54 2025 +0100
refactor(connectors): split connectors configs from the runtime config
---
core/connectors/runtime/config.toml | 6 +
.../runtime/{ => example_config}/config.toml | 6 +
.../sinks/quickwit.toml} | 92 +---------
.../sinks/stdout.toml} | 38 ++--
.../sources/random.toml} | 41 ++---
core/connectors/runtime/src/configs.rs | 194 ++++++++++++++++++++-
core/connectors/runtime/src/context.rs | 19 +-
core/connectors/runtime/src/error.rs | 2 +
core/connectors/runtime/src/main.rs | 24 ++-
core/connectors/runtime/src/sink.rs | 7 +-
core/connectors/runtime/src/source.rs | 9 +-
11 files changed, 280 insertions(+), 158 deletions(-)
diff --git a/core/connectors/runtime/config.toml b/core/connectors/runtime/config.toml
index 848010341..875455d3d 100644
--- a/core/connectors/runtime/config.toml
+++ b/core/connectors/runtime/config.toml
@@ -42,3 +42,9 @@ token = "" # Personal Access Token (PAT) can be used instead of username and pas
[state]
path = "local_state"
+
+[sinks]
+config_dir = ""
+
+[sources]
+config_dir = ""
diff --git a/core/connectors/runtime/config.toml b/core/connectors/runtime/example_config/config.toml
similarity index 91%
copy from core/connectors/runtime/config.toml
copy to core/connectors/runtime/example_config/config.toml
index 848010341..139facb62 100644
--- a/core/connectors/runtime/config.toml
+++ b/core/connectors/runtime/example_config/config.toml
@@ -42,3 +42,9 @@ token = "" # Personal Access Token (PAT) can be used instead of username and pas
[state]
path = "local_state"
+
+[sinks]
+config_dir = "core/connectors/runtime/example_config/sinks"
+
+[sources]
+config_dir = "core/connectors/runtime/example_config/sources"
diff --git a/core/connectors/runtime/example_config.toml b/core/connectors/runtime/example_config/sinks/quickwit.toml
similarity index 51%
rename from core/connectors/runtime/example_config.toml
rename to core/connectors/runtime/example_config/sinks/quickwit.toml
index ae3b0fa7e..3bd383920 100644
--- a/core/connectors/runtime/example_config.toml
+++ b/core/connectors/runtime/example_config/sinks/quickwit.toml
@@ -15,90 +15,12 @@
# specific language governing permissions and limitations
# under the License.
-[http] # Optional HTTP API configuration
-enabled = true
-address = "127.0.0.1:8081"
-api_key = "" # Optional API key for authentication to be passed as `api-key` header
-
-[http.cors] # Optional CORS configuration for HTTP API
-enabled = false
-allowed_methods = ["GET", "POST", "PUT", "DELETE"]
-allowed_origins = ["*"]
-allowed_headers = ["content-type"]
-exposed_headers = [""]
-allow_credentials = false
-allow_private_network = false
-
-[http.tls] # Optional TLS configuration for HTTP API
-enabled = false
-cert_file = "core/certs/iggy_cert.pem"
-key_file = "core/certs/iggy_key.pem"
-
-[iggy]
-address = "localhost:8090"
-username = "iggy"
-password = "iggy"
-token = "" # Personal Access Token (PAT) can be used instead of username and password
-
-[state]
-path = "local_state"
-
-[sinks.stdout]
-enabled = true
-name = "Stdout sink"
-path = "target/release/libiggy_connector_stdout_sink"
-
-[[sinks.stdout.streams]]
-stream = "example_stream"
-topics = ["example_topic"]
-schema = "json"
-batch_length = 100
-poll_interval = "5ms"
-consumer_group = "stdout_sink_connector"
-
-[sinks.stdout.config]
-print_payload = false
-
-[sinks.stdout.transforms.add_fields]
-enabled = true
-
-[[sinks.stdout.transforms.add_fields.fields]]
-key = "message"
-value.static = "hello"
-
-[sources.random]
-enabled = true
-name = "Random source"
-path = "target/release/libiggy_connector_random_source"
-config_format = "json"
-
-[[sources.random.streams]]
-stream = "example_stream"
-topic = "example_topic"
-schema = "json"
-batch_length = 1000
-linger_time = "5ms"
-
-[sources.random.config]
-interval = "100ms"
-max_count = 1000
-messages_range = [10, 50]
-payload_size = 200
-
-[sources.random.transforms.add_fields]
-enabled = true
-
-[[sources.random.transforms.add_fields.fields]]
-key = "test_field"
-value.static = "hello!"
-
-[sinks.quickwit]
enabled = true
name = "Quickwit sink 1"
path = "target/release/libiggy_connector_quickwit_sink"
config_format = "yaml"
-[[sinks.quickwit.streams]]
+[[streams]]
stream = "qw"
topics = ["records"]
schema = "json"
@@ -106,26 +28,26 @@ batch_length = 1000
poll_interval = "5ms"
consumer_group = "qw_sink_connector"
-[sinks.quickwit.transforms.add_fields]
+[transforms.add_fields]
enabled = true
-[[sinks.quickwit.transforms.add_fields.fields]]
+[[transforms.add_fields.fields]]
key = "service_name"
value.static = "qw_connector"
-[[sinks.quickwit.transforms.add_fields.fields]]
+[[transforms.add_fields.fields]]
key = "timestamp"
value.computed = "timestamp_millis"
-[[sinks.quickwit.transforms.add_fields.fields]]
+[[transforms.add_fields.fields]]
key = "random_id"
value.computed = "uuid_v7"
-[sinks.quickwit.transforms.delete_fields]
+[transforms.delete_fields]
enabled = true
fields = ["email", "created_at"]
-[sinks.quickwit.config]
+[config]
url = "http://localhost:7280"
index = """
version: 0.9
diff --git a/core/connectors/runtime/config.toml b/core/connectors/runtime/example_config/sinks/stdout.toml
similarity index 52%
copy from core/connectors/runtime/config.toml
copy to core/connectors/runtime/example_config/sinks/stdout.toml
index 848010341..e2581712f 100644
--- a/core/connectors/runtime/config.toml
+++ b/core/connectors/runtime/example_config/sinks/stdout.toml
@@ -15,30 +15,24 @@
# specific language governing permissions and limitations
# under the License.
-[http] # Optional HTTP API configuration
enabled = true
-address = "127.0.0.1:8081"
-api_key = "" # Optional API key for authentication to be passed as `api-key` header
+name = "Stdout sink"
+path = "target/release/libiggy_connector_stdout_sink"
-[http.cors] # Optional CORS configuration for HTTP API
-enabled = false
-allowed_methods = ["GET", "POST", "PUT", "DELETE"]
-allowed_origins = ["*"]
-allowed_headers = ["content-type"]
-exposed_headers = [""]
-allow_credentials = false
-allow_private_network = false
+[[streams]]
+stream = "example_stream"
+topics = ["example_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "stdout_sink_connector"
-[http.tls] # Optional TLS configuration for HTTP API
-enabled = false
-cert_file = "core/certs/iggy_cert.pem"
-key_file = "core/certs/iggy_key.pem"
+[config]
+print_payload = false
-[iggy]
-address = "localhost:8090"
-username = "iggy"
-password = "iggy"
-token = "" # Personal Access Token (PAT) can be used instead of username and password
+[transforms.add_fields]
+enabled = true
-[state]
-path = "local_state"
+[[transforms.add_fields.fields]]
+key = "message"
+value.static = "hello"
diff --git a/core/connectors/runtime/config.toml b/core/connectors/runtime/example_config/sources/random.toml
similarity index 52%
copy from core/connectors/runtime/config.toml
copy to core/connectors/runtime/example_config/sources/random.toml
index 848010341..ecce6aa6a 100644
--- a/core/connectors/runtime/config.toml
+++ b/core/connectors/runtime/example_config/sources/random.toml
@@ -15,30 +15,27 @@
# specific language governing permissions and limitations
# under the License.
-[http] # Optional HTTP API configuration
enabled = true
-address = "127.0.0.1:8081"
-api_key = "" # Optional API key for authentication to be passed as `api-key` header
+name = "Random source"
+path = "target/release/libiggy_connector_random_source"
+config_format = "json"
-[http.cors] # Optional CORS configuration for HTTP API
-enabled = false
-allowed_methods = ["GET", "POST", "PUT", "DELETE"]
-allowed_origins = ["*"]
-allowed_headers = ["content-type"]
-exposed_headers = [""]
-allow_credentials = false
-allow_private_network = false
+[[streams]]
+stream = "example_stream"
+topic = "example_topic"
+schema = "json"
+batch_length = 1000
+linger_time = "5ms"
-[http.tls] # Optional TLS configuration for HTTP API
-enabled = false
-cert_file = "core/certs/iggy_cert.pem"
-key_file = "core/certs/iggy_key.pem"
+[config]
+interval = "100ms"
+max_count = 1000
+messages_range = [10, 50]
+payload_size = 200
-[iggy]
-address = "localhost:8090"
-username = "iggy"
-password = "iggy"
-token = "" # Personal Access Token (PAT) can be used instead of username and password
+[transforms.add_fields]
+enabled = true
-[state]
-path = "local_state"
+[[transforms.add_fields.fields]]
+key = "test_field"
+value.static = "hello!"
diff --git a/core/connectors/runtime/src/configs.rs b/core/connectors/runtime/src/configs.rs
index 32b71707f..e91654b49 100644
--- a/core/connectors/runtime/src/configs.rs
+++ b/core/connectors/runtime/src/configs.rs
@@ -17,17 +17,19 @@
*/
use crate::api::config::HttpConfig;
+use crate::error::RuntimeError;
use figment::{
Metadata, Profile, Provider,
providers::{Format, Toml},
value::Dict,
};
use iggy::prelude::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
-use iggy_common::{CustomEnvProvider, FileConfigProvider};
+use iggy_common::{ConfigProvider, CustomEnvProvider, FileConfigProvider};
use iggy_connector_sdk::{Schema, transforms::TransformType};
use serde::{Deserialize, Serialize};
use std::{collections::HashMap, fmt::Formatter};
use strum::Display;
+use tracing::{debug, info, warn};
#[derive(
Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Deserialize, Serialize, Display,
@@ -47,11 +49,23 @@ pub enum ConfigFormat {
#[derive(Debug, Default, Clone, Deserialize, Serialize)]
#[serde(default)]
-pub struct ConnectorsConfig {
+pub struct SinksCommonConfig {
+ pub config_dir: String,
+}
+
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[serde(default)]
+pub struct SourcesCommonConfig {
+ pub config_dir: String,
+}
+
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[serde(default)]
+pub struct ConnectorsRuntimeConfig {
pub http: HttpConfig,
pub iggy: IggyConfig,
- pub sinks: HashMap<String, SinkConfig>,
- pub sources: HashMap<String, SourceConfig>,
+ pub sinks: SinksCommonConfig,
+ pub sources: SourcesCommonConfig,
pub state: StateConfig,
}
@@ -82,6 +96,12 @@ pub struct StreamProducerConfig {
pub linger_time: Option<String>,
}
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[serde(default)]
+pub struct SinksConfig {
+ pub configs: HashMap<String, SinkConfig>,
+}
+
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct SinkConfig {
pub enabled: bool,
@@ -93,6 +113,12 @@ pub struct SinkConfig {
pub config: Option<serde_json::Value>,
}
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[serde(default)]
+pub struct SourcesConfig {
+ pub configs: HashMap<String, SourceConfig>,
+}
+
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct SourceConfig {
pub enabled: bool,
@@ -126,7 +152,7 @@ impl std::fmt::Display for StateConfig {
}
}
-impl std::fmt::Display for ConnectorsConfig {
+impl std::fmt::Display for ConnectorsRuntimeConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
@@ -250,7 +276,7 @@ impl Default for IggyConfig {
}
}
-impl ConnectorsConfig {
+impl ConnectorsRuntimeConfig {
pub fn config_provider(path: String) -> FileConfigProvider<ConnectorsEnvProvider> {
let default_config = Toml::string(include_str!("../../../connectors/runtime/config.toml"));
FileConfigProvider::new(
@@ -264,7 +290,7 @@ impl ConnectorsConfig {
#[derive(Debug, Clone)]
pub struct ConnectorsEnvProvider {
- provider: CustomEnvProvider<ConnectorsConfig>,
+ provider: CustomEnvProvider<ConnectorsRuntimeConfig>,
}
impl Default for ConnectorsEnvProvider {
@@ -286,3 +312,157 @@ impl Provider for ConnectorsEnvProvider {
})
}
}
+
+impl SinksConfig {
+ pub async fn load_configs(configuration: &SinksCommonConfig) -> Result<Self, RuntimeError> {
+ if configuration.config_dir.is_empty() {
+ info!("Sinks configuration directory not provided, skipping initialization");
+ return Ok(SinksConfig::default());
+ }
+ if !std::fs::exists(&configuration.config_dir)? {
+ warn!(
+ "Sinks configuration directory does not exist: {}",
+ configuration.config_dir
+ );
+ return Ok(SinksConfig::default());
+ }
+ let mut configs = HashMap::new();
+ info!(
+ "Loading sinks configuration from: {}",
+ configuration.config_dir
+ );
+ let entries = std::fs::read_dir(&configuration.config_dir)?;
+ for entry in entries.flatten() {
+ let path = entry.path();
+ if path.is_file() {
+ debug!("Loading sink configuration from: {:?}", path);
+ let sink_name = path.file_stem().unwrap().to_string_lossy().to_string();
+ let connector_config =
+ SinkConfig::config_provider(path.to_str().unwrap().to_string(), &sink_name)
+ .load_config()
+ .await
+ .expect("Failed to load sink configuration");
+ configs.insert(sink_name, connector_config);
+ }
+ }
+ Ok(Self { configs })
+ }
+}
+
+impl SinkConfig {
+ pub fn config_provider(path: String, sink_name: &str) -> FileConfigProvider<SinkEnvProvider> {
+ FileConfigProvider::new(path, SinkEnvProvider::with_sink_name(sink_name), true, None)
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct SinkEnvProvider {
+ sink_name: String,
+ provider: CustomEnvProvider<SinkConfig>,
+}
+
+impl SinkEnvProvider {
+ fn with_sink_name(sink_name: &str) -> Self {
+ let prefix = format!("IGGY_SINK_{}_", sink_name.to_uppercase());
+ Self {
+ sink_name: sink_name.to_owned(),
+ provider: CustomEnvProvider::new(&prefix, &[]),
+ }
+ }
+}
+
+impl Provider for SinkEnvProvider {
+ fn metadata(&self) -> Metadata {
+ Metadata::named(format!("iggy-sink-config-{}", self.sink_name))
+ }
+
+ fn data(&self) -> Result<figment::value::Map<Profile, Dict>, figment::Error> {
+ self.provider.deserialize().map_err(|_| {
+ figment::Error::from(format!(
+ "Cannot deserialize environment variables for sink config {}",
+ self.sink_name
+ ))
+ })
+ }
+}
+
+impl SourcesConfig {
+ pub async fn load_configs(configuration: &SourcesCommonConfig) -> Result<Self, RuntimeError> {
+ if configuration.config_dir.is_empty() {
+ info!("Sources configuration directory not provided, skipping initialization");
+ return Ok(SourcesConfig::default());
+ }
+ if !std::fs::exists(&configuration.config_dir)? {
+ warn!(
+ "Sources configuration directory does not exist: {}",
+ configuration.config_dir
+ );
+ return Ok(SourcesConfig::default());
+ }
+ let mut configs = HashMap::new();
+ info!(
+ "Loading sources configuration from: {}",
+ configuration.config_dir
+ );
+ let entries = std::fs::read_dir(&configuration.config_dir)?;
+ for entry in entries.flatten() {
+ let path = entry.path();
+ if path.is_file() {
+ debug!("Loading source configuration from: {:?}", path);
+ let source_name = path.file_stem().unwrap().to_string_lossy().to_string();
+ let connector_config =
+ SourceConfig::config_provider(path.to_str().unwrap().to_string(), &source_name)
+ .load_config()
+ .await
+ .expect("Failed to load source configuration");
+ configs.insert(source_name, connector_config);
+ }
+ }
+ Ok(Self { configs })
+ }
+}
+
+impl SourceConfig {
+ pub fn config_provider(
+ path: String,
+ source_name: &str,
+ ) -> FileConfigProvider<SourceEnvProvider> {
+ FileConfigProvider::new(
+ path,
+ SourceEnvProvider::with_source_name(source_name),
+ true,
+ None,
+ )
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct SourceEnvProvider {
+ source_name: String,
+ provider: CustomEnvProvider<SourceConfig>,
+}
+
+impl SourceEnvProvider {
+ fn with_source_name(source_name: &str) -> Self {
+ let prefix = format!("IGGY_SOURCE_{}_", source_name.to_uppercase());
+ Self {
+ source_name: source_name.to_owned(),
+ provider: CustomEnvProvider::new(&prefix, &[]),
+ }
+ }
+}
+
+impl Provider for SourceEnvProvider {
+ fn metadata(&self) -> Metadata {
+ Metadata::named(format!("iggy-source-config-{}", self.source_name))
+ }
+
+ fn data(&self) -> Result<figment::value::Map<Profile, Dict>, figment::Error> {
+ self.provider.deserialize().map_err(|_| {
+ figment::Error::from(format!(
+ "Cannot deserialize environment variables for source config {}",
+ self.source_name
+ ))
+ })
+ }
+}
diff --git a/core/connectors/runtime/src/context.rs b/core/connectors/runtime/src/context.rs
index 1a9d6a2f9..4e15b5826 100644
--- a/core/connectors/runtime/src/context.rs
+++ b/core/connectors/runtime/src/context.rs
@@ -16,9 +16,10 @@
* under the License.
*/
+use crate::configs::{SinksConfig, SourcesConfig};
use crate::{
SinkConnectorWrapper, SourceConnectorWrapper,
- configs::ConnectorsConfig,
+ configs::ConnectorsRuntimeConfig,
manager::{
sink::{SinkDetails, SinkInfo, SinkManager},
source::{SourceDetails, SourceInfo, SourceManager},
@@ -33,25 +34,27 @@ pub struct RuntimeContext {
}
pub fn init(
- config: &ConnectorsConfig,
+ config: &ConnectorsRuntimeConfig,
+ sinks_config: &SinksConfig,
+ sources_config: &SourcesConfig,
sink_wrappers: &[SinkConnectorWrapper],
source_wrappers: &[SourceConnectorWrapper],
) -> RuntimeContext {
RuntimeContext {
- sinks: SinkManager::new(map_sinks(config, sink_wrappers)),
- sources: SourceManager::new(map_sources(config, source_wrappers)),
+ sinks: SinkManager::new(map_sinks(sinks_config, sink_wrappers)),
+ sources: SourceManager::new(map_sources(sources_config, source_wrappers)),
api_key: config.http.api_key.to_owned(),
}
}
fn map_sinks(
- config: &ConnectorsConfig,
+ sinks_config: &SinksConfig,
sink_wrappers: &[SinkConnectorWrapper],
) -> Vec<SinkDetails> {
let mut sinks = vec![];
for sink_wrapper in sink_wrappers.iter() {
for sink_plugin in sink_wrapper.plugins.iter() {
- let Some(sink_config) = config.sinks.get(&sink_plugin.key) else {
+ let Some(sink_config) = sinks_config.configs.get(&sink_plugin.key) else {
error!("Missing sink config for: {}", sink_plugin.key);
continue;
};
@@ -76,13 +79,13 @@ fn map_sinks(
}
fn map_sources(
- config: &ConnectorsConfig,
+ sources_config: &SourcesConfig,
source_wrappers: &[SourceConnectorWrapper],
) -> Vec<SourceDetails> {
let mut sources = vec![];
for source_wrapper in source_wrappers.iter() {
for source_plugin in source_wrapper.plugins.iter() {
- let Some(source_config) = config.sources.get(&source_plugin.key) else {
+ let Some(source_config) = sources_config.configs.get(&source_plugin.key) else {
error!("Missing source config for: {}", source_plugin.key);
continue;
};
diff --git a/core/connectors/runtime/src/error.rs b/core/connectors/runtime/src/error.rs
index c622feabe..227a4c5ec 100644
--- a/core/connectors/runtime/src/error.rs
+++ b/core/connectors/runtime/src/error.rs
@@ -46,6 +46,8 @@ pub enum RuntimeError {
SourceNotFound(String),
#[error("Cannot convert configuration")]
CannotConvertConfiguration,
+ #[error("IO operation failed with error: {0:?}")]
+ IoError(#[from] std::io::Error),
}
impl RuntimeError {
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
index 319b96b9c..b4311f652 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
@@ -16,7 +17,8 @@
* under the License.
*/
-use configs::{ConfigFormat, ConnectorsConfig};
+use crate::configs::{SinksConfig, SourcesConfig};
+use configs::{ConfigFormat, ConnectorsRuntimeConfig};
use dlopen2::wrapper::{Container, WrapperApi};
use dotenvy::dotenv;
use error::RuntimeError;
@@ -112,7 +114,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: ConnectorsConfig = ConnectorsConfig::config_provider(config_path)
+ let config: ConnectorsRuntimeConfig = ConnectorsRuntimeConfig::config_provider(config_path)
.load_config()
.await
.expect("Failed to load configuration");
@@ -122,13 +124,17 @@ async fn main() -> Result<(), RuntimeError> {
info!("State will be stored in: {}", config.state.path);
let iggy_clients = stream::init(config.iggy.clone()).await?;
+
+ let sources_config = SourcesConfig::load_configs(&config.sources).await?;
let sources = source::init(
- config.sources.clone(),
+ sources_config.clone(),
&iggy_clients.producer,
&config.state.path,
)
.await?;
- let sinks = sink::init(config.sinks.clone(), &iggy_clients.consumer).await?;
+
+ let sinks_config = SinksConfig::load_configs(&config.sinks).await?;
+ let sinks = sink::init(sinks_config.clone(), &iggy_clients.consumer).await?;
let mut sink_wrappers = vec![];
let mut sink_with_plugins = HashMap::new();
@@ -164,7 +170,13 @@ async fn main() -> Result<(), RuntimeError> {
);
}
- let context = context::init(&config, &sink_wrappers, &source_wrappers);
+ let context = context::init(
+ &config,
+ &sinks_config,
+ &sources_config,
+ &sink_wrappers,
+ &source_wrappers,
+ );
let context = Arc::new(context);
api::init(&config.http, context).await;
diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs
index ee0187966..f928c94d1 100644
--- a/core/connectors/runtime/src/sink.rs
+++ b/core/connectors/runtime/src/sink.rs
@@ -16,9 +16,10 @@
* under the License.
*/
+use crate::configs::SinksConfig;
use crate::{
PLUGIN_ID, RuntimeError, SinkApi, SinkConnector, SinkConnectorConsumer, SinkConnectorPlugin,
- SinkConnectorWrapper, configs::SinkConfig, resolve_plugin_path, transform,
+ SinkConnectorWrapper, resolve_plugin_path, transform,
};
use dlopen2::wrapper::Container;
use futures::StreamExt;
@@ -39,11 +40,11 @@ use std::{
use tracing::{error, info, warn};
pub async fn init(
- sink_configs: HashMap<String, SinkConfig>,
+ sink_configs: SinksConfig,
iggy_client: &IggyClient,
) -> Result<HashMap<String, SinkConnector>, RuntimeError> {
let mut sink_connectors: HashMap<String, SinkConnector> = HashMap::new();
- for (key, config) in sink_configs {
+ for (key, config) in sink_configs.configs {
let name = config.name;
if !config.enabled {
warn!("Sink: {name} is disabled ({key})");
diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs
index 346fb150c..e62da06a8 100644
--- a/core/connectors/runtime/src/source.rs
+++ b/core/connectors/runtime/src/source.rs
@@ -34,11 +34,10 @@ use std::{
};
use tracing::{debug, error, info, warn};
+use crate::configs::SourcesConfig;
use crate::{
PLUGIN_ID, RuntimeError, SourceApi, SourceConnector, SourceConnectorPlugin,
- SourceConnectorProducer, SourceConnectorWrapper,
- configs::SourceConfig,
- resolve_plugin_path,
+ SourceConnectorProducer, SourceConnectorWrapper, resolve_plugin_path,
state::{FileStateProvider, StateProvider, StateStorage},
transform,
};
@@ -46,12 +45,12 @@ use crate::{
pub static SOURCE_SENDERS: Lazy<DashMap<u32, Sender<ProducedMessages>>> = Lazy::new(DashMap::new);
pub async fn init(
- source_configs: HashMap<String, SourceConfig>,
+ source_configs: SourcesConfig,
iggy_client: &IggyClient,
state_path: &str,
) -> Result<HashMap<String, SourceConnector>, RuntimeError> {
let mut source_connectors: HashMap<String, SourceConnector> = HashMap::new();
- for (key, config) in source_configs {
+ for (key, config) in source_configs.configs {
let name = config.name;
if !config.enabled {
warn!("Source: {name} is disabled ({key})");