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/18 17:05:16 UTC
(iggy) branch master updated: feat(connectors): extend runtime API with config management options (#2348)
This is an automated email from the ASF dual-hosted git repository.
maciej pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 558502b19 feat(connectors): extend runtime API with config management options (#2348)
558502b19 is described below
commit 558502b194cc710d90dd7205791c9f6d14d5739b
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Tue Nov 18 18:05:11 2025 +0100
feat(connectors): extend runtime API with config management options (#2348)
---
Cargo.lock | 1 +
core/connectors/runtime/Cargo.toml | 1 +
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 | 173 +++++
core/connectors/runtime/src/api/config.rs | 16 +-
core/connectors/runtime/src/api/models.rs | 26 +-
core/connectors/runtime/src/api/sink.rs | 140 +++-
core/connectors/runtime/src/api/source.rs | 143 ++++-
core/connectors/runtime/src/configs/connectors.rs | 159 ++++-
.../src/configs/connectors/local_provider.rs | 702 +++++++++++++++++++--
core/connectors/runtime/src/context.rs | 18 +-
core/connectors/runtime/src/error.rs | 6 +
core/connectors/runtime/src/main.rs | 11 +-
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 +-
.../tests/connectors/postgres/config.toml} | 27 +-
.../postgres/{ => connectors_config}/postgres.toml | 8 +-
core/integration/tests/connectors/postgres/mod.rs | 13 +-
30 files changed, 1410 insertions(+), 229 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 3030058f0..6a817c3cd 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4281,6 +4281,7 @@ dependencies = [
"async-trait",
"axum",
"axum-server",
+ "chrono",
"dashmap",
"dlopen2",
"dotenvy",
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index 29821da47..ccfea1064 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -32,6 +32,7 @@ readme = "README.md"
async-trait = { workspace = true }
axum = { workspace = true }
axum-server = { workspace = true }
+chrono = { workspace = true }
dashmap = { workspace = true }
dlopen2 = { workspace = true }
dotenvy = { workspace = true }
diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md
index 483047a4f..dec986b6b 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}/configs/active`: active configuration details.
+- `PUT /sinks/{key}/configs/active`: activate a specific configuration version for the sink.
+- `GET /sinks/{key}/configs/plugin`: 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}/configs/active`: active configuration details.
+- `PUT /sources/{key}/configs/active`: activate a specific configuration version for the source.
+- `GET /sources/{key}/configs/plugin`: 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..7c08bef4f
--- /dev/null
+++ b/core/connectors/runtime/runtime.http
@@ -0,0 +1,173 @@
+# 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
+Content-Type: application/json
+
+{
+ "enabled": true,
+ "name": "Stdout sink",
+ "path": "target/release/libiggy_connector_stdout_sink",
+ "transforms": {
+ "add_fields": {
+ "enabled": true,
+ "fields": [
+ {
+ "key": "message",
+ "value": {
+ "static": "hello"
+ }
+ }
+ ]
+ }
+ },
+ "streams": [
+ {
+ "stream": "example_stream",
+ "topics": [
+ "example_topic"
+ ],
+ "schema": "json",
+ "batch_length": 100,
+ "poll_interval": "5ms",
+ "consumer_group": "stdout_sink_connector"
+ }
+ ],
+ "plugin_config_format": null,
+ "plugin_config": {
+ "print_payload": false
+ }
+ }
+
+###
+GET {{url}}/sinks/{{sink_key}}/configs/{{sink_version}}
+
+###
+DELETE {{url}}/sinks/{{sink_key}}/configs?version={{sink_version}}
+
+###
+GET {{url}}/sinks/{{sink_key}}/configs/active
+
+###
+PUT {{url}}/sinks/{{sink_key}}/configs/active
+Content-Type: application/json
+
+{
+ "version": {{sink_version}}
+}
+
+###
+GET {{url}}/sinks/{{sink_key}}/configs/plugin
+
+###
+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
+Content-Type: application/json
+
+{
+ "enabled": true,
+ "name": "Random source",
+ "path": "target/release/libiggy_connector_random_source",
+ "transforms": {
+ "add_fields": {
+ "enabled": true,
+ "fields": [
+ {
+ "key": "test_field",
+ "value": {
+ "static": "hello!"
+ }
+ }
+ ]
+ }
+ },
+ "streams": [
+ {
+ "stream": "example_stream",
+ "topic": "example_topic",
+ "schema": "json",
+ "batch_length": 1000,
+ "linger_time": "5ms"
+ }
+ ],
+ "plugin_config_format": "json",
+ "plugin_config": {
+ "interval": "1000ms",
+ "max_count": 1000000,
+ "messages_range": [
+ 1,
+ 5
+ ],
+ "payload_size": 200
+ }
+}
+
+###
+GET {{url}}/sources/{{source_key}}/configs/{{source_version}}
+
+###
+DELETE {{url}}/sources/{{source_key}}/configs?version={{source_version}}
+
+###
+GET {{url}}/sources/{{source_key}}/configs/active
+
+###
+PUT {{url}}/sources/{{source_key}}/configs/active
+Content-Type: application/json
+
+{
+ "version": {{source_version}}
+}
+
+###
+GET {{url}}/sources/{{source_key}}/configs/plugin
+
+###
+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..0a2c8791b 100644
--- a/core/connectors/runtime/src/api/sink.rs
+++ b/core/connectors/runtime/src/api/sink.rs
@@ -22,7 +22,8 @@ use super::{
error::ApiError,
models::{SinkDetailsResponse, SinkInfoResponse, TransformResponse},
};
-use crate::configs::connectors::ConfigFormat;
+use crate::api::models::SinkConfigResponse;
+use crate::configs::connectors::{ConfigFormat, CreateSinkConfig};
use crate::{context::RuntimeContext, error::RuntimeError};
use axum::{
Json, Router,
@@ -38,8 +39,19 @@ 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}/transforms", get(get_sink_transforms))
+ .route(
+ "/sinks/{key}/configs",
+ get(get_sink_configs)
+ .post(create_sink_config)
+ .delete(delete_sink_config),
+ )
+ .route("/sinks/{key}/configs/{version}", get(get_sink_config))
+ .route("/sinks/{key}/configs/plugin", get(get_sink_plugin_config))
+ .route(
+ "/sinks/{key}/configs/active",
+ get(get_sink_active_config).put(update_sink_active_config),
+ )
.with_state(state)
}
@@ -66,11 +78,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 +91,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 +117,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 +132,117 @@ 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 create_sink_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+ Json(config): Json<CreateSinkConfig>,
+) -> Result<Json<SinkConfigResponse>, ApiError> {
+ let created_config = context
+ .config_provider
+ .create_sink_config(&key, config.clone())
+ .await?;
+
+ Ok(Json(SinkConfigResponse {
+ config: created_config,
+ active: false,
+ }))
+}
+
+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, Some(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,
+ }))
+}
+
+#[derive(Debug, Deserialize)]
+struct UpdateSinkActiveConfig {
+ version: u64,
+}
+
+async fn update_sink_active_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+ Json(update): Json<UpdateSinkActiveConfig>,
+) -> Result<StatusCode, ApiError> {
+ context
+ .config_provider
+ .set_active_sink_version(&key, update.version)
+ .await?;
+ Ok(StatusCode::NO_CONTENT)
+}
+
+#[derive(Debug, Deserialize)]
+struct DeleteSinkConfig {
+ version: Option<u64>,
+}
+
+async fn delete_sink_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+ Query(query): Query<DeleteSinkConfig>,
+) -> Result<StatusCode, ApiError> {
+ context
+ .config_provider
+ .delete_sink_config(&key, query.version)
+ .await?;
+ Ok(StatusCode::NO_CONTENT)
+}
diff --git a/core/connectors/runtime/src/api/source.rs b/core/connectors/runtime/src/api/source.rs
index ea940bac9..7db1c9974 100644
--- a/core/connectors/runtime/src/api/source.rs
+++ b/core/connectors/runtime/src/api/source.rs
@@ -22,7 +22,8 @@ use super::{
error::ApiError,
models::{SourceDetailsResponse, SourceInfoResponse, TransformResponse},
};
-use crate::configs::connectors::ConfigFormat;
+use crate::api::models::SourceConfigResponse;
+use crate::configs::connectors::{ConfigFormat, CreateSourceConfig};
use crate::{context::RuntimeContext, error::RuntimeError};
use axum::{
Json, Router,
@@ -38,8 +39,22 @@ 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}/transforms", get(get_source_transforms))
+ .route(
+ "/sources/{key}/configs",
+ get(get_source_configs)
+ .post(create_source_config)
+ .delete(delete_source_config),
+ )
+ .route("/sources/{key}/configs/{version}", get(get_source_config))
+ .route(
+ "/sources/{key}/configs/plugin",
+ get(get_source_plugin_config),
+ )
+ .route(
+ "/sources/{key}/configs/active",
+ get(get_source_active_config).put(update_source_active_config),
+ )
.with_state(state)
}
@@ -66,11 +81,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 +94,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 +120,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 +135,117 @@ 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 create_source_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+ Json(config): Json<CreateSourceConfig>,
+) -> Result<Json<SourceConfigResponse>, ApiError> {
+ let created_config = context
+ .config_provider
+ .create_source_config(&key, config.clone())
+ .await?;
+
+ Ok(Json(SourceConfigResponse {
+ config: created_config,
+ active: false,
+ }))
+}
+
+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, Some(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,
+ }))
+}
+
+#[derive(Debug, Deserialize)]
+struct UpdateSourceActiveConfig {
+ version: u64,
+}
+
+async fn update_source_active_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+ Json(update): Json<UpdateSourceActiveConfig>,
+) -> Result<StatusCode, ApiError> {
+ context
+ .config_provider
+ .set_active_source_version(&key, update.version)
+ .await?;
+ Ok(StatusCode::NO_CONTENT)
+}
+
+#[derive(Debug, Deserialize)]
+struct DeleteSourceConfig {
+ version: Option<u64>,
+}
+
+async fn delete_source_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path((key,)): Path<(String,)>,
+ Query(query): Query<DeleteSourceConfig>,
+) -> Result<StatusCode, ApiError> {
+ context
+ .config_provider
+ .delete_source_config(&key, query.version)
+ .await?;
+ Ok(StatusCode::NO_CONTENT)
+}
diff --git a/core/connectors/runtime/src/configs/connectors.rs b/core/connectors/runtime/src/configs/connectors.rs
index c1c3e9e0e..7d2502cd3 100644
--- a/core/connectors/runtime/src/configs/connectors.rs
+++ b/core/connectors/runtime/src/configs/connectors.rs
@@ -23,6 +23,7 @@ use crate::configs::connectors::local_provider::LocalConnectorsConfigProvider;
use crate::configs::runtime::ConnectorsConfig as RuntimeConnectorsConfig;
use crate::error::RuntimeError;
use async_trait::async_trait;
+use chrono::{DateTime, Utc};
use iggy_connector_sdk::Schema;
use iggy_connector_sdk::transforms::TransformType;
use serde::{Deserialize, Serialize};
@@ -59,30 +60,93 @@ impl Default for ConnectorConfig {
}
}
+impl ConnectorConfig {
+ fn version(&self) -> u64 {
+ match self {
+ ConnectorConfig::Sink(config) => config.version,
+ ConnectorConfig::Source(config) => config.version,
+ }
+ }
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct CreateSinkConfig {
+ pub enabled: bool,
+ pub name: String,
+ pub path: String,
+ pub transforms: Option<TransformsConfig>,
+ pub streams: Vec<StreamConsumerConfig>,
+ pub plugin_config_format: Option<ConfigFormat>,
+ pub plugin_config: Option<serde_json::Value>,
+}
+
+impl CreateSinkConfig {
+ fn to_sink_config(&self, key: &str, version: u64) -> SinkConfig {
+ SinkConfig {
+ key: key.to_owned(),
+ enabled: self.enabled,
+ version,
+ name: self.name.clone(),
+ path: self.path.clone(),
+ transforms: self.transforms.clone(),
+ streams: self.streams.clone(),
+ plugin_config_format: self.plugin_config_format,
+ plugin_config: self.plugin_config.clone(),
+ }
+ }
+}
+
#[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 CreateSourceConfig {
+ pub enabled: bool,
+ pub name: String,
+ pub path: String,
+ pub transforms: Option<TransformsConfig>,
+ pub streams: Vec<StreamProducerConfig>,
+ pub plugin_config_format: Option<ConfigFormat>,
+ pub plugin_config: Option<serde_json::Value>,
+}
+
+impl CreateSourceConfig {
+ fn to_source_config(&self, key: &str, version: u64) -> SourceConfig {
+ SourceConfig {
+ key: key.to_owned(),
+ enabled: self.enabled,
+ version,
+ name: self.name.clone(),
+ path: self.path.clone(),
+ transforms: self.transforms.clone(),
+ streams: self.streams.clone(),
+ plugin_config_format: self.plugin_config_format,
+ plugin_config: self.plugin_config.clone(),
+ }
+ }
}
#[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)]
@@ -110,17 +174,64 @@ pub struct StreamProducerConfig {
pub linger_time: Option<String>,
}
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ConnectorConfigVersionInfo {
+ pub version: u64,
+ pub created_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+pub struct ConnectorConfigVersions {
+ pub sinks: HashMap<String, ConnectorConfigVersionInfo>,
+ pub sources: HashMap<String, ConnectorConfigVersionInfo>,
+}
+
#[async_trait]
-pub trait ConnectorsConfigProvider {
- async fn load_configs(&self) -> Result<ConnectorsConfig, RuntimeError>;
+pub trait ConnectorsConfigProvider: Send + Sync {
+ async fn create_sink_config(
+ &self,
+ key: &str,
+ config: CreateSinkConfig,
+ ) -> Result<SinkConfig, RuntimeError>;
+ async fn create_source_config(
+ &self,
+ key: &str,
+ config: CreateSourceConfig,
+ ) -> Result<SourceConfig, RuntimeError>;
+ async fn get_active_configs(&self) -> Result<ConnectorsConfig, RuntimeError>;
+ #[allow(dead_code)]
+ async fn get_active_configs_versions(&self) -> Result<ConnectorConfigVersions, RuntimeError>;
+ async fn set_active_sink_version(&self, key: &str, version: u64) -> Result<(), RuntimeError>;
+ async fn set_active_source_version(&self, key: &str, version: u64) -> Result<(), RuntimeError>;
+ async fn get_sink_configs(&self, key: &str) -> Result<Vec<SinkConfig>, RuntimeError>;
+ async fn get_sink_config(
+ &self,
+ key: &str,
+ version: Option<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: Option<u64>,
+ ) -> Result<Option<SourceConfig>, RuntimeError>;
+ async fn delete_sink_config(&self, key: &str, version: Option<u64>)
+ -> Result<(), RuntimeError>;
+ async fn delete_source_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<(), RuntimeError>;
}
-impl From<RuntimeConnectorsConfig> for Box<dyn ConnectorsConfigProvider> {
- fn from(value: RuntimeConnectorsConfig) -> Self {
- match value {
- RuntimeConnectorsConfig::Local(config) => {
- Box::new(LocalConnectorsConfigProvider::new(&config.config_dir))
- }
+pub async fn create_connectors_config_provider(
+ config: &RuntimeConnectorsConfig,
+) -> Result<Box<dyn ConnectorsConfigProvider>, RuntimeError> {
+ match config {
+ RuntimeConnectorsConfig::Local(config) => {
+ let provider = LocalConnectorsConfigProvider::new(&config.config_dir);
+ let provider = provider.init().await?;
+ Ok(Box::new(provider))
}
}
}
@@ -154,7 +265,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 +275,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 +284,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 +294,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 +340,15 @@ impl std::fmt::Display for StreamProducerConfig {
}
impl ConnectorsConfig {
- pub fn sinks(&self) -> HashMap<String, SinkConfig> {
- self.sinks.clone()
+ pub fn new(sinks: HashMap<String, SinkConfig>, sources: HashMap<String, SourceConfig>) -> Self {
+ Self { sinks, sources }
+ }
+
+ pub fn sinks(&self) -> &HashMap<String, SinkConfig> {
+ &self.sinks
}
- pub fn sources(&self) -> HashMap<String, SourceConfig> {
- self.sources.clone()
+ pub fn sources(&self) -> &HashMap<String, SourceConfig> {
+ &self.sources
}
}
diff --git a/core/connectors/runtime/src/configs/connectors/local_provider.rs b/core/connectors/runtime/src/configs/connectors/local_provider.rs
index 066951ac9..bd9565f02 100644
--- a/core/connectors/runtime/src/configs/connectors/local_provider.rs
+++ b/core/connectors/runtime/src/configs/connectors/local_provider.rs
@@ -17,28 +17,261 @@
* under the License.
*/
-use crate::configs::connectors::{ConnectorConfig, ConnectorsConfig, ConnectorsConfigProvider};
+use crate::configs::connectors::{
+ ConnectorConfig, ConnectorConfigVersionInfo, ConnectorConfigVersions, ConnectorsConfig,
+ ConnectorsConfigProvider, CreateSinkConfig, CreateSourceConfig, SinkConfig, SourceConfig,
+};
use crate::error::RuntimeError;
use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use dashmap::DashMap;
use figment::value::Dict;
use figment::{Metadata, Profile, Provider};
use iggy_common::{ConfigProvider, CustomEnvProvider, FileConfigProvider};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::Path;
use tracing::{debug, info, warn};
-pub struct LocalConnectorsConfigProvider {
+#[derive(Eq, PartialEq, Hash, Clone, Debug)]
+struct ConnectorId {
+ key: String,
+ version: u64,
+}
+
+impl ConnectorId {
+ fn to_filename_key(&self) -> String {
+ format!("{}_{}", self.key, self.version)
+ }
+}
+
+impl From<&ConnectorConfig> for ConnectorId {
+ fn from(value: &ConnectorConfig) -> Self {
+ match value {
+ ConnectorConfig::Sink(config) => ConnectorId {
+ key: config.key.clone(),
+ version: config.version,
+ },
+ ConnectorConfig::Source(config) => ConnectorId {
+ key: config.key.clone(),
+ version: config.version,
+ },
+ }
+ }
+}
+
+#[derive(Clone)]
+struct SinkConfigFile {
+ config: SinkConfig,
+ #[allow(dead_code)]
+ created_at: DateTime<Utc>,
+ path: String,
+}
+
+#[derive(Clone)]
+struct SourceConfigFile {
+ config: SourceConfig,
+ #[allow(dead_code)]
+ created_at: DateTime<Utc>,
+ path: String,
+}
+
+#[derive(Default)]
+struct ImportedConfigurations {
+ sinks: DashMap<ConnectorId, SinkConfigFile>,
+ sources: DashMap<ConnectorId, SourceConfigFile>,
+}
+
+impl ImportedConfigurations {
+ fn sinks(&self) -> &DashMap<ConnectorId, SinkConfigFile> {
+ &self.sinks
+ }
+
+ fn sinks_grouped_by_key(&self) -> HashMap<String, Vec<SinkConfigFile>> {
+ let mut grouped: HashMap<String, Vec<SinkConfigFile>> = HashMap::new();
+ for entry in self.sinks.iter() {
+ let key = entry.key().key.clone();
+ grouped.entry(key).or_default().push(entry.value().clone());
+ }
+ grouped
+ }
+
+ fn sources(&self) -> &DashMap<ConnectorId, SourceConfigFile> {
+ &self.sources
+ }
+
+ fn sources_grouped_by_key(&self) -> HashMap<String, Vec<SourceConfigFile>> {
+ let mut grouped: HashMap<String, Vec<SourceConfigFile>> = HashMap::new();
+ for entry in self.sources.iter() {
+ let key = entry.key().key.clone();
+ grouped.entry(key).or_default().push(entry.value().clone());
+ }
+ grouped
+ }
+}
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+struct ActiveConfigVersions {
+ #[serde(default)]
+ sinks: HashMap<String, u64>,
+ #[serde(default)]
+ sources: HashMap<String, u64>,
+}
+
+pub trait ProviderState {}
+
+pub struct Created {}
+
+impl ProviderState for Created {}
+
+#[derive(Default)]
+pub struct Initialized {
+ connectors_config: ImportedConfigurations,
+}
+
+impl ProviderState for Initialized {}
+
+pub struct LocalConnectorsConfigProvider<S: ProviderState> {
config_dir: String,
+ state: S,
}
-impl LocalConnectorsConfigProvider {
+impl LocalConnectorsConfigProvider<Created> {
pub fn new(config_dir: &str) -> Self {
Self {
config_dir: config_dir.to_owned(),
+ state: Created {},
}
}
+ pub async fn init(&self) -> Result<LocalConnectorsConfigProvider<Initialized>, RuntimeError> {
+ if self.config_dir.is_empty() {
+ return Err(RuntimeError::InvalidConfiguration(
+ "Connectors configuration directory not provided".to_string(),
+ ));
+ }
+ if !std::fs::exists(&self.config_dir)? {
+ warn!(
+ "Connectors configuration directory does not exist: {}",
+ self.config_dir
+ );
+ std::fs::create_dir_all(&self.config_dir)?;
+ return Ok(LocalConnectorsConfigProvider {
+ config_dir: self.config_dir.clone(),
+ state: Initialized::default(),
+ });
+ }
+
+ let sinks: DashMap<ConnectorId, SinkConfigFile> = DashMap::new();
+ let sources: DashMap<ConnectorId, SourceConfigFile> = DashMap::new();
+ info!("Loading connectors configuration from: {}", self.config_dir);
+ let entries = std::fs::read_dir(&self.config_dir)?;
+ for entry in entries.flatten() {
+ let path = entry.path();
+ if path.is_file() {
+ if let Some(file_name) = path.file_name().and_then(|n| n.to_str())
+ && file_name.starts_with('.')
+ {
+ debug!("Skipping hidden file: {:?}", path);
+ continue;
+ }
+
+ debug!("Loading connector configuration from: {:?}", path);
+ let base_config = Self::read_base_config(&path)?;
+ debug!("Loaded base configuration: {:?}", base_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");
+
+ let created_at: DateTime<Utc> = entry.metadata()?.created()?.into();
+ let connector_id: ConnectorId = (&connector_config).into();
+ match connector_config.clone() {
+ ConnectorConfig::Sink(sink_config) => {
+ sinks.insert(
+ connector_id,
+ SinkConfigFile {
+ config: sink_config,
+ created_at,
+ path,
+ },
+ );
+ }
+ ConnectorConfig::Source(source_config) => {
+ sources.insert(
+ connector_id,
+ SourceConfigFile {
+ config: source_config,
+ created_at,
+ path,
+ },
+ );
+ }
+ }
+
+ debug!(
+ "Loaded connector configuration with key {}, version {}, created at {}",
+ base_config.key(),
+ connector_config.version(),
+ created_at.to_rfc3339()
+ );
+ }
+ }
+ Ok(LocalConnectorsConfigProvider {
+ config_dir: self.config_dir.clone(),
+ state: Initialized {
+ connectors_config: ImportedConfigurations { sinks, sources },
+ },
+ })
+ }
+}
+
+impl LocalConnectorsConfigProvider<Initialized> {
+ fn active_versions_file_path(&self) -> String {
+ format!("{}/.active_versions.toml", self.config_dir)
+ }
+
+ fn load_active_versions(&self) -> ActiveConfigVersions {
+ let path = self.active_versions_file_path();
+ if !Path::new(&path).exists() {
+ return ActiveConfigVersions::default();
+ }
+
+ match std::fs::read(&path) {
+ Ok(data) => toml::from_slice(&data).unwrap_or_else(|err| {
+ warn!(
+ "Failed to parse active versions file '{}': {}",
+ path,
+ err.message()
+ );
+ ActiveConfigVersions::default()
+ }),
+ Err(err) => {
+ warn!("Failed to read active versions file '{}': {}", path, err);
+ ActiveConfigVersions::default()
+ }
+ }
+ }
+
+ fn save_active_versions(&self, versions: &ActiveConfigVersions) -> Result<(), RuntimeError> {
+ let path = self.active_versions_file_path();
+ let content = toml::to_string(versions).map_err(|err| {
+ RuntimeError::InvalidConfiguration(format!(
+ "Failed to serialize active versions: {}",
+ err
+ ))
+ })?;
+ std::fs::write(&path, content)?;
+ Ok(())
+ }
+}
+
+impl<S: ProviderState> LocalConnectorsConfigProvider<S> {
fn create_file_config_provider(
path: String,
base_config: &BaseConnectorConfig,
@@ -66,15 +299,15 @@ impl LocalConnectorsConfigProvider {
#[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,
}
}
@@ -87,50 +320,413 @@ 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());
+impl ConnectorsConfigProvider for LocalConnectorsConfigProvider<Initialized> {
+ async fn create_sink_config(
+ &self,
+ key: &str,
+ cmd: CreateSinkConfig,
+ ) -> Result<SinkConfig, RuntimeError> {
+ let sinks = self.state.connectors_config.sinks();
+ let next_version = sinks
+ .iter()
+ .filter(|entry| entry.key().key == key)
+ .max_by_key(|entry| entry.config.version)
+ .map(|entry| entry.config.version + 1)
+ .unwrap_or(0);
+
+ let config = cmd.to_sink_config(key, next_version);
+ let connector_config = ConnectorConfig::Sink(config.clone());
+ let connector_id: ConnectorId = (&connector_config).into();
+
+ let path = format!(
+ "{}/sink_{}.toml",
+ self.config_dir,
+ connector_id.to_filename_key()
+ );
+ std::fs::write(&path, toml::to_string(&connector_config).unwrap())?;
+ sinks.insert(
+ connector_id,
+ SinkConfigFile {
+ config: config.clone(),
+ created_at: Utc::now(),
+ path: path.clone(),
+ },
+ );
+
+ Ok(config)
+ }
+
+ async fn create_source_config(
+ &self,
+ key: &str,
+ cmd: CreateSourceConfig,
+ ) -> Result<SourceConfig, RuntimeError> {
+ let sources = &self.state.connectors_config.sources;
+ let next_version = sources
+ .iter()
+ .filter(|entry| entry.key().key == key)
+ .max_by_key(|entry| entry.config.version)
+ .map(|entry| entry.config.version + 1)
+ .unwrap_or(0);
+
+ let config = cmd.to_source_config(key, next_version);
+ let connector_config = ConnectorConfig::Source(config.clone());
+ let connector_id: ConnectorId = (&connector_config).into();
+
+ let path = format!(
+ "{}/source_{}.toml",
+ self.config_dir,
+ connector_id.to_filename_key()
+ );
+ std::fs::write(&path, toml::to_string(&connector_config).unwrap())?;
+ sources.insert(
+ connector_id,
+ SourceConfigFile {
+ config: config.clone(),
+ created_at: Utc::now(),
+ path: path.clone(),
+ },
+ );
+
+ Ok(config)
+ }
+
+ async fn get_active_configs(&self) -> Result<ConnectorsConfig, RuntimeError> {
+ let all_configs = &self.state.connectors_config;
+ let active_versions = self.load_active_versions();
+
+ let sinks = all_configs
+ .sinks_grouped_by_key()
+ .iter()
+ .filter_map(|(key, config_files)| {
+ if config_files.is_empty() {
+ return None;
+ }
+ let active_config = if let Some(&version) = active_versions.sinks.get(key) {
+ config_files
+ .iter()
+ .find(|c| c.config.version == version)
+ .cloned()
+ } else {
+ config_files
+ .iter()
+ .max_by_key(|c| c.config.version)
+ .cloned()
+ };
+ active_config.map(|config_file| (key.clone(), config_file.config.clone()))
+ })
+ .collect();
+
+ let sources = all_configs
+ .sources_grouped_by_key()
+ .iter()
+ .filter_map(|(key, config_files)| {
+ if config_files.is_empty() {
+ return None;
+ }
+ let active_config = if let Some(&version) = active_versions.sources.get(key) {
+ config_files
+ .iter()
+ .find(|c| c.config.version == version)
+ .cloned()
+ } else {
+ config_files
+ .iter()
+ .max_by_key(|c| c.config.version)
+ .cloned()
+ };
+ active_config.map(|config_file| (key.clone(), config_file.config.clone()))
+ })
+ .collect();
+
+ Ok(ConnectorsConfig::new(sinks, sources))
+ }
+
+ async fn get_active_configs_versions(&self) -> Result<ConnectorConfigVersions, RuntimeError> {
+ let all_configs = &self.state.connectors_config;
+ let active_versions = self.load_active_versions();
+
+ let sinks = all_configs
+ .sinks_grouped_by_key()
+ .iter()
+ .filter_map(|(key, config_files)| {
+ if config_files.is_empty() {
+ return None;
+ }
+ let latest_version = config_files
+ .iter()
+ .map(|c| c.config.version)
+ .max()
+ .expect("At least one config version must exist");
+ let active_version = active_versions
+ .sinks
+ .get(key)
+ .copied()
+ .unwrap_or(latest_version);
+
+ config_files
+ .iter()
+ .find(|config_file| config_file.config.version == active_version)
+ .map(|config_file| ConnectorConfigVersionInfo {
+ version: config_file.config.version,
+ created_at: config_file.created_at,
+ })
+ .map(|config| (key.clone(), config))
+ })
+ .collect();
+
+ let sources = all_configs
+ .sources_grouped_by_key()
+ .iter()
+ .filter_map(|(key, config_files)| {
+ if config_files.is_empty() {
+ return None;
+ }
+ let latest_version = config_files
+ .iter()
+ .map(|c| c.config.version)
+ .max()
+ .expect("At least one config version must exist");
+ let active_version = active_versions
+ .sources
+ .get(key)
+ .copied()
+ .unwrap_or(latest_version);
+
+ config_files
+ .iter()
+ .find(|config_file| config_file.config.version == active_version)
+ .map(|config_file| ConnectorConfigVersionInfo {
+ version: config_file.config.version,
+ created_at: config_file.created_at,
+ })
+ .map(|config| (key.clone(), config))
+ })
+ .collect();
+
+ Ok(ConnectorConfigVersions { sinks, sources })
+ }
+
+ async fn set_active_sink_version(&self, key: &str, version: u64) -> Result<(), RuntimeError> {
+ let connector_id = ConnectorId {
+ key: key.to_owned(),
+ version,
+ };
+ if self
+ .state
+ .connectors_config
+ .sinks()
+ .get(&connector_id)
+ .is_none()
+ {
+ return Err(RuntimeError::SinkConfigNotFound(key.to_owned(), version));
}
- if !std::fs::exists(&self.config_dir)? {
- warn!(
- "Connectors configuration directory does not exist: {}",
- self.config_dir
- );
- return Ok(ConnectorsConfig::default());
+
+ let mut active_versions = self.load_active_versions();
+ active_versions.sinks.insert(key.to_owned(), version);
+ self.save_active_versions(&active_versions)
+ }
+
+ async fn set_active_source_version(&self, key: &str, version: u64) -> Result<(), RuntimeError> {
+ let connector_id = ConnectorId {
+ key: key.to_owned(),
+ version,
+ };
+ if self
+ .state
+ .connectors_config
+ .sources()
+ .get(&connector_id)
+ .is_none()
+ {
+ return Err(RuntimeError::SourceConfigNotFound(key.to_owned(), version));
}
- let mut sinks = HashMap::new();
- let mut sources = HashMap::new();
- info!("Loading connectors configuration from: {}", self.config_dir);
- let entries = std::fs::read_dir(&self.config_dir)?;
- for entry in entries.flatten() {
- let path = entry.path();
- if path.is_file() {
- 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 {
- ConnectorConfig::Sink(sink_config) => {
- sinks.insert(base_config.id().to_owned(), sink_config);
- }
- ConnectorConfig::Source(source_config) => {
- sources.insert(base_config.id().to_owned(), source_config);
- }
- }
- }
+
+ let mut active_versions = self.load_active_versions();
+ active_versions.sources.insert(key.to_owned(), version);
+ self.save_active_versions(&active_versions)
+ }
+
+ async fn get_sink_configs(&self, key: &str) -> Result<Vec<SinkConfig>, RuntimeError> {
+ Ok(self
+ .state
+ .connectors_config
+ .sinks_grouped_by_key()
+ .get(key)
+ .cloned()
+ .unwrap_or_default()
+ .into_iter()
+ .map(|config_file| config_file.config)
+ .collect())
+ }
+
+ async fn get_sink_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<Option<SinkConfig>, RuntimeError> {
+ if let Some(version) = version {
+ let connector_id = ConnectorId {
+ key: key.to_owned(),
+ version,
+ };
+ Ok(self
+ .state
+ .connectors_config
+ .sinks()
+ .get(&connector_id)
+ .map(|entry| entry.config.clone()))
+ } else {
+ Ok(self
+ .get_sink_configs(key)
+ .await?
+ .into_iter()
+ .max_by_key(|config| config.version))
+ }
+ }
+
+ async fn get_source_configs(&self, key: &str) -> Result<Vec<SourceConfig>, RuntimeError> {
+ Ok(self
+ .state
+ .connectors_config
+ .sources_grouped_by_key()
+ .get(key)
+ .cloned()
+ .unwrap_or_default()
+ .into_iter()
+ .map(|config_file| config_file.config)
+ .collect())
+ }
+
+ async fn get_source_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<Option<SourceConfig>, RuntimeError> {
+ if let Some(version) = version {
+ let connector_id = ConnectorId {
+ key: key.to_owned(),
+ version,
+ };
+ Ok(self
+ .state
+ .connectors_config
+ .sources()
+ .get(&connector_id)
+ .map(|entry| entry.config.clone()))
+ } else {
+ Ok(self
+ .get_source_configs(key)
+ .await?
+ .into_iter()
+ .max_by_key(|config| config.version))
}
- Ok(ConnectorsConfig { sinks, sources })
+ }
+
+ async fn delete_sink_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<(), RuntimeError> {
+ debug!("Deleting sink config: {}@{:?}", &key, &version);
+ let sinks = self.state.connectors_config.sinks();
+ let active_versions = self.load_active_versions();
+
+ let version_to_delete = version
+ .or(active_versions.sinks.get(key).copied())
+ .ok_or_else(|| RuntimeError::SinkConfigNotFound(key.to_owned(), 0))?;
+
+ let connector_id = ConnectorId {
+ key: key.to_owned(),
+ version: version_to_delete,
+ };
+
+ let config_file = {
+ sinks
+ .get(&connector_id)
+ .ok_or_else(|| RuntimeError::SinkConfigNotFound(key.to_owned(), version_to_delete))?
+ .value()
+ .clone()
+ };
+
+ std::fs::remove_file(&config_file.path)?;
+ sinks.remove(&connector_id);
+
+ let mut active_versions = self.load_active_versions();
+ let remaining_versions: Vec<u64> = sinks
+ .iter()
+ .filter(|entry| entry.key().key == key)
+ .map(|entry| entry.key().version)
+ .collect();
+
+ if remaining_versions.is_empty() {
+ active_versions.sinks.remove(key);
+ } else if Some(version_to_delete) == active_versions.sinks.get(key).copied() {
+ let latest_version = remaining_versions
+ .into_iter()
+ .max()
+ .expect("At least one version must exist");
+ active_versions.sinks.insert(key.to_owned(), latest_version);
+ }
+
+ self.save_active_versions(&active_versions)?;
+ debug!("Deleted sink configuration: {}@{:?}", &key, &version);
+ Ok(())
+ }
+
+ async fn delete_source_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<(), RuntimeError> {
+ debug!("Deleting source config: {}@{:?}", &key, &version);
+ let sources = self.state.connectors_config.sources();
+ let active_versions = self.load_active_versions();
+
+ let version_to_delete = version
+ .or(active_versions.sources.get(key).copied())
+ .ok_or_else(|| RuntimeError::SourceConfigNotFound(key.to_owned(), 0))?;
+
+ let connector_id = ConnectorId {
+ key: key.to_owned(),
+ version: version_to_delete,
+ };
+
+ let config_file = {
+ sources
+ .get(&connector_id)
+ .ok_or_else(|| {
+ RuntimeError::SourceConfigNotFound(key.to_owned(), version_to_delete)
+ })?
+ .value()
+ .clone()
+ };
+
+ std::fs::remove_file(&config_file.path)?;
+ sources.remove(&connector_id);
+
+ let mut active_versions = self.load_active_versions();
+ let remaining_versions: Vec<u64> = sources
+ .iter()
+ .filter(|entry| entry.key().key == key)
+ .map(|entry| entry.key().version)
+ .collect();
+
+ if remaining_versions.is_empty() {
+ active_versions.sources.remove(key);
+ } else if Some(version_to_delete) == active_versions.sources.get(key).copied() {
+ let latest_version = remaining_versions
+ .into_iter()
+ .max()
+ .expect("At least one version must exist");
+ active_versions
+ .sources
+ .insert(key.to_owned(), latest_version);
+ }
+
+ self.save_active_versions(&active_versions)?;
+ debug!("Deleted source configuration: {}@{:?}", &key, &version);
+ Ok(())
}
}
@@ -143,10 +739,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..8378e702b 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,7 +61,9 @@ 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::SourceConfigNotFound(_, _) => "source_config_not_found",
RuntimeError::MissingIggyCredentials => "invalid_configuration",
RuntimeError::InvalidConfiguration(_) => "invalid_configuration",
_ => "error",
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
index dd975d48d..a266921b1 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -17,7 +17,7 @@
* under the License.
*/
-use crate::configs::connectors::ConnectorsConfigProvider;
+use crate::configs::connectors::{ConnectorsConfigProvider, create_connectors_config_provider};
use configs::connectors::ConfigFormat;
use configs::runtime::ConnectorsRuntimeConfig;
use dlopen2::wrapper::{Container, WrapperApi};
@@ -127,9 +127,9 @@ async fn main() -> Result<(), RuntimeError> {
let iggy_clients = stream::init(config.iggy.clone()).await?;
let connectors_config_provider: Box<dyn ConnectorsConfigProvider> =
- config.connectors.clone().into();
+ create_connectors_config_provider(&config.connectors).await?;
- let connectors_config = connectors_config_provider.load_configs().await?;
+ let connectors_config = connectors_config_provider.get_active_configs().await?;
let sources_config = connectors_config.sources();
let sources = source::init(
sources_config.clone(),
@@ -177,10 +177,11 @@ async fn main() -> Result<(), RuntimeError> {
let context = context::init(
&config,
- &sinks_config,
- &sources_config,
+ sinks_config,
+ 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]
diff --git a/core/connectors/runtime/example_config/connectors/stdout.toml b/core/integration/tests/connectors/postgres/config.toml
similarity index 64%
copy from core/connectors/runtime/example_config/connectors/stdout.toml
copy to core/integration/tests/connectors/postgres/config.toml
index 8e910f279..fff72e37e 100644
--- a/core/connectors/runtime/example_config/connectors/stdout.toml
+++ b/core/integration/tests/connectors/postgres/config.toml
@@ -15,27 +15,6 @@
# specific language governing permissions and limitations
# under the License.
-type = "sink"
-id = "stdout"
-enabled = true
-version = 0
-name = "Stdout sink"
-path = "target/release/libiggy_connector_stdout_sink"
-
-[[streams]]
-stream = "example_stream"
-topics = ["example_topic"]
-schema = "json"
-batch_length = 100
-poll_interval = "5ms"
-consumer_group = "stdout_sink_connector"
-
-[config]
-print_payload = false
-
-[transforms.add_fields]
-enabled = true
-
-[[transforms.add_fields.fields]]
-key = "message"
-value.static = "hello"
+[connectors]
+config_type = "local"
+config_dir = "postgres/connectors_config"
diff --git a/core/integration/tests/connectors/postgres/postgres.toml b/core/integration/tests/connectors/postgres/connectors_config/postgres.toml
similarity index 94%
rename from core/integration/tests/connectors/postgres/postgres.toml
rename to core/integration/tests/connectors/postgres/connectors_config/postgres.toml
index 24e84d4df..7602cd596 100644
--- a/core/integration/tests/connectors/postgres/postgres.toml
+++ b/core/integration/tests/connectors/postgres/connectors_config/postgres.toml
@@ -15,12 +15,14 @@
# specific language governing permissions and limitations
# under the License.
-[sinks.postgres]
+type = "sink"
+key = "postgres"
enabled = true
+version = 0
name = "Postgres sink"
path = "../../target/debug/libiggy_connector_postgres_sink"
-[[sinks.postgres.streams]]
+[[streams]]
stream = ""
topics = [""]
schema = "json"
@@ -28,7 +30,7 @@ batch_length = 100
poll_interval = "5ms"
consumer_group = "test"
-[sinks.postgres.config]
+[plugin_config]
connection_string = ""
target_table = "iggy_messages"
batch_size = 100
diff --git a/core/integration/tests/connectors/postgres/mod.rs b/core/integration/tests/connectors/postgres/mod.rs
index 8de202201..75acc741e 100644
--- a/core/integration/tests/connectors/postgres/mod.rs
+++ b/core/integration/tests/connectors/postgres/mod.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
@@ -37,24 +38,24 @@ async fn setup() -> IggyClient {
let iggy_setup = IggySetup::default();
let connection_string = format!("postgres://postgres:postgres@localhost:{host_port}");
envs.insert(
- "IGGY_CONNECTORS_SINKS_POSTGRES_CONFIG_CONNECTION_STRING".to_owned(),
+ "IGGY_CONNECTORS_SINK_POSTGRES_CONFIG_CONNECTION_STRING".to_owned(),
connection_string,
);
envs.insert(
- "IGGY_CONNECTORS_SINKS_POSTGRES_STREAMS_0_STREAM".to_owned(),
+ "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_STREAM".to_owned(),
iggy_setup.stream.to_owned(),
);
envs.insert(
- "IGGY_CONNECTORS_SINKS_POSTGRES_STREAMS_0_TOPICS_0".to_owned(),
+ "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_TOPICS_0".to_owned(),
iggy_setup.topic.to_owned(),
);
envs.insert(
- "IGGY_CONNECTORS_SINKS_POSTGRES_STREAMS_0_SCHEMA".to_owned(),
+ "IGGY_CONNECTORS_SINK_POSTGRES_STREAMS_0_SCHEMA".to_owned(),
"json".to_owned(),
);
let mut runtime = setup_runtime();
runtime
- .init("postgres/postgres.toml", Some(envs), iggy_setup)
+ .init("postgres/config.toml", Some(envs), iggy_setup)
.await;
runtime.create_client().await
}