You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "mmodzelewski (via GitHub)" <gi...@apache.org> on 2025/11/07 12:35:06 UTC
[PR] refactor(connectors): split connectors configs from the runtime config [iggy]
mmodzelewski opened a new pull request, #2317:
URL: https://github.com/apache/iggy/pull/2317
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(connectors): split connectors configs from the runtime config [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz merged PR #2317:
URL: https://github.com/apache/iggy/pull/2317
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(connectors): split connectors configs from the runtime config [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #2317:
URL: https://github.com/apache/iggy/pull/2317#discussion_r2503961841
##########
core/connectors/runtime/src/configs.rs:
##########
@@ -286,3 +312,160 @@ 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();
Review Comment:
Maybe let's use `expect()` or `let Ok(...) = {}` construct, to avoid direct unwrap in the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(connectors): split connectors configs from the runtime config [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #2317:
URL: https://github.com/apache/iggy/pull/2317#discussion_r2503966560
##########
core/connectors/runtime/src/configs.rs:
##########
@@ -286,3 +312,160 @@ 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_CONNECTORS_SINKS_{}_", 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-connectors-sinks-{}-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();
Review Comment:
Same remarks as for sink config
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(connectors): split connectors configs from the runtime config [iggy]
Posted by "mmodzelewski (via GitHub)" <gi...@apache.org>.
mmodzelewski commented on PR #2317:
URL: https://github.com/apache/iggy/pull/2317#issuecomment-3504619336
> Looks good overall! Just wondering, what if someone provides the same path for both, sink and source configs directory? Can we ensure, that it will also work?
It doesn't work with the current implementation. In such a case, each config is imported as both source and sink. In theory, if the connector binary implements both traits, then it should work, but the configuration would be shared. I think this is an unusual case.
If you think it should work under one dir, then probably I'll need to introduce some discriminator value to the config file itself or rely on a configuration file name convention, e.g. `quickwit.sink.toml`, `random.source.toml`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(connectors): split connectors configs from the runtime config [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #2317:
URL: https://github.com/apache/iggy/pull/2317#discussion_r2506911359
##########
core/connectors/runtime/src/configs.rs:
##########
@@ -286,3 +326,114 @@ impl Provider for ConnectorsEnvProvider {
})
}
}
+
+impl ConnectorsConfig {
+ pub async fn load_configs(
+ configuration: &ConnectorsCommonConfig,
+ ) -> Result<Self, RuntimeError> {
+ if configuration.config_dir.is_empty() {
+ info!("Connectors configuration directory not provided, skipping initialization");
+ return Ok(ConnectorsConfig::default());
+ }
+ if !std::fs::exists(&configuration.config_dir)? {
+ warn!(
+ "Connectors configuration directory does not exist: {}",
+ configuration.config_dir
+ );
+ return Ok(ConnectorsConfig::default());
+ }
+ let mut configs = HashMap::new();
+ info!(
+ "Loading connectors 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 connector configuration from: {:?}", path);
+ let connector_name = path
+ .file_stem()
+ .expect("Failed to get connector configuration name")
+ .to_string_lossy()
+ .to_string();
+ let connector_config = ConnectorConfig::config_provider(
+ path.to_str()
+ .expect("Failed to convert connector configuration path to string")
+ .to_string(),
+ &connector_name,
+ )
+ .load_config()
+ .await
+ .expect("Failed to load connector configuration");
+ configs.insert(connector_name, connector_config);
+ }
+ }
+ Ok(Self { configs })
+ }
+
+ pub fn sinks(&self) -> HashMap<String, SinkConfig> {
+ self.configs
+ .iter()
+ .filter_map(|(k, v)| match v {
+ ConnectorConfig::Sink(config) => Some((k.to_owned(), config.clone())),
+ _ => None,
+ })
+ .collect()
+ }
+
+ pub fn sources(&self) -> HashMap<String, SourceConfig> {
+ self.configs
+ .iter()
+ .filter_map(|(k, v)| match v {
+ ConnectorConfig::Source(config) => Some((k.to_owned(), config.clone())),
+ _ => None,
+ })
+ .collect()
+ }
+}
+
+impl ConnectorConfig {
+ pub fn config_provider(
+ path: String,
+ connector_name: &str,
+ ) -> FileConfigProvider<ConnectorEnvProvider> {
+ FileConfigProvider::new(
+ path,
+ ConnectorEnvProvider::with_connector_name(connector_name),
+ true,
Review Comment:
Let's set this to false by default, not to print the connector config which may contain some sensitive fields.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(connectors): split connectors configs from the runtime config [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on PR #2317:
URL: https://github.com/apache/iggy/pull/2317#issuecomment-3503009538
Looks good overall! Just wondering, what if someone provides the same path for both, sink and source configs directory? Can we ensure, that it will also work?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(connectors): split connectors configs from the runtime config [iggy]
Posted by "mmodzelewski (via GitHub)" <gi...@apache.org>.
mmodzelewski commented on code in PR #2317:
URL: https://github.com/apache/iggy/pull/2317#discussion_r2505176275
##########
core/connectors/runtime/src/configs.rs:
##########
@@ -286,3 +312,160 @@ 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();
Review Comment:
got it, updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(connectors): split connectors configs from the runtime config [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #2317:
URL: https://github.com/apache/iggy/pull/2317#discussion_r2503963223
##########
core/connectors/runtime/src/configs.rs:
##########
@@ -286,3 +312,160 @@ 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)
Review Comment:
Also here regarding the unwrap :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org