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/14 15:26:20 UTC
(iggy) 01/01: feat(connectors): extend runtime API with config management options
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 5bcfd08c0c35566c1269f1d9e23710621b0e8163
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Fri Nov 14 10:51:15 2025 +0100
feat(connectors): extend runtime API with config management options
---
core/connectors/runtime/README.md | 14 +-
.../example_config/connectors/quickwit.toml | 6 +-
.../connectors/{random.toml => random_v0.toml} | 6 +-
.../connectors/{random.toml => random_v1.toml} | 14 +-
.../runtime/example_config/connectors/stdout.toml | 4 +-
core/connectors/runtime/runtime.http | 82 ++++++
core/connectors/runtime/src/api/config.rs | 16 +-
core/connectors/runtime/src/api/models.rs | 26 +-
core/connectors/runtime/src/api/sink.rs | 80 +++++-
core/connectors/runtime/src/api/source.rs | 86 +++++-
core/connectors/runtime/src/configs/connectors.rs | 72 +++--
.../src/configs/connectors/local_provider.rs | 295 ++++++++++++++++++---
core/connectors/runtime/src/context.rs | 18 +-
core/connectors/runtime/src/error.rs | 5 +
core/connectors/runtime/src/main.rs | 7 +-
core/connectors/runtime/src/manager/sink.rs | 35 ++-
core/connectors/runtime/src/manager/source.rs | 35 ++-
core/connectors/runtime/src/sink.rs | 18 +-
core/connectors/runtime/src/source.rs | 21 +-
core/connectors/sdk/README.md | 6 +-
core/connectors/sinks/README.md | 16 +-
core/connectors/sinks/quickwit_sink/README.md | 2 +-
core/connectors/sinks/stdout_sink/README.md | 2 +-
core/connectors/sources/README.md | 14 +-
core/connectors/sources/random_source/README.md | 2 +-
25 files changed, 713 insertions(+), 169 deletions(-)
diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md
index a70c62b9d..4daa7465c 100644
--- a/core/connectors/runtime/README.md
+++ b/core/connectors/runtime/README.md
@@ -71,9 +71,19 @@ Currently, it does expose the following endpoints:
- `GET /health`: health status of the runtime.
- `GET /sinks`: list of sinks.
- `GET /sinks/{key}`: sink details.
-- `GET /sinks/{key}/config`: sink config, including the optional `format` query parameter to specify the config format.
+- `GET /sinks/{key}/configs`: list of configuration versions for the sink.
+- `POST /sinks/{key}/configs`: add a new configuration version for the sink.
+- `GET /sinks/{key}/configs/{version}`: configuration details for a specific version.
+- `GET /sinks/{key}/active_config`: active configuration details.
+- `PUT /sinks/{key}/active_config`: activate a specific configuration version for the sink.
+- `GET /sinks/{key}/plugin_config`: sink plugin config, including the optional `format` query parameter to specify the config format.
- `GET /sinks/{key}/transforms`: sink transforms to be applied to the fields.
- `GET /sources`: list of sources.
- `GET /sources/{key}`: source details.
-- `GET /sources/{key}/config`: source config, including the optional `format` query parameter to specify the config format.
+- `GET /sources/{key}/configs`: list of configuration versions for the source.
+- `POST /sources/{key}/configs`: add a new configuration version for the source.
+- `GET /sources/{key}/configs/{version}`: configuration details for a specific version.
+- `GET /sources/{key}/active_config`: active configuration details.
+- `PUT /sources/{key}/active_config`: activate a specific configuration version for the source.
+- `GET /sources/{key}/plugin_config`: source plugin config, including the optional `format` query parameter to specify the config format.
- `GET /sources/{key}/transforms`: source transforms to be applied to the fields.
diff --git a/core/connectors/runtime/example_config/connectors/quickwit.toml b/core/connectors/runtime/example_config/connectors/quickwit.toml
index 0e30345ac..5b6abd83e 100644
--- a/core/connectors/runtime/example_config/connectors/quickwit.toml
+++ b/core/connectors/runtime/example_config/connectors/quickwit.toml
@@ -16,12 +16,12 @@
# under the License.
type = "sink"
-id = "quickwit"
+key = "quickwit"
enabled = true
version = 0
name = "Quickwit sink 1"
path = "target/release/libiggy_connector_quickwit_sink"
-config_format = "yaml"
+plugin_config_format = "yaml"
[[streams]]
stream = "qw"
@@ -50,7 +50,7 @@ value.computed = "uuid_v7"
enabled = true
fields = ["email", "created_at"]
-[config]
+[plugin_config]
url = "http://localhost:7280"
index = """
version: 0.9
diff --git a/core/connectors/runtime/example_config/connectors/random.toml b/core/connectors/runtime/example_config/connectors/random_v0.toml
similarity index 95%
copy from core/connectors/runtime/example_config/connectors/random.toml
copy to core/connectors/runtime/example_config/connectors/random_v0.toml
index 16cbc1787..323fba91d 100644
--- a/core/connectors/runtime/example_config/connectors/random.toml
+++ b/core/connectors/runtime/example_config/connectors/random_v0.toml
@@ -16,12 +16,12 @@
# under the License.
type = "source"
-id = "random"
+key = "random"
enabled = true
version = 0
name = "Random source"
path = "target/release/libiggy_connector_random_source"
-config_format = "json"
+plugin_config_format = "json"
[[streams]]
stream = "example_stream"
@@ -30,7 +30,7 @@ schema = "json"
batch_length = 1000
linger_time = "5ms"
-[config]
+[plugin_config]
interval = "100ms"
max_count = 1000
messages_range = [10, 50]
diff --git a/core/connectors/runtime/example_config/connectors/random.toml b/core/connectors/runtime/example_config/connectors/random_v1.toml
similarity index 89%
rename from core/connectors/runtime/example_config/connectors/random.toml
rename to core/connectors/runtime/example_config/connectors/random_v1.toml
index 16cbc1787..bca5e9bb9 100644
--- a/core/connectors/runtime/example_config/connectors/random.toml
+++ b/core/connectors/runtime/example_config/connectors/random_v1.toml
@@ -16,12 +16,12 @@
# under the License.
type = "source"
-id = "random"
+key = "random"
enabled = true
-version = 0
+version = 1
name = "Random source"
path = "target/release/libiggy_connector_random_source"
-config_format = "json"
+plugin_config_format = "json"
[[streams]]
stream = "example_stream"
@@ -30,10 +30,10 @@ schema = "json"
batch_length = 1000
linger_time = "5ms"
-[config]
-interval = "100ms"
-max_count = 1000
-messages_range = [10, 50]
+[plugin_config]
+interval = "1000ms"
+max_count = 1_000_000
+messages_range = [1, 5]
payload_size = 200
[transforms.add_fields]
diff --git a/core/connectors/runtime/example_config/connectors/stdout.toml b/core/connectors/runtime/example_config/connectors/stdout.toml
index 8e910f279..c74a5820d 100644
--- a/core/connectors/runtime/example_config/connectors/stdout.toml
+++ b/core/connectors/runtime/example_config/connectors/stdout.toml
@@ -16,7 +16,7 @@
# under the License.
type = "sink"
-id = "stdout"
+key = "stdout"
enabled = true
version = 0
name = "Stdout sink"
@@ -30,7 +30,7 @@ batch_length = 100
poll_interval = "5ms"
consumer_group = "stdout_sink_connector"
-[config]
+[plugin_config]
print_payload = false
[transforms.add_fields]
diff --git a/core/connectors/runtime/runtime.http b/core/connectors/runtime/runtime.http
new file mode 100644
index 000000000..e4e7921d6
--- /dev/null
+++ b/core/connectors/runtime/runtime.http
@@ -0,0 +1,82 @@
+# 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
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+@url = http://localhost:8081
+@sink_key = stdout
+@sink_version = 0
+@source_key = random
+@source_version = 0
+
+###
+GET {{url}}
+
+###
+GET {{url}}/health
+
+###
+GET {{url}}/sinks
+
+###
+GET {{url}}/sinks/{{sink_key}}
+
+###
+GET {{url}}/sinks/{{sink_key}}/configs
+
+###
+#POST {{url}}/sinks/{{sink_key}}/configs
+
+###
+GET {{url}}/sinks/{{sink_key}}/configs/{{sink_version}}
+
+###
+GET {{url}}/sinks/{{sink_key}}/active_config
+
+###
+#PUT {{url}}/sinks/{{sink_key}}/active_config
+
+###
+GET {{url}}/sinks/{{sink_key}}/plugin_config
+
+###
+GET {{url}}/sinks/{{sink_key}}/transforms
+
+###
+GET {{url}}/sources
+
+###
+GET {{url}}/sources/{{source_key}}
+
+###
+GET {{url}}/sources/{{source_key}}/configs
+
+###
+#POST {{url}}/sources/{{source_key}}/configs
+
+###
+GET {{url}}/sources/{{source_key}}/configs/{{source_version}}
+
+###
+GET {{url}}/sources/{{source_key}}/active_config
+
+###
+#PUT {{url}}/sources/{{source_key}}/active_config
+
+###
+GET {{url}}/sources/{{source_key}}/plugin_config
+
+###
+GET {{url}}/sources/{{source_key}}/transforms
diff --git a/core/connectors/runtime/src/api/config.rs b/core/connectors/runtime/src/api/config.rs
index 06505eb5f..1e11791b8 100644
--- a/core/connectors/runtime/src/api/config.rs
+++ b/core/connectors/runtime/src/api/config.rs
@@ -58,30 +58,30 @@ pub struct HttpTlsConfig {
}
pub fn map_connector_config(
- config: &serde_json::Value,
+ plugin_config: &serde_json::Value,
format: ConfigFormat,
) -> Result<(HeaderValue, String), RuntimeError> {
match format {
- ConfigFormat::Json => Ok((JSON_HEADER, config.to_string())),
+ ConfigFormat::Json => Ok((JSON_HEADER, plugin_config.to_string())),
ConfigFormat::Yaml => {
- let config = serde_yaml_ng::to_value(config).map_err(|error| {
+ let plugin_config = serde_yaml_ng::to_value(plugin_config).map_err(|error| {
error!("Failed to convert configuration to YAML. {error}");
RuntimeError::CannotConvertConfiguration
})?;
- let config = serde_yaml_ng::to_string(&config).map_err(|error| {
+ let plugin_config = serde_yaml_ng::to_string(&plugin_config).map_err(|error| {
error!("Failed to serialize YAML configuration. {error}");
RuntimeError::CannotConvertConfiguration
})?;
- Ok((YAML_HEADER, config))
+ Ok((YAML_HEADER, plugin_config))
}
ConfigFormat::Toml => {
- let config = toml::to_string(config).map_err(|error| {
+ let plugin_config = toml::to_string(plugin_config).map_err(|error| {
error!("Failed to convert configuration to TOML. {error}");
RuntimeError::CannotConvertConfiguration
})?;
- Ok((TOML_HEADER, config))
+ Ok((TOML_HEADER, plugin_config))
}
- ConfigFormat::Text => Ok((TEXT_HEADER, config.to_string())),
+ ConfigFormat::Text => Ok((TEXT_HEADER, plugin_config.to_string())),
}
}
diff --git a/core/connectors/runtime/src/api/models.rs b/core/connectors/runtime/src/api/models.rs
index c03af6bb7..f8832dfd1 100644
--- a/core/connectors/runtime/src/api/models.rs
+++ b/core/connectors/runtime/src/api/models.rs
@@ -17,7 +17,9 @@
* under the License.
*/
-use crate::configs::connectors::{ConfigFormat, StreamConsumerConfig, StreamProducerConfig};
+use crate::configs::connectors::{
+ ConfigFormat, SinkConfig, SourceConfig, StreamConsumerConfig, StreamProducerConfig,
+};
use crate::manager::{sink::SinkInfo, source::SourceInfo};
use iggy_connector_sdk::transforms::TransformType;
use serde::{Deserialize, Serialize};
@@ -30,7 +32,7 @@ pub struct SinkInfoResponse {
pub path: String,
pub enabled: bool,
pub running: bool,
- pub config_format: Option<ConfigFormat>,
+ pub plugin_config_format: Option<ConfigFormat>,
}
#[derive(Debug, Serialize, Deserialize)]
@@ -40,6 +42,13 @@ pub struct SinkDetailsResponse {
pub streams: Vec<StreamConsumerConfig>,
}
+#[derive(Debug, Serialize, Deserialize)]
+pub struct SinkConfigResponse {
+ #[serde(flatten)]
+ pub config: SinkConfig,
+ pub active: bool,
+}
+
#[derive(Debug, Serialize, Deserialize)]
pub struct SourceInfoResponse {
pub id: u32,
@@ -48,7 +57,7 @@ pub struct SourceInfoResponse {
pub path: String,
pub enabled: bool,
pub running: bool,
- pub config_format: Option<ConfigFormat>,
+ pub plugin_config_format: Option<ConfigFormat>,
}
#[derive(Debug, Serialize, Deserialize)]
@@ -58,6 +67,13 @@ pub struct SourceDetailsResponse {
pub streams: Vec<StreamProducerConfig>,
}
+#[derive(Debug, Serialize, Deserialize)]
+pub struct SourceConfigResponse {
+ #[serde(flatten)]
+ pub config: SourceConfig,
+ pub active: bool,
+}
+
#[derive(Debug, Serialize, Deserialize)]
pub struct TransformResponse {
pub r#type: TransformType,
@@ -73,7 +89,7 @@ impl From<SinkInfo> for SinkInfoResponse {
path: sink.path,
enabled: sink.enabled,
running: sink.running,
- config_format: sink.config_format,
+ plugin_config_format: sink.plugin_config_format,
}
}
}
@@ -87,7 +103,7 @@ impl From<SourceInfo> for SourceInfoResponse {
path: source.path,
enabled: source.enabled,
running: source.running,
- config_format: source.config_format,
+ plugin_config_format: source.plugin_config_format,
}
}
}
diff --git a/core/connectors/runtime/src/api/sink.rs b/core/connectors/runtime/src/api/sink.rs
index 6ee37882d..3ccd054cb 100644
--- a/core/connectors/runtime/src/api/sink.rs
+++ b/core/connectors/runtime/src/api/sink.rs
@@ -22,6 +22,7 @@ use super::{
error::ApiError,
models::{SinkDetailsResponse, SinkInfoResponse, TransformResponse},
};
+use crate::api::models::SinkConfigResponse;
use crate::configs::connectors::ConfigFormat;
use crate::{context::RuntimeContext, error::RuntimeError};
use axum::{
@@ -38,8 +39,11 @@ pub fn router(state: Arc<RuntimeContext>) -> Router {
Router::new()
.route("/sinks", get(get_sinks))
.route("/sinks/{key}", get(get_sink))
- .route("/sinks/{key}/config", get(get_sink_config))
+ .route("/sinks/{key}/plugin_config", get(get_sink_plugin_config))
.route("/sinks/{key}/transforms", get(get_sink_transforms))
+ .route("/sinks/{key}/configs", get(get_sink_configs))
+ .route("/sinks/{key}/configs/{version}", get(get_sink_config))
+ .route("/sinks/{key}/active_config", get(get_sink_active_config))
.with_state(state)
}
@@ -66,11 +70,11 @@ async fn get_sink(
let sink = sink.lock().await;
Ok(Json(SinkDetailsResponse {
info: sink.info.clone().into(),
- streams: sink.streams.to_vec(),
+ streams: sink.config.streams.to_vec(),
}))
}
-async fn get_sink_config(
+async fn get_sink_plugin_config(
State(context): State<Arc<RuntimeContext>>,
Path(key): Path<String>,
Query(query): Query<GetSinkConfig>,
@@ -79,13 +83,13 @@ async fn get_sink_config(
return Err(ApiError::Error(RuntimeError::SinkNotFound(key)));
};
let sink = sink.lock().await;
- let Some(config) = sink.config.as_ref() else {
+ let Some(config) = sink.config.plugin_config.as_ref() else {
return Ok(StatusCode::NOT_FOUND.into_response());
};
let format = query
.format
- .unwrap_or(sink.info.config_format.unwrap_or_default());
+ .unwrap_or(sink.info.plugin_config_format.unwrap_or_default());
let (content_type, config) = map_connector_config(config, format)?;
let mut headers = HeaderMap::new();
headers.insert(header::CONTENT_TYPE, content_type);
@@ -105,7 +109,7 @@ async fn get_sink_transforms(
return Err(ApiError::Error(RuntimeError::SinkNotFound(key)));
};
let sink = sink.lock().await;
- let Some(transforms) = sink.transforms.as_ref() else {
+ let Some(transforms) = sink.config.transforms.as_ref() else {
return Ok(Json(vec![]));
};
@@ -120,3 +124,67 @@ async fn get_sink_transforms(
.collect(),
))
}
+
+async fn get_sink_configs(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+) -> Result<Json<Vec<SinkConfigResponse>>, ApiError> {
+ let active_config = context
+ .sinks
+ .get_config(&key)
+ .await
+ .ok_or(ApiError::Error(RuntimeError::SinkNotFound(key.clone())))?;
+ let configs = context.config_provider.get_sink_configs(&key).await?;
+ let configs = configs
+ .into_iter()
+ .map(|config| {
+ let active = config.version == active_config.version;
+ SinkConfigResponse { config, active }
+ })
+ .collect();
+ Ok(Json(configs))
+}
+
+async fn get_sink_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key, version)): Path<(String, u64)>,
+) -> Result<Json<SinkConfigResponse>, ApiError> {
+ let active_config = context
+ .sinks
+ .get_config(&key)
+ .await
+ .ok_or(ApiError::Error(RuntimeError::SinkNotFound(key.clone())))?;
+
+ let config = context
+ .config_provider
+ .get_sink_config(&key, version)
+ .await?;
+
+ match config {
+ Some(sink_config) => {
+ let active = sink_config.version == active_config.version;
+ Ok(Json(SinkConfigResponse {
+ config: sink_config,
+ active,
+ }))
+ }
+ None => Err(ApiError::Error(RuntimeError::SinkConfigNotFound(
+ key, version,
+ ))),
+ }
+}
+
+async fn get_sink_active_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+) -> Result<Json<SinkConfigResponse>, ApiError> {
+ let config = context
+ .sinks
+ .get_config(&key)
+ .await
+ .ok_or(ApiError::Error(RuntimeError::SinkNotFound(key)))?;
+ Ok(Json(SinkConfigResponse {
+ config,
+ active: true,
+ }))
+}
diff --git a/core/connectors/runtime/src/api/source.rs b/core/connectors/runtime/src/api/source.rs
index ea940bac9..1b76d2f88 100644
--- a/core/connectors/runtime/src/api/source.rs
+++ b/core/connectors/runtime/src/api/source.rs
@@ -22,6 +22,7 @@ use super::{
error::ApiError,
models::{SourceDetailsResponse, SourceInfoResponse, TransformResponse},
};
+use crate::api::models::SourceConfigResponse;
use crate::configs::connectors::ConfigFormat;
use crate::{context::RuntimeContext, error::RuntimeError};
use axum::{
@@ -38,8 +39,17 @@ pub fn router(state: Arc<RuntimeContext>) -> Router {
Router::new()
.route("/sources", get(get_sources))
.route("/sources/{key}", get(get_source))
- .route("/sources/{key}/config", get(get_source_config))
+ .route(
+ "/sources/{key}/plugin_config",
+ get(get_source_plugin_config),
+ )
.route("/sources/{key}/transforms", get(get_source_transforms))
+ .route("/sources/{key}/configs", get(get_source_configs))
+ .route("/sources/{key}/configs/{version}", get(get_source_config))
+ .route(
+ "/sources/{key}/active_config",
+ get(get_source_active_config),
+ )
.with_state(state)
}
@@ -66,11 +76,11 @@ async fn get_source(
let source = source.lock().await;
Ok(Json(SourceDetailsResponse {
info: source.info.clone().into(),
- streams: source.streams.to_vec(),
+ streams: source.config.streams.to_vec(),
}))
}
-async fn get_source_config(
+async fn get_source_plugin_config(
State(context): State<Arc<RuntimeContext>>,
Path(key): Path<String>,
Query(query): Query<GetSourceConfig>,
@@ -79,13 +89,13 @@ async fn get_source_config(
return Err(ApiError::Error(RuntimeError::SourceNotFound(key)));
};
let source = source.lock().await;
- let Some(config) = source.config.as_ref() else {
+ let Some(config) = source.config.plugin_config.as_ref() else {
return Ok(StatusCode::NOT_FOUND.into_response());
};
let format = query
.format
- .unwrap_or(source.info.config_format.unwrap_or_default());
+ .unwrap_or(source.info.plugin_config_format.unwrap_or_default());
let (content_type, config) = map_connector_config(config, format)?;
let mut headers = HeaderMap::new();
headers.insert(header::CONTENT_TYPE, content_type);
@@ -105,7 +115,7 @@ async fn get_source_transforms(
return Err(ApiError::Error(RuntimeError::SourceNotFound(key)));
};
let source = source.lock().await;
- let Some(transforms) = source.transforms.as_ref() else {
+ let Some(transforms) = source.config.transforms.as_ref() else {
return Ok(Json(vec![]));
};
@@ -120,3 +130,67 @@ async fn get_source_transforms(
.collect(),
))
}
+
+async fn get_source_configs(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+) -> Result<Json<Vec<SourceConfigResponse>>, ApiError> {
+ let active_config = context
+ .sources
+ .get_config(&key)
+ .await
+ .ok_or(ApiError::Error(RuntimeError::SourceNotFound(key.clone())))?;
+ let configs = context.config_provider.get_source_configs(&key).await?;
+ let configs = configs
+ .into_iter()
+ .map(|config| {
+ let active = config.version == active_config.version;
+ SourceConfigResponse { config, active }
+ })
+ .collect();
+ Ok(Json(configs))
+}
+
+async fn get_source_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key, version)): Path<(String, u64)>,
+) -> Result<Json<SourceConfigResponse>, ApiError> {
+ let active_config = context
+ .sources
+ .get_config(&key)
+ .await
+ .ok_or(ApiError::Error(RuntimeError::SourceNotFound(key.clone())))?;
+
+ let config = context
+ .config_provider
+ .get_source_config(&key, version)
+ .await?;
+
+ match config {
+ Some(source_config) => {
+ let active = source_config.version == active_config.version;
+ Ok(Json(SourceConfigResponse {
+ config: source_config,
+ active,
+ }))
+ }
+ None => Err(ApiError::Error(RuntimeError::SourceConfigNotFound(
+ key, version,
+ ))),
+ }
+}
+
+async fn get_source_active_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+) -> Result<Json<SourceConfigResponse>, ApiError> {
+ let config = context
+ .sources
+ .get_config(&key)
+ .await
+ .ok_or(ApiError::Error(RuntimeError::SourceNotFound(key)))?;
+ Ok(Json(SourceConfigResponse {
+ config,
+ active: true,
+ }))
+}
diff --git a/core/connectors/runtime/src/configs/connectors.rs b/core/connectors/runtime/src/configs/connectors.rs
index c1c3e9e0e..f782621c7 100644
--- a/core/connectors/runtime/src/configs/connectors.rs
+++ b/core/connectors/runtime/src/configs/connectors.rs
@@ -61,28 +61,28 @@ impl Default for ConnectorConfig {
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct SinkConfig {
- pub id: String,
+ pub key: String,
pub enabled: bool,
pub version: u64,
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 plugin_config_format: Option<ConfigFormat>,
+ pub plugin_config: Option<serde_json::Value>,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct SourceConfig {
- pub id: String,
+ pub key: String,
pub enabled: bool,
pub version: u64,
pub name: String,
pub path: String,
pub transforms: Option<TransformsConfig>,
pub streams: Vec<StreamProducerConfig>,
- pub config_format: Option<ConfigFormat>,
- pub config: Option<serde_json::Value>,
+ pub plugin_config_format: Option<ConfigFormat>,
+ pub plugin_config: Option<serde_json::Value>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
@@ -111,8 +111,20 @@ pub struct StreamProducerConfig {
}
#[async_trait]
-pub trait ConnectorsConfigProvider {
- async fn load_configs(&self) -> Result<ConnectorsConfig, RuntimeError>;
+pub trait ConnectorsConfigProvider: Send + Sync {
+ async fn get_all_configs(&self) -> Result<ConnectorsConfig, RuntimeError>;
+ async fn get_sink_configs(&self, key: &str) -> Result<Vec<SinkConfig>, RuntimeError>;
+ async fn get_sink_config(
+ &self,
+ key: &str,
+ version: u64,
+ ) -> Result<Option<SinkConfig>, RuntimeError>;
+ async fn get_source_configs(&self, key: &str) -> Result<Vec<SourceConfig>, RuntimeError>;
+ async fn get_source_config(
+ &self,
+ key: &str,
+ version: u64,
+ ) -> Result<Option<SourceConfig>, RuntimeError>;
}
impl From<RuntimeConnectorsConfig> for Box<dyn ConnectorsConfigProvider> {
@@ -128,8 +140,8 @@ impl From<RuntimeConnectorsConfig> for Box<dyn ConnectorsConfigProvider> {
#[derive(Debug, Default, Clone, Deserialize, Serialize)]
#[serde(default)]
pub struct ConnectorsConfig {
- sinks: HashMap<String, SinkConfig>,
- sources: HashMap<String, SourceConfig>,
+ sinks: HashMap<String, Vec<SinkConfig>>,
+ sources: HashMap<String, Vec<SourceConfig>>,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
@@ -154,7 +166,7 @@ impl std::fmt::Display for SinkConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ enabled: {}, name: {}, path: {}, transforms: {:?}, streams: [{}], config_format: {:?} }}",
+ "{{ enabled: {}, name: {}, path: {}, transforms: {:?}, streams: [{}], plugin_config_format: {:?} }}",
self.enabled,
self.name,
self.path,
@@ -164,7 +176,7 @@ impl std::fmt::Display for SinkConfig {
.map(|s| s.to_string())
.collect::<Vec<String>>()
.join(", "),
- self.config_format,
+ self.plugin_config_format,
)
}
}
@@ -173,7 +185,7 @@ impl std::fmt::Display for SourceConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ enabled: {}, name: {}, path: {}, transforms: {:?}, streams: [{}], config_format: {:?} }}",
+ "{{ enabled: {}, name: {}, path: {}, transforms: {:?}, streams: [{}], plugin_config_format: {:?} }}",
self.enabled,
self.name,
self.path,
@@ -183,7 +195,7 @@ impl std::fmt::Display for SourceConfig {
.map(|s| s.to_string())
.collect::<Vec<String>>()
.join(", "),
- self.config_format,
+ self.plugin_config_format,
)
}
}
@@ -229,11 +241,35 @@ impl std::fmt::Display for StreamProducerConfig {
}
impl ConnectorsConfig {
- pub fn sinks(&self) -> HashMap<String, SinkConfig> {
- self.sinks.clone()
+ pub fn sinks_latest(&self) -> HashMap<String, SinkConfig> {
+ self.sinks
+ .iter()
+ .filter(|(_, v)| !v.is_empty())
+ .map(|(k, v)| {
+ (
+ k.clone(),
+ v.iter()
+ .max_by_key(|config| config.version)
+ .expect("at least one sink config must be present")
+ .clone(),
+ )
+ })
+ .collect()
}
- pub fn sources(&self) -> HashMap<String, SourceConfig> {
- self.sources.clone()
+ pub fn sources_latest(&self) -> HashMap<String, SourceConfig> {
+ self.sources
+ .iter()
+ .filter(|(_, v)| !v.is_empty())
+ .map(|(k, v)| {
+ (
+ k.clone(),
+ v.iter()
+ .max_by_key(|config| config.version)
+ .expect("at least one source config must be present")
+ .clone(),
+ )
+ })
+ .collect()
}
}
diff --git a/core/connectors/runtime/src/configs/connectors/local_provider.rs b/core/connectors/runtime/src/configs/connectors/local_provider.rs
index 066951ac9..b2f27b681 100644
--- a/core/connectors/runtime/src/configs/connectors/local_provider.rs
+++ b/core/connectors/runtime/src/configs/connectors/local_provider.rs
@@ -17,25 +17,32 @@
* under the License.
*/
-use crate::configs::connectors::{ConnectorConfig, ConnectorsConfig, ConnectorsConfigProvider};
+use crate::configs::connectors::local_provider::ConnectorType::{Sink, Source};
+use crate::configs::connectors::{
+ ConnectorConfig, ConnectorsConfig, ConnectorsConfigProvider, SinkConfig, SourceConfig,
+};
use crate::error::RuntimeError;
use async_trait::async_trait;
+use dashmap::DashMap;
use figment::value::Dict;
use figment::{Metadata, Profile, Provider};
use iggy_common::{ConfigProvider, CustomEnvProvider, FileConfigProvider};
use serde::Deserialize;
use std::collections::HashMap;
use std::path::Path;
+use strum::Display;
use tracing::{debug, info, warn};
pub struct LocalConnectorsConfigProvider {
config_dir: String,
+ file_mapping: DashMap<String, String>,
}
impl LocalConnectorsConfigProvider {
pub fn new(config_dir: &str) -> Self {
Self {
config_dir: config_dir.to_owned(),
+ file_mapping: DashMap::new(),
}
}
@@ -61,20 +68,107 @@ impl LocalConnectorsConfigProvider {
))
})
}
+
+ fn config_dir_exists(&self) -> Result<bool, RuntimeError> {
+ if self.config_dir.is_empty() {
+ warn!("Connectors configuration directory not provided");
+ return Ok(false);
+ }
+ if !std::fs::exists(&self.config_dir)? {
+ warn!(
+ "Connectors configuration directory does not exist: {}",
+ self.config_dir
+ );
+ return Ok(false);
+ }
+ Ok(true)
+ }
+}
+
+#[derive(Display, Debug)]
+enum ConnectorType {
+ Sink,
+ Source,
+}
+
+#[derive(Debug)]
+struct ConnectorConfigFileId {
+ connector_type: ConnectorType,
+ key: String,
+ version: u64,
+}
+
+impl ConnectorConfigFileId {
+ fn from_sink_key_and_version(key: &str, version: u64) -> Self {
+ Self {
+ connector_type: Sink,
+ key: key.to_owned(),
+ version,
+ }
+ }
+
+ fn from_sink_key(key: &str) -> Self {
+ Self {
+ connector_type: Sink,
+ key: key.to_owned(),
+ version: 0,
+ }
+ }
+
+ fn from_source_key_and_version(key: &str, version: u64) -> Self {
+ Self {
+ connector_type: Source,
+ key: key.to_owned(),
+ version,
+ }
+ }
+
+ fn from_source_key(key: &str) -> Self {
+ Self {
+ connector_type: Source,
+ key: key.to_owned(),
+ version: 0,
+ }
+ }
+
+ fn to_file_mapping_key(&self) -> String {
+ format!("{}_{}_{}", self.connector_type, self.key, self.version)
+ }
+
+ fn to_file_mapping_key_prefix(&self) -> String {
+ format!("{}_{}_", self.connector_type, self.key)
+ }
+}
+
+impl From<ConnectorConfig> for ConnectorConfigFileId {
+ fn from(value: ConnectorConfig) -> Self {
+ match value {
+ ConnectorConfig::Sink(config) => Self {
+ connector_type: Sink,
+ key: config.key,
+ version: config.version,
+ },
+ ConnectorConfig::Source(config) => Self {
+ connector_type: Source,
+ key: config.key,
+ version: config.version,
+ },
+ }
+ }
}
#[derive(Debug, Deserialize)]
#[serde(tag = "type", rename_all = "lowercase")]
pub enum BaseConnectorConfig {
- Sink { id: String },
- Source { id: String },
+ Sink { key: String },
+ Source { key: String },
}
impl BaseConnectorConfig {
- fn id(&self) -> &str {
+ fn key(&self) -> &str {
match self {
- BaseConnectorConfig::Sink { id } => id,
- BaseConnectorConfig::Source { id } => id,
+ BaseConnectorConfig::Sink { key, .. } => key,
+ BaseConnectorConfig::Source { key, .. } => key,
}
}
@@ -88,20 +182,13 @@ impl BaseConnectorConfig {
#[async_trait]
impl ConnectorsConfigProvider for LocalConnectorsConfigProvider {
- async fn load_configs(&self) -> Result<ConnectorsConfig, RuntimeError> {
- if self.config_dir.is_empty() {
- info!("Connectors configuration directory not provided, skipping initialization");
- return Ok(ConnectorsConfig::default());
- }
- if !std::fs::exists(&self.config_dir)? {
- warn!(
- "Connectors configuration directory does not exist: {}",
- self.config_dir
- );
+ async fn get_all_configs(&self) -> Result<ConnectorsConfig, RuntimeError> {
+ if !self.config_dir_exists()? {
return Ok(ConnectorsConfig::default());
}
- let mut sinks = HashMap::new();
- let mut sources = HashMap::new();
+
+ let mut sinks: HashMap<String, Vec<SinkConfig>> = HashMap::new();
+ let mut sources: HashMap<String, Vec<SourceConfig>> = HashMap::new();
info!("Loading connectors configuration from: {}", self.config_dir);
let entries = std::fs::read_dir(&self.config_dir)?;
for entry in entries.flatten() {
@@ -110,28 +197,168 @@ impl ConnectorsConfigProvider for LocalConnectorsConfigProvider {
debug!("Loading connector configuration from: {:?}", path);
let base_config = Self::read_base_config(&path)?;
debug!("Loaded base configuration: {:?}", base_config);
- let connector_config: ConnectorConfig = Self::create_file_config_provider(
- path.to_str()
- .expect("Failed to convert connector configuration path to string")
- .to_string(),
- &base_config,
- )
- .load_config()
- .await
- .expect("Failed to load connector configuration");
-
- match connector_config {
+ let path = path
+ .to_str()
+ .expect("Failed to convert connector configuration path to string")
+ .to_string();
+ let connector_config: ConnectorConfig =
+ Self::create_file_config_provider(path.clone(), &base_config)
+ .load_config()
+ .await
+ .expect("Failed to load connector configuration");
+
+ match connector_config.clone() {
ConnectorConfig::Sink(sink_config) => {
- sinks.insert(base_config.id().to_owned(), sink_config);
+ sinks
+ .entry(base_config.key().to_owned())
+ .or_default()
+ .push(sink_config);
}
ConnectorConfig::Source(source_config) => {
- sources.insert(base_config.id().to_owned(), source_config);
+ sources
+ .entry(base_config.key().to_owned())
+ .or_default()
+ .push(source_config);
}
}
+
+ let file_id: ConnectorConfigFileId = connector_config.into();
+ self.file_mapping
+ .insert(file_id.to_file_mapping_key(), path);
}
}
Ok(ConnectorsConfig { sinks, sources })
}
+
+ async fn get_sink_configs(&self, key: &str) -> Result<Vec<SinkConfig>, RuntimeError> {
+ if !self.config_dir_exists()? {
+ return Ok(Vec::new());
+ }
+
+ let file_id = ConnectorConfigFileId::from_sink_key(key);
+ let file_mapping_key_prefix = file_id.to_file_mapping_key_prefix();
+
+ let paths = self
+ .file_mapping
+ .iter()
+ .filter(|item| item.key().starts_with(&file_mapping_key_prefix))
+ .map(|item| item.value().clone())
+ .collect::<Vec<_>>();
+
+ let mut configs = Vec::new();
+ for path in paths {
+ let base_config = BaseConnectorConfig::Sink {
+ key: key.to_owned(),
+ };
+ let connector_config: ConnectorConfig =
+ Self::create_file_config_provider(path, &base_config)
+ .load_config()
+ .await
+ .expect("Failed to load sink configuration");
+ if let ConnectorConfig::Sink(sink_config) = connector_config {
+ configs.push(sink_config);
+ }
+ }
+ Ok(configs)
+ }
+
+ async fn get_sink_config(
+ &self,
+ key: &str,
+ version: u64,
+ ) -> Result<Option<SinkConfig>, RuntimeError> {
+ if !self.config_dir_exists()? {
+ return Ok(None);
+ }
+
+ let file_id = ConnectorConfigFileId::from_sink_key_and_version(key, version);
+ if let Some(path) = self.file_mapping.get(&file_id.to_file_mapping_key()) {
+ let path = path.value().clone();
+ let base_config = BaseConnectorConfig::Sink {
+ key: key.to_owned(),
+ };
+ let connector_config: ConnectorConfig =
+ Self::create_file_config_provider(path, &base_config)
+ .load_config()
+ .await
+ .expect("Failed to load sink configuration");
+ if let ConnectorConfig::Sink(sink_config) = connector_config {
+ Ok(Some(sink_config))
+ } else {
+ Err(RuntimeError::InvalidConfiguration(
+ "Configuration is not a sink configuration".to_owned(),
+ ))
+ }
+ } else {
+ debug!("No file mapping found for connector config: {:?}", file_id);
+ Ok(None)
+ }
+ }
+ async fn get_source_configs(&self, key: &str) -> Result<Vec<SourceConfig>, RuntimeError> {
+ if !self.config_dir_exists()? {
+ return Ok(Vec::new());
+ }
+
+ let file_id = ConnectorConfigFileId::from_source_key(key);
+ let file_mapping_key_prefix = file_id.to_file_mapping_key_prefix();
+
+ let paths = self
+ .file_mapping
+ .iter()
+ .filter(|item| item.key().starts_with(&file_mapping_key_prefix))
+ .map(|item| item.value().clone())
+ .collect::<Vec<_>>();
+
+ let mut configs = Vec::new();
+ for path in paths {
+ let base_config = BaseConnectorConfig::Sink {
+ key: key.to_owned(),
+ };
+ let connector_config: ConnectorConfig =
+ Self::create_file_config_provider(path, &base_config)
+ .load_config()
+ .await
+ .expect("Failed to load sink configuration");
+ if let ConnectorConfig::Source(sink_config) = connector_config {
+ configs.push(sink_config);
+ }
+ }
+ Ok(configs)
+ }
+
+ async fn get_source_config(
+ &self,
+ key: &str,
+ version: u64,
+ ) -> Result<Option<SourceConfig>, RuntimeError> {
+ if !self.config_dir_exists()? {
+ return Ok(None);
+ }
+
+ let file_id = ConnectorConfigFileId::from_source_key_and_version(key, version);
+
+ if let Some(path) = self.file_mapping.get(&file_id.to_file_mapping_key()) {
+ let path = path.value().clone();
+ let base_config = BaseConnectorConfig::Sink {
+ key: key.to_owned(),
+ };
+ let connector_config: ConnectorConfig =
+ Self::create_file_config_provider(path, &base_config)
+ .load_config()
+ .await
+ .expect("Failed to load sink configuration");
+ if let ConnectorConfig::Source(sink_config) = connector_config {
+ Ok(Some(sink_config))
+ } else {
+ Err(RuntimeError::InvalidConfiguration(
+ "Configuration is not a source configuration".to_owned(),
+ ))
+ }
+ } else {
+ debug!("No file mapping found for connector config: {:?}", file_id);
+ Ok(None)
+ }
+ }
}
#[derive(Debug, Clone)]
@@ -143,10 +370,10 @@ pub struct ConnectorEnvProvider {
impl ConnectorEnvProvider {
fn with_connector_base_config(base_config: &BaseConnectorConfig) -> Self {
let connector_type = base_config.connector_type().to_uppercase();
- let id = base_config.id().to_uppercase();
- let prefix = format!("IGGY_CONNECTORS_{}_{}_", connector_type, id);
+ let key = base_config.key().to_uppercase();
+ let prefix = format!("IGGY_CONNECTORS_{}_{}_", connector_type, key);
Self {
- connector_name: base_config.id().to_owned(),
+ connector_name: base_config.key().to_owned(),
provider: CustomEnvProvider::new(&prefix, &[]),
}
}
diff --git a/core/connectors/runtime/src/context.rs b/core/connectors/runtime/src/context.rs
index e81005e98..ef9b18910 100644
--- a/core/connectors/runtime/src/context.rs
+++ b/core/connectors/runtime/src/context.rs
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-use crate::configs::connectors::{SinkConfig, SourceConfig};
+use crate::configs::connectors::{ConnectorsConfigProvider, SinkConfig, SourceConfig};
use crate::configs::runtime::ConnectorsRuntimeConfig;
use crate::{
SinkConnectorWrapper, SourceConnectorWrapper,
@@ -26,12 +26,14 @@ use crate::{
},
};
use std::collections::HashMap;
+use std::sync::Arc;
use tracing::error;
pub struct RuntimeContext {
pub sinks: SinkManager,
pub sources: SourceManager,
pub api_key: String,
+ pub config_provider: Arc<dyn ConnectorsConfigProvider>,
}
pub fn init(
@@ -40,11 +42,13 @@ pub fn init(
sources_config: &HashMap<String, SourceConfig>,
sink_wrappers: &[SinkConnectorWrapper],
source_wrappers: &[SourceConnectorWrapper],
+ config_provider: Box<dyn ConnectorsConfigProvider>,
) -> RuntimeContext {
RuntimeContext {
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(),
+ config_provider: Arc::from(config_provider),
}
}
@@ -68,11 +72,9 @@ fn map_sinks(
path: sink_plugin.path.to_owned(),
enabled: sink_config.enabled,
running: sink_config.enabled,
- config_format: sink_plugin.config_format,
+ plugin_config_format: sink_plugin.config_format,
},
- config: sink_config.config.clone(),
- transforms: sink_config.transforms.clone(),
- streams: sink_config.streams.clone(),
+ config: sink_config.clone(),
});
}
}
@@ -99,11 +101,9 @@ fn map_sources(
path: source_plugin.path.to_owned(),
enabled: source_config.enabled,
running: source_config.enabled,
- config_format: source_plugin.config_format,
+ plugin_config_format: source_plugin.config_format,
},
- config: source_config.config.clone(),
- transforms: source_config.transforms.clone(),
- streams: source_config.streams.clone(),
+ config: source_config.clone(),
});
}
}
diff --git a/core/connectors/runtime/src/error.rs b/core/connectors/runtime/src/error.rs
index 640400bef..742e31b19 100644
--- a/core/connectors/runtime/src/error.rs
+++ b/core/connectors/runtime/src/error.rs
@@ -45,8 +45,12 @@ pub enum RuntimeError {
JsonError(#[from] serde_json::Error),
#[error("Sink not found with key: {0}")]
SinkNotFound(String),
+ #[error("Sink config not found with key: {0}, version: {1}")]
+ SinkConfigNotFound(String, u64),
#[error("Source not found with key: {0}")]
SourceNotFound(String),
+ #[error("Source config not found with key: {0}, version: {1}")]
+ SourceConfigNotFound(String, u64),
#[error("Cannot convert configuration")]
CannotConvertConfiguration,
#[error("IO operation failed with error: {0:?}")]
@@ -57,6 +61,7 @@ impl RuntimeError {
pub fn as_code(&self) -> &'static str {
match self {
RuntimeError::SinkNotFound(_) => "sink_not_found",
+ RuntimeError::SinkConfigNotFound(_, _) => "sink_config_not_found",
RuntimeError::SourceNotFound(_) => "source_not_found",
RuntimeError::MissingIggyCredentials => "invalid_configuration",
RuntimeError::InvalidConfiguration(_) => "invalid_configuration",
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
index dd975d48d..d0df71406 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -129,8 +129,8 @@ async fn main() -> Result<(), RuntimeError> {
let connectors_config_provider: Box<dyn ConnectorsConfigProvider> =
config.connectors.clone().into();
- let connectors_config = connectors_config_provider.load_configs().await?;
- let sources_config = connectors_config.sources();
+ let connectors_config = connectors_config_provider.get_all_configs().await?;
+ let sources_config = connectors_config.sources_latest();
let sources = source::init(
sources_config.clone(),
&iggy_clients.producer,
@@ -138,7 +138,7 @@ async fn main() -> Result<(), RuntimeError> {
)
.await?;
- let sinks_config = connectors_config.sinks();
+ let sinks_config = connectors_config.sinks_latest();
let sinks = sink::init(sinks_config.clone(), &iggy_clients.consumer).await?;
let mut sink_wrappers = vec![];
@@ -181,6 +181,7 @@ async fn main() -> Result<(), RuntimeError> {
&sources_config,
&sink_wrappers,
&source_wrappers,
+ connectors_config_provider,
);
let context = Arc::new(context);
api::init(&config.http, context).await;
diff --git a/core/connectors/runtime/src/manager/sink.rs b/core/connectors/runtime/src/manager/sink.rs
index 744ffc6ee..65d594fa0 100644
--- a/core/connectors/runtime/src/manager/sink.rs
+++ b/core/connectors/runtime/src/manager/sink.rs
@@ -16,37 +16,46 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-use crate::configs::connectors::{ConfigFormat, StreamConsumerConfig, TransformsConfig};
-use std::{collections::HashMap, sync::Arc};
+use crate::configs::connectors::{ConfigFormat, SinkConfig};
+use dashmap::DashMap;
+use std::collections::HashMap;
+use std::sync::Arc;
use tokio::sync::Mutex;
#[derive(Debug)]
pub struct SinkManager {
- sinks: Mutex<HashMap<String, Arc<Mutex<SinkDetails>>>>,
+ sinks: DashMap<String, Arc<Mutex<SinkDetails>>>,
}
impl SinkManager {
pub fn new(sinks: Vec<SinkDetails>) -> Self {
Self {
- sinks: Mutex::new(
+ sinks: DashMap::from_iter(
sinks
.into_iter()
.map(|sink| (sink.info.key.to_owned(), Arc::new(Mutex::new(sink))))
- .collect(),
+ .collect::<HashMap<_, _>>(),
),
}
}
pub async fn get(&self, key: &str) -> Option<Arc<Mutex<SinkDetails>>> {
- let sinks = self.sinks.lock().await;
- sinks.get(key).cloned()
+ self.sinks.get(key).map(|entry| entry.value().clone())
+ }
+
+ pub async fn get_config(&self, key: &str) -> Option<SinkConfig> {
+ if let Some(sink) = self.sinks.get(key).map(|entry| entry.value().clone()) {
+ let sink = sink.lock().await;
+ Some(sink.config.clone())
+ } else {
+ None
+ }
}
pub async fn get_all(&self) -> Vec<SinkInfo> {
- let sinks = self.sinks.lock().await;
+ let sinks = &self.sinks;
let mut results = Vec::with_capacity(sinks.len());
- for sink in sinks.values() {
+ for sink in sinks.iter().map(|entry| entry.value().clone()) {
let sink = sink.lock().await;
results.push(sink.info.clone());
}
@@ -62,13 +71,11 @@ pub struct SinkInfo {
pub path: String,
pub enabled: bool,
pub running: bool,
- pub config_format: Option<ConfigFormat>,
+ pub plugin_config_format: Option<ConfigFormat>,
}
#[derive(Debug)]
pub struct SinkDetails {
pub info: SinkInfo,
- pub transforms: Option<TransformsConfig>,
- pub streams: Vec<StreamConsumerConfig>,
- pub config: Option<serde_json::Value>,
+ pub config: SinkConfig,
}
diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs
index 94bdae58f..258151837 100644
--- a/core/connectors/runtime/src/manager/source.rs
+++ b/core/connectors/runtime/src/manager/source.rs
@@ -16,37 +16,46 @@
* specific language governing permissions and limitations
* under the License.
*/
-
-use crate::configs::connectors::{ConfigFormat, StreamProducerConfig, TransformsConfig};
-use std::{collections::HashMap, sync::Arc};
+use crate::configs::connectors::{ConfigFormat, SourceConfig};
+use dashmap::DashMap;
+use std::collections::HashMap;
+use std::sync::Arc;
use tokio::sync::Mutex;
#[derive(Debug)]
pub struct SourceManager {
- sources: Mutex<HashMap<String, Arc<Mutex<SourceDetails>>>>,
+ sources: DashMap<String, Arc<Mutex<SourceDetails>>>,
}
impl SourceManager {
pub fn new(sources: Vec<SourceDetails>) -> Self {
Self {
- sources: Mutex::new(
+ sources: DashMap::from_iter(
sources
.into_iter()
.map(|source| (source.info.key.to_owned(), Arc::new(Mutex::new(source))))
- .collect(),
+ .collect::<HashMap<_, _>>(),
),
}
}
pub async fn get(&self, key: &str) -> Option<Arc<Mutex<SourceDetails>>> {
- let sources = self.sources.lock().await;
- sources.get(key).cloned()
+ self.sources.get(key).map(|entry| entry.value().clone())
+ }
+
+ pub async fn get_config(&self, key: &str) -> Option<SourceConfig> {
+ if let Some(source) = self.sources.get(key).map(|entry| entry.value().clone()) {
+ let source = source.lock().await;
+ Some(source.config.clone())
+ } else {
+ None
+ }
}
pub async fn get_all(&self) -> Vec<SourceInfo> {
- let sources = self.sources.lock().await;
+ let sources = &self.sources;
let mut results = Vec::with_capacity(sources.len());
- for source in sources.values() {
+ for source in sources.iter().map(|entry| entry.value().clone()) {
let source = source.lock().await;
results.push(source.info.clone());
}
@@ -62,13 +71,11 @@ pub struct SourceInfo {
pub path: String,
pub enabled: bool,
pub running: bool,
- pub config_format: Option<ConfigFormat>,
+ pub plugin_config_format: Option<ConfigFormat>,
}
#[derive(Debug)]
pub struct SourceDetails {
pub info: SourceInfo,
- pub transforms: Option<TransformsConfig>,
- pub streams: Vec<StreamProducerConfig>,
- pub config: Option<serde_json::Value>,
+ pub config: SourceConfig,
}
diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs
index 3097c85b0..6b1c9970c 100644
--- a/core/connectors/runtime/src/sink.rs
+++ b/core/connectors/runtime/src/sink.rs
@@ -62,7 +62,7 @@ pub async fn init(
info!("Sink container for plugin: {path} is already loaded.",);
init_sink(
&container.container,
- &config.config.unwrap_or_default(),
+ &config.plugin_config.unwrap_or_default(),
plugin_id,
);
container.plugins.push(SinkConnectorPlugin {
@@ -70,14 +70,18 @@ pub async fn init(
key: key.to_owned(),
name: name.to_owned(),
path: path.to_owned(),
- config_format: config.config_format,
+ config_format: config.plugin_config_format,
consumers: vec![],
});
} else {
let container: Container<SinkApi> =
unsafe { Container::load(&path).expect("Failed to load sink container") };
info!("Sink container for plugin: {path} loaded successfully.",);
- init_sink(&container, &config.config.unwrap_or_default(), plugin_id);
+ init_sink(
+ &container,
+ &config.plugin_config.unwrap_or_default(),
+ plugin_id,
+ );
sink_connectors.insert(
path.to_owned(),
SinkConnector {
@@ -87,7 +91,7 @@ pub async fn init(
key: key.to_owned(),
name: name.to_owned(),
path: path.to_owned(),
- config_format: config.config_format,
+ config_format: config.plugin_config_format,
consumers: vec![],
}],
},
@@ -258,9 +262,9 @@ async fn consume_messages(
Ok(())
}
-fn init_sink(container: &Container<SinkApi>, config: &serde_json::Value, id: u32) {
- let config = serde_json::to_string(config).expect("Invalid sink config.");
- (container.open)(id, config.as_ptr(), config.len());
+fn init_sink(container: &Container<SinkApi>, plugin_config: &serde_json::Value, id: u32) {
+ let plugin_config = serde_json::to_string(plugin_config).expect("Invalid sink plugin config.");
+ (container.open)(id, plugin_config.as_ptr(), plugin_config.len());
}
async fn process_messages(
diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs
index f5a305e43..d61803995 100644
--- a/core/connectors/runtime/src/source.rs
+++ b/core/connectors/runtime/src/source.rs
@@ -72,7 +72,7 @@ pub async fn init(
info!("Source container for plugin: {path} is already loaded.",);
init_source(
&container.container,
- &config.config.unwrap_or_default(),
+ &config.plugin_config.unwrap_or_default(),
plugin_id,
state,
);
@@ -81,7 +81,7 @@ pub async fn init(
key: key.to_owned(),
name: name.to_owned(),
path: path.to_owned(),
- config_format: config.config_format,
+ config_format: config.plugin_config_format,
producer: None,
transforms: vec![],
state_storage,
@@ -92,7 +92,7 @@ pub async fn init(
info!("Source container for plugin: {path} loaded successfully.",);
init_source(
&container,
- &config.config.unwrap_or_default(),
+ &config.plugin_config.unwrap_or_default(),
plugin_id,
state,
);
@@ -105,7 +105,7 @@ pub async fn init(
key: key.to_owned(),
name: name.to_owned(),
path: path.to_owned(),
- config_format: config.config_format,
+ config_format: config.plugin_config_format,
producer: None,
transforms: vec![],
state_storage,
@@ -171,14 +171,21 @@ pub async fn init(
fn init_source(
container: &Container<SourceApi>,
- config: &serde_json::Value,
+ plugin_config: &serde_json::Value,
id: u32,
state: Option<ConnectorState>,
) {
- let config = serde_json::to_string(config).expect("Invalid source config.");
+ let plugin_config =
+ serde_json::to_string(plugin_config).expect("Invalid source plugin config.");
let state_ptr = state.as_ref().map_or(std::ptr::null(), |s| s.0.as_ptr());
let state_len = state.as_ref().map_or(0, |s| s.0.len());
- (container.open)(id, config.as_ptr(), config.len(), state_ptr, state_len);
+ (container.open)(
+ id,
+ plugin_config.as_ptr(),
+ plugin_config.len(),
+ state_ptr,
+ state_len,
+ );
}
fn get_state_storage(state_path: &str, key: &str) -> StateStorage {
diff --git a/core/connectors/sdk/README.md b/core/connectors/sdk/README.md
index db9917332..1da28597c 100644
--- a/core/connectors/sdk/README.md
+++ b/core/connectors/sdk/README.md
@@ -42,7 +42,7 @@ config_dir = "path/to/connectors"
```toml
type = "source"
-id = "protobuf"
+key = "protobuf"
enabled = true
version = 0
name = "Protobuf Source"
@@ -55,7 +55,7 @@ schema = "proto"
batch_size = 1000
send_interval = "5ms"
-[config]
+[plugin_config]
schema_path = "schemas/message.proto"
message_type = "com.example.Message"
use_any_wrapper = true
@@ -65,7 +65,7 @@ use_any_wrapper = true
```toml
type = "sink"
-id = "protobuf"
+key = "protobuf"
enabled = true
version = 0
name = "Protobuf Sink"
diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md
index 9d8e95a4e..f29b4b92f 100644
--- a/core/connectors/sinks/README.md
+++ b/core/connectors/sinks/README.md
@@ -31,15 +31,15 @@ Each sink connector is configured in its own separate configuration file within
```rust
pub struct SinkConfig {
- pub id: String,
+ pub key: String,
pub enabled: bool,
pub version: u64,
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 plugin_config_format: Option<ConfigFormat>,
+ pub plugin_config: Option<serde_json::Value>,
}
```
@@ -56,14 +56,14 @@ config_dir = "path/to/connectors"
```toml
# Type of connector (sink or source)
type = "sink"
-id = "stdout" # Unique sink ID
+key = "stdout" # Unique sink ID
# Required configuration for a sink connector
enabled = true
version = 0
name = "Stdout sink"
path = "target/release/libiggy_connector_stdout_sink"
-config_format = "toml"
+plugin_config_format = "toml"
# Collection of the streams from which messages are consumed
[[streams]]
@@ -74,8 +74,8 @@ batch_length = 100
poll_interval = "5ms"
consumer_group = "stdout_sink_connector"
-# Custom configuration for the sink connector, deserialized to type T from `config` field
-[config]
+# Custom configuration for the sink connector, deserialized to type T from `plugin_config` field
+[plugin_config]
print_payload = true
# Optional data transformation(s) to be applied after consuming messages from the stream
@@ -90,7 +90,7 @@ value.static = "hello"
### Environment Variable Overrides
-Configuration properties can be overridden using environment variables. The pattern follows: `IGGY_CONNECTORS_SINK_[ID]_[PROPERTY]`
+Configuration properties can be overridden using environment variables. The pattern follows: `IGGY_CONNECTORS_SINK_[KEY]_[PROPERTY]`
For example, to override the `enabled` property for a sink with ID `stdout`:
diff --git a/core/connectors/sinks/quickwit_sink/README.md b/core/connectors/sinks/quickwit_sink/README.md
index 38dd4dc03..c8fc338ce 100644
--- a/core/connectors/sinks/quickwit_sink/README.md
+++ b/core/connectors/sinks/quickwit_sink/README.md
@@ -8,7 +8,7 @@ The Quickwit connector allows you to send data to the Quickwit API using HTTP. T
- `index`: The index configuration using YAML, as described in the [Quickwit index configuration docs](https://quickwit.io/docs/configuration/index-config)
```toml
-[sinks.quickwit.config]
+[plugin_config]
url = "http://localhost:7280"
index = """
version: 0.9
diff --git a/core/connectors/sinks/stdout_sink/README.md b/core/connectors/sinks/stdout_sink/README.md
index 9e826892d..a3bbed5e8 100644
--- a/core/connectors/sinks/stdout_sink/README.md
+++ b/core/connectors/sinks/stdout_sink/README.md
@@ -7,6 +7,6 @@ The Stdout sink allows you to print the data to the standard output (stdout). It
- `print_payload`: A boolean value indicating whether to print the payload or not. Defaults to `false`.
```toml
-[sinks.stdout.config]
+[plugin_config]
print_payload = false
```
diff --git a/core/connectors/sources/README.md b/core/connectors/sources/README.md
index 2864b74a0..4bdb8e985 100644
--- a/core/connectors/sources/README.md
+++ b/core/connectors/sources/README.md
@@ -26,15 +26,15 @@ Each source connector is configured in its own separate configuration file withi
```rust
pub struct SourceConfig {
- pub id: String,
+ pub key: String,
pub enabled: bool,
pub version: u64,
pub name: String,
pub path: String,
pub transforms: Option<TransformsConfig>,
pub streams: Vec<StreamProducerConfig>,
- pub config_format: Option<ConfigFormat>,
- pub config: Option<serde_json::Value>,
+ pub plugin_config_format: Option<ConfigFormat>,
+ pub plugin_config: Option<serde_json::Value>,
}
```
@@ -51,7 +51,7 @@ config_dir = "path/to/connectors"
```toml
# Type of connector (sink or source)
type = "source"
-id = "random" # Unique source ID
+key = "random" # Unique source ID
# Required configuration for a source connector
enabled = true # Toggle source on/off
@@ -68,8 +68,8 @@ schema = "json"
batch_length = 100
linger_time = "5ms"
-# Custom configuration for the source connector, deserialized to type T from `config` field
-[config]
+# Custom configuration for the source connector, deserialized to type T from `plugin_config` field
+[plugin_config]
messages_count = 10
# Optional data transformation(s) to be applied before sending messages to the stream
@@ -84,7 +84,7 @@ value.static = "hello"
### Environment Variable Overrides
-Configuration properties can be overridden using environment variables. The pattern follows: `IGGY_CONNECTORS_SOURCE_[ID]_[PROPERTY]`
+Configuration properties can be overridden using environment variables. The pattern follows: `IGGY_CONNECTORS_SOURCE_[KEY]_[PROPERTY]`
For example, to override the `enabled` property for a source with ID `random`:
diff --git a/core/connectors/sources/random_source/README.md b/core/connectors/sources/random_source/README.md
index b597ebea0..730304e34 100644
--- a/core/connectors/sources/random_source/README.md
+++ b/core/connectors/sources/random_source/README.md
@@ -10,7 +10,7 @@ The Random Source connector generates random data and sends it to the specified
- `payload_size`: An integer representing the size in bytes of the `text` field payload to generate. Defaults to `100`.
```toml
-[sources.random.config]
+[plugin_config]
interval = "100ms"
max_count = 1000
messages_range = [10, 50]