You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2025/11/24 16:02:41 UTC
(iggy) branch master updated: feat(connectors): implement http configuration provider (#2401)
This is an automated email from the ASF dual-hosted git repository.
piotr 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 734cb3430 feat(connectors): implement http configuration provider (#2401)
734cb3430 is described below
commit 734cb34306428507e6ee63a931e02a92e2d85af5
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Mon Nov 24 17:02:37 2025 +0100
feat(connectors): implement http configuration provider (#2401)
---
Cargo.lock | 2 +
core/connectors/README.md | 3 +-
core/connectors/runtime/Cargo.toml | 2 +
core/connectors/runtime/README.md | 74 +++-
core/connectors/runtime/src/configs/connectors.rs | 12 +
.../src/configs/connectors/http_provider.rs | 433 +++++++++++++++++++++
.../connectors/http_provider/response_extractor.rs | 304 +++++++++++++++
.../connectors/http_provider/url_builder.rs | 263 +++++++++++++
core/connectors/runtime/src/configs/runtime.rs | 58 ++-
core/connectors/runtime/src/error.rs | 3 +
10 files changed, 1143 insertions(+), 11 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 76612eb86..b8a20690d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4735,8 +4735,10 @@ dependencies = [
"mimalloc",
"once_cell",
"postcard",
+ "reqwest",
"serde",
"serde_json",
+ "serde_with",
"serde_yaml_ng",
"strum 0.27.2",
"thiserror 2.0.17",
diff --git a/core/connectors/README.md b/core/connectors/README.md
index 6ffdc51ab..32ddc08a1 100644
--- a/core/connectors/README.md
+++ b/core/connectors/README.md
@@ -15,7 +15,8 @@ The [docker image](https://hub.docker.com/r/apache/iggy-connect) is available, a
- **Statically Typed**: Ensures type safety and compile-time checks, reducing runtime errors.
- **Easy Customization**: Provides a simple interface for implementing custom connectors, making it easy to create new plugins.
- **Data transformation**: Supports data transformation with the help of existing functions.
-- **Powerful configuration**: Define your sinks, sources, and transformations in the configuration file.
+- **Powerful configuration**: Define your sinks, sources, and transformations in the configuration file or fetch them from a remote HTTP API.
+- **Flexible configuration providers**: Support for local file-based and HTTP-based configuration providers for centralized configuration management.
## Quick Start
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index 146543cce..9f233fd97 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -46,8 +46,10 @@ iggy_connector_sdk = { workspace = true }
mimalloc = { workspace = true }
once_cell = { workspace = true }
postcard = { workspace = true }
+reqwest = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
+serde_with = { workspace = true }
serde_yaml_ng = { workspace = true }
strum = { workspace = true }
thiserror = { workspace = true }
diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md
index dec986b6b..5dc83c94f 100644
--- a/core/connectors/runtime/README.md
+++ b/core/connectors/runtime/README.md
@@ -14,7 +14,7 @@ To start the connector runtime, simply run `cargo run --bin iggy-connectors`.
The [docker image](https://hub.docker.com/r/apache/iggy-connect) is available, and can be fetched via `docker pull apache/iggy-connect`.
-The minimal viable configuration requires at least the Iggy credentials to create 2 separate instances of producer & consumer connections, the state directory path where source connectors can store their optional state, and the `connectors.config_dir` setting that specifies the directory containing individual connector configuration files.
+The minimal viable configuration requires at least the Iggy credentials to create 2 separate instances of producer & consumer connections, the state directory path where source connectors can store their optional state, and the connectors configuration provider settings.
```toml
[iggy]
@@ -36,10 +36,78 @@ config_type = "local"
config_dir = "path/to/connectors"
```
-Each connector (source or sink) is configured in its own separate file within the directory specified by `connectors.config_dir`. If `config_dir` is empty or the directory doesn't exist, no connectors will be loaded.
-
The path to the configuration can be overridden by `IGGY_CONNECTORS_CONFIG_PATH` environment variable. Each configuration section can be also additionally updated by using the following convention `IGGY_CONNECTORS_SECTION_NAME.KEY_NAME` e.g. `IGGY_CONNECTORS_IGGY_USERNAME` and so on.
+## Configuration Providers
+
+The runtime supports two types of configuration providers for managing connector configurations:
+
+### Local File Provider
+
+The default configuration provider reads connector configurations from local files. Each connector (source or sink) is configured in its own separate file within the directory specified by `connectors.config_dir`. If `config_dir` is empty or the directory doesn't exist, no connectors will be loaded.
+
+```toml
+[connectors]
+config_type = "local"
+config_dir = "path/to/connectors"
+```
+
+### HTTP Configuration Provider
+
+The HTTP configuration provider allows the runtime to fetch connector configurations from a remote HTTP/REST API. This enables centralized configuration management and dynamic configuration updates.
+
+```toml
+[connectors]
+config_type = "http"
+base_url = "http://localhost:8080/api"
+timeout = "10s"
+
+[connectors.request_headers]
+api-key = "your-api-key"
+
+[connectors.url_templates]
+# Optional: Customize URL templates for specific operations
+# If not specified, default RESTful URL patterns are used
+create_sink = "/sinks/{key}/configs"
+create_source = "/sources/{key}/configs"
+get_active_configs = "/configs/active"
+
+[connectors.response]
+# Optional: Extract data from nested response structures
+data_path = "data" # Path to data in response (e.g., {"data": {...}})
+error_path = "error" # Path to error in response (e.g., {"error": "..."})
+```
+
+#### Configuration Options
+
+- **base_url** (required): Base URL of the configuration API endpoint
+- **timeout** (optional): HTTP request timeout (default: 10s)
+- **request_headers** (optional): Custom headers to include in all HTTP requests (e.g., authentication headers)
+- **url_templates** (optional): Custom URL templates for API endpoints. Supports variable substitution with `{key}` and `{version}` placeholders.
+- **response.data_path** (optional): JSON path to extract response data from nested structures (e.g., "data.config")
+- **response.error_path** (optional): JSON path to check for errors in responses
+
+#### Default URL Templates
+
+If not customized, the HTTP provider uses the following RESTful URL patterns:
+
+- Create sink config: `POST {base_url}/sinks/{key}/configs`
+- Create source config: `POST {base_url}/sources/{key}/configs`
+- Get active configs: `GET {base_url}/configs/active`
+- Get active versions: `GET {base_url}/configs/active/versions`
+- Set active sink version: `PUT {base_url}/sinks/{key}/configs/active`
+- Set active source version: `PUT {base_url}/sources/{key}/configs/active`
+- Get sink configs: `GET {base_url}/sinks/{key}/configs`
+- Get sink config by version: `GET {base_url}/sinks/{key}/configs/{version}`
+- Get active sink config: `GET {base_url}/sinks/{key}/configs/active`
+- Get source configs: `GET {base_url}/sources/{key}/configs`
+- Get source config by version: `GET {base_url}/sources/{key}/configs/{version}`
+- Get active source config: `GET {base_url}/sources/{key}/configs/active`
+- Delete sink config: `DELETE {base_url}/sinks/{key}/configs`
+- Delete source config: `DELETE {base_url}/sources/{key}/configs`
+
+The HTTP provider expects the remote API to implement these endpoints and return connector configuration data in the same format as used by the local provider.
+
## HTTP API
Connector runtime has an optional HTTP API that can be enabled by setting the `enabled` flag to `true` in the `[http]` section.
diff --git a/core/connectors/runtime/src/configs/connectors.rs b/core/connectors/runtime/src/configs/connectors.rs
index 7d2502cd3..67af54dc9 100644
--- a/core/connectors/runtime/src/configs/connectors.rs
+++ b/core/connectors/runtime/src/configs/connectors.rs
@@ -17,8 +17,10 @@
* under the License.
*/
+pub mod http_provider;
mod local_provider;
+use crate::configs::connectors::http_provider::HttpConnectorsConfigProvider;
use crate::configs::connectors::local_provider::LocalConnectorsConfigProvider;
use crate::configs::runtime::ConnectorsConfig as RuntimeConnectorsConfig;
use crate::error::RuntimeError;
@@ -233,6 +235,16 @@ pub async fn create_connectors_config_provider(
let provider = provider.init().await?;
Ok(Box::new(provider))
}
+ RuntimeConnectorsConfig::Http(config) => {
+ let provider = HttpConnectorsConfigProvider::new(
+ &config.base_url,
+ config.timeout.get_duration(),
+ &config.request_headers,
+ &config.url_templates,
+ &config.response,
+ )?;
+ Ok(Box::new(provider))
+ }
}
}
diff --git a/core/connectors/runtime/src/configs/connectors/http_provider.rs b/core/connectors/runtime/src/configs/connectors/http_provider.rs
new file mode 100644
index 000000000..41618289d
--- /dev/null
+++ b/core/connectors/runtime/src/configs/connectors/http_provider.rs
@@ -0,0 +1,433 @@
+/*
+ * 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.
+ */
+
+mod response_extractor;
+mod url_builder;
+
+use crate::configs::connectors::http_provider::response_extractor::ResponseExtractor;
+use crate::configs::connectors::http_provider::url_builder::{TemplateKeys, UrlBuilder};
+use crate::configs::connectors::{
+ ConnectorConfigVersions, ConnectorsConfig, ConnectorsConfigProvider, CreateSinkConfig,
+ CreateSourceConfig, SinkConfig, SourceConfig,
+};
+use crate::configs::runtime::ResponseConfig;
+use crate::error::RuntimeError;
+use async_trait::async_trait;
+use reqwest;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+
+#[derive(Debug, Serialize, Deserialize)]
+struct SetActiveVersionRequest {
+ version: u64,
+}
+
+pub struct HttpConnectorsConfigProvider {
+ url_builder: UrlBuilder,
+ response_extractor: ResponseExtractor,
+ client: reqwest::Client,
+}
+
+impl HttpConnectorsConfigProvider {
+ pub fn new(
+ base_url: &str,
+ timeout: Duration,
+ request_headers: &HashMap<String, String>,
+ url_templates: &HashMap<String, String>,
+ response_config: &ResponseConfig,
+ ) -> Result<Self, RuntimeError> {
+ let mut headers = reqwest::header::HeaderMap::new();
+ for (key, value) in request_headers {
+ let header_name = reqwest::header::HeaderName::from_str(key).map_err(|err| {
+ RuntimeError::InvalidConfiguration(format!("Invalid header name '{key}': {err}"))
+ })?;
+ let header_value = reqwest::header::HeaderValue::from_str(value).map_err(|err| {
+ RuntimeError::InvalidConfiguration(format!(
+ "Invalid header value for '{key}': {err}"
+ ))
+ })?;
+ headers.insert(header_name, header_value);
+ }
+
+ let client = reqwest::Client::builder()
+ .default_headers(headers)
+ .timeout(timeout)
+ .build()
+ .map_err(|err| {
+ RuntimeError::InvalidConfiguration(format!("Failed to build HTTP client: {err}"))
+ })?;
+
+ let url_builder = UrlBuilder::new(base_url, url_templates);
+ let response_extractor = ResponseExtractor::new(response_config);
+
+ Ok(Self {
+ url_builder,
+ response_extractor,
+ client,
+ })
+ }
+
+ async fn check_response_status(
+ response: reqwest::Response,
+ ) -> Result<reqwest::Response, RuntimeError> {
+ if !response.status().is_success() {
+ let status = response.status();
+ let error_text = response
+ .text()
+ .await
+ .unwrap_or_else(|_| "Unknown error".to_string());
+ return Err(RuntimeError::HttpRequestFailed(format!(
+ "HTTP {status} - {error_text}"
+ )));
+ }
+ Ok(response)
+ }
+}
+
+#[async_trait]
+impl ConnectorsConfigProvider for HttpConnectorsConfigProvider {
+ async fn create_sink_config(
+ &self,
+ key: &str,
+ config: CreateSinkConfig,
+ ) -> Result<SinkConfig, RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+ let url = self.url_builder.build(TemplateKeys::CREATE_SINK, &vars);
+
+ let response = self
+ .client
+ .post(&url)
+ .json(&config)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ let response = Self::check_response_status(response).await?;
+
+ let response_text = response.text().await.map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!("Failed to read response: {err}"))
+ })?;
+
+ self.response_extractor.extract(&response_text)
+ }
+
+ async fn create_source_config(
+ &self,
+ key: &str,
+ config: CreateSourceConfig,
+ ) -> Result<SourceConfig, RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+ let url = self.url_builder.build(TemplateKeys::CREATE_SOURCE, &vars);
+
+ let response = self
+ .client
+ .post(&url)
+ .json(&config)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ let response = Self::check_response_status(response).await?;
+
+ let response_text = response.text().await.map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!("Failed to read response: {err}"))
+ })?;
+
+ self.response_extractor.extract(&response_text)
+ }
+
+ async fn get_active_configs(&self) -> Result<ConnectorsConfig, RuntimeError> {
+ let vars = HashMap::new();
+ let url = self
+ .url_builder
+ .build(TemplateKeys::GET_ACTIVE_CONFIGS, &vars);
+
+ let response = self
+ .client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ let response = Self::check_response_status(response).await?;
+
+ let response_text = response.text().await.map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!("Failed to read response: {err}"))
+ })?;
+
+ self.response_extractor.extract(&response_text)
+ }
+
+ async fn get_active_configs_versions(&self) -> Result<ConnectorConfigVersions, RuntimeError> {
+ let vars = HashMap::new();
+ let url = self
+ .url_builder
+ .build(TemplateKeys::GET_ACTIVE_VERSIONS, &vars);
+
+ let response = self
+ .client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ let response = Self::check_response_status(response).await?;
+
+ let response_text = response.text().await.map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!("Failed to read response: {err}"))
+ })?;
+
+ self.response_extractor.extract(&response_text)
+ }
+
+ async fn set_active_sink_version(&self, key: &str, version: u64) -> Result<(), RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+ let url = self.url_builder.build(TemplateKeys::SET_ACTIVE_SINK, &vars);
+
+ let request_body = SetActiveVersionRequest { version };
+ let response = self
+ .client
+ .put(&url)
+ .json(&request_body)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ Self::check_response_status(response).await?;
+
+ Ok(())
+ }
+
+ async fn set_active_source_version(&self, key: &str, version: u64) -> Result<(), RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+ let url = self
+ .url_builder
+ .build(TemplateKeys::SET_ACTIVE_SOURCE, &vars);
+
+ let request_body = SetActiveVersionRequest { version };
+ let response = self
+ .client
+ .put(&url)
+ .json(&request_body)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ Self::check_response_status(response).await?;
+
+ Ok(())
+ }
+
+ async fn get_sink_configs(&self, key: &str) -> Result<Vec<SinkConfig>, RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+ let url = self
+ .url_builder
+ .build(TemplateKeys::GET_SINK_CONFIGS, &vars);
+
+ let response = self
+ .client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ let response = Self::check_response_status(response).await?;
+
+ let response_text = response.text().await.map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!("Failed to read response: {err}"))
+ })?;
+
+ self.response_extractor.extract(&response_text)
+ }
+
+ async fn get_sink_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<Option<SinkConfig>, RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+
+ let url = match version {
+ Some(v) => {
+ let version_str = v.to_string();
+ vars.insert("version", &version_str);
+ self.url_builder.build(TemplateKeys::GET_SINK_CONFIG, &vars)
+ }
+ None => self
+ .url_builder
+ .build(TemplateKeys::GET_ACTIVE_SINK_CONFIG, &vars),
+ };
+
+ let response = self
+ .client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ if response.status() == reqwest::StatusCode::NOT_FOUND {
+ return Ok(None);
+ }
+
+ let response = Self::check_response_status(response).await?;
+
+ let response_text = response.text().await.map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!("Failed to read response: {err}"))
+ })?;
+
+ let config: SinkConfig = self.response_extractor.extract(&response_text)?;
+ Ok(Some(config))
+ }
+
+ async fn get_source_configs(&self, key: &str) -> Result<Vec<SourceConfig>, RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+ let url = self
+ .url_builder
+ .build(TemplateKeys::GET_SOURCE_CONFIGS, &vars);
+
+ let response = self
+ .client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ let response = Self::check_response_status(response).await?;
+
+ let response_text = response.text().await.map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!("Failed to read response: {err}"))
+ })?;
+
+ self.response_extractor.extract(&response_text)
+ }
+
+ async fn get_source_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<Option<SourceConfig>, RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+
+ let url = match version {
+ Some(v) => {
+ let version_str = v.to_string();
+ vars.insert("version", &version_str);
+ self.url_builder
+ .build(TemplateKeys::GET_SOURCE_CONFIG, &vars)
+ }
+ None => self
+ .url_builder
+ .build(TemplateKeys::GET_ACTIVE_SOURCE_CONFIG, &vars),
+ };
+
+ let response = self
+ .client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ if response.status() == reqwest::StatusCode::NOT_FOUND {
+ return Ok(None);
+ }
+
+ let response = Self::check_response_status(response).await?;
+
+ let response_text = response.text().await.map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!("Failed to read response: {err}"))
+ })?;
+
+ let config: SourceConfig = self.response_extractor.extract(&response_text)?;
+ Ok(Some(config))
+ }
+
+ async fn delete_sink_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<(), RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+
+ let mut query_params = HashMap::new();
+ let version = version.map(|v| v.to_string()).unwrap_or_default();
+ if !version.is_empty() {
+ query_params.insert("version", version.as_str());
+ }
+
+ let url = self.url_builder.build_with_query(
+ TemplateKeys::DELETE_SINK_CONFIG,
+ &vars,
+ &query_params,
+ );
+
+ let response = self
+ .client
+ .delete(&url)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ Self::check_response_status(response).await?;
+
+ Ok(())
+ }
+
+ async fn delete_source_config(
+ &self,
+ key: &str,
+ version: Option<u64>,
+ ) -> Result<(), RuntimeError> {
+ let mut vars = HashMap::new();
+ vars.insert("key", key);
+
+ let mut query_params = HashMap::new();
+ let version = version.map(|v| v.to_string()).unwrap_or_default();
+ if !version.is_empty() {
+ query_params.insert("version", version.as_str());
+ }
+
+ let url = self.url_builder.build_with_query(
+ TemplateKeys::DELETE_SOURCE_CONFIG,
+ &vars,
+ &query_params,
+ );
+
+ let response = self
+ .client
+ .delete(&url)
+ .send()
+ .await
+ .map_err(|err| RuntimeError::HttpRequestFailed(err.to_string()))?;
+
+ Self::check_response_status(response).await?;
+
+ Ok(())
+ }
+}
diff --git a/core/connectors/runtime/src/configs/connectors/http_provider/response_extractor.rs b/core/connectors/runtime/src/configs/connectors/http_provider/response_extractor.rs
new file mode 100644
index 000000000..b2c4357d5
--- /dev/null
+++ b/core/connectors/runtime/src/configs/connectors/http_provider/response_extractor.rs
@@ -0,0 +1,304 @@
+/*
+ * 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.
+ */
+
+use crate::configs::runtime::ResponseConfig;
+use crate::error::RuntimeError;
+use serde::de::DeserializeOwned;
+use serde_json::Value;
+
+/// Extracts data from JSON responses using configured paths
+pub struct ResponseExtractor {
+ data_path: Option<String>,
+ error_path: Option<String>,
+}
+
+impl ResponseExtractor {
+ /// Creates a new ResponseExtractor with optional configuration
+ pub fn new(config: &ResponseConfig) -> Self {
+ Self {
+ data_path: config.data_path.clone(),
+ error_path: config.error_path.clone(),
+ }
+ }
+
+ /// Extracts data from a JSON response body
+ ///
+ /// If data_path is configured, navigates to that path and deserializes.
+ /// Otherwise, deserializes the entire response body directly.
+ pub fn extract<T: DeserializeOwned>(&self, response_body: &str) -> Result<T, RuntimeError> {
+ let json_value: Value = serde_json::from_str(response_body).map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!("Failed to parse JSON response: {err}"))
+ })?;
+
+ if let Some(error_path) = &self.error_path
+ && let Some(error_value) = self.navigate_path(&json_value, error_path)
+ && !error_value.is_null()
+ {
+ let error_msg = match error_value {
+ Value::String(s) => s.clone(),
+ _ => error_value.to_string(),
+ };
+ return Err(RuntimeError::HttpRequestFailed(format!(
+ "API returned error: {error_msg}"
+ )));
+ }
+
+ let target_value = if let Some(data_path) = &self.data_path {
+ self.navigate_path(&json_value, data_path).ok_or_else(|| {
+ let structure = Self::summarize_structure(&json_value);
+ RuntimeError::HttpRequestFailed(format!(
+ "Data path '{data_path}' not found in response. Response structure: {structure}"
+ ))
+ })?
+ } else {
+ &json_value
+ };
+
+ serde_json::from_value(target_value.clone()).map_err(|err| {
+ RuntimeError::HttpRequestFailed(format!(
+ "Failed to deserialize data: {err}. Value: {target_value}"
+ ))
+ })
+ }
+
+ /// Navigates through a JSON structure using dot-notation path
+ ///
+ /// Example: "data.config" navigates to json["data"]["config"]
+ fn navigate_path<'a>(&self, json: &'a Value, path: &str) -> Option<&'a Value> {
+ let parts: Vec<&str> = path.split('.').collect();
+ let mut current = json;
+
+ for part in parts {
+ current = match current {
+ Value::Object(map) => map.get(part)?,
+ Value::Array(arr) => {
+ if let Ok(index) = part.parse::<usize>() {
+ arr.get(index)?
+ } else {
+ return None;
+ }
+ }
+ _ => return None,
+ };
+ }
+
+ Some(current)
+ }
+
+ /// Creates a summary of JSON structure for error messages
+ fn summarize_structure(json: &Value) -> String {
+ match json {
+ Value::Object(map) => {
+ let keys: Vec<&String> = map.keys().collect();
+ if keys.len() <= 5 {
+ format!(
+ "{{ {} }}",
+ keys.iter()
+ .map(|k| k.as_str())
+ .collect::<Vec<_>>()
+ .join(", ")
+ )
+ } else {
+ format!(
+ "{{ {} ... and {} more fields }}",
+ keys[..5]
+ .iter()
+ .map(|k| k.as_str())
+ .collect::<Vec<_>>()
+ .join(", "),
+ keys.len() - 5
+ )
+ }
+ }
+ Value::Array(arr) => format!("[array of {} items]", arr.len()),
+ Value::String(_) => "string".to_string(),
+ Value::Number(_) => "number".to_string(),
+ Value::Bool(_) => "bool".to_string(),
+ Value::Null => "null".to_string(),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use serde::{Deserialize, Serialize};
+
+ #[derive(Debug, Serialize, Deserialize, PartialEq)]
+ struct TestData {
+ name: String,
+ value: i32,
+ }
+
+ #[test]
+ fn test_direct_extraction_no_path() {
+ let extractor = ResponseExtractor::new(&ResponseConfig::default());
+ let response = r#"{"name": "test", "value": 42}"#;
+
+ let result: Result<TestData, _> = extractor.extract(response);
+ assert!(result.is_ok());
+ assert_eq!(
+ result.unwrap(),
+ TestData {
+ name: "test".to_string(),
+ value: 42
+ }
+ );
+ }
+
+ #[test]
+ fn test_nested_extraction_with_path() {
+ let config = ResponseConfig {
+ data_path: Some("data.config".to_string()),
+ error_path: None,
+ };
+ let extractor = ResponseExtractor::new(&config);
+ let response = r#"{"status": "ok", "data": {"config": {"name": "test", "value": 42}}}"#;
+
+ let result: Result<TestData, _> = extractor.extract(response);
+ assert!(result.is_ok());
+ assert_eq!(
+ result.unwrap(),
+ TestData {
+ name: "test".to_string(),
+ value: 42
+ }
+ );
+ }
+
+ #[test]
+ fn test_wrapped_response() {
+ let config = ResponseConfig {
+ data_path: Some("result".to_string()),
+ error_path: None,
+ };
+ let extractor = ResponseExtractor::new(&config);
+ let response = r#"{"result": {"name": "wrapped", "value": 99}}"#;
+
+ let result: Result<TestData, _> = extractor.extract(response);
+ assert!(result.is_ok());
+ assert_eq!(
+ result.unwrap(),
+ TestData {
+ name: "wrapped".to_string(),
+ value: 99
+ }
+ );
+ }
+
+ #[test]
+ fn test_missing_path_error() {
+ let config = ResponseConfig {
+ data_path: Some("missing.path".to_string()),
+ error_path: None,
+ };
+ let extractor = ResponseExtractor::new(&config);
+ let response = r#"{"status": "ok", "data": {"config": {}}}"#;
+
+ let result: Result<TestData, _> = extractor.extract(response);
+ assert!(result.is_err());
+ assert!(
+ result
+ .unwrap_err()
+ .to_string()
+ .contains("Data path 'missing.path' not found")
+ );
+ }
+
+ #[test]
+ fn test_error_path_detection() {
+ let config = ResponseConfig {
+ data_path: Some("data".to_string()),
+ error_path: Some("error.message".to_string()),
+ };
+ let extractor = ResponseExtractor::new(&config);
+ let response = r#"{"error": {"message": "Something went wrong"}, "data": null}"#;
+
+ let result: Result<TestData, _> = extractor.extract(response);
+ assert!(result.is_err());
+ assert!(
+ result
+ .unwrap_err()
+ .to_string()
+ .contains("API returned error: Something went wrong")
+ );
+ }
+
+ #[test]
+ fn test_error_path_null_is_ok() {
+ let config = ResponseConfig {
+ data_path: Some("data".to_string()),
+ error_path: Some("error".to_string()),
+ };
+ let extractor = ResponseExtractor::new(&config);
+ let response = r#"{"error": null, "data": {"name": "success", "value": 1}}"#;
+
+ let result: Result<TestData, _> = extractor.extract(response);
+ assert!(result.is_ok());
+ assert_eq!(
+ result.unwrap(),
+ TestData {
+ name: "success".to_string(),
+ value: 1
+ }
+ );
+ }
+
+ #[test]
+ fn test_deep_nested_path() {
+ let config = ResponseConfig {
+ data_path: Some("response.body.result.data".to_string()),
+ error_path: None,
+ };
+ let extractor = ResponseExtractor::new(&config);
+ let response =
+ r#"{"response": {"body": {"result": {"data": {"name": "deep", "value": 123}}}}}"#;
+
+ let result: Result<TestData, _> = extractor.extract(response);
+ assert!(result.is_ok());
+ assert_eq!(
+ result.unwrap(),
+ TestData {
+ name: "deep".to_string(),
+ value: 123
+ }
+ );
+ }
+
+ #[test]
+ fn test_array_index_access() {
+ let config = ResponseConfig {
+ data_path: Some("items.0".to_string()),
+ error_path: None,
+ };
+ let extractor = ResponseExtractor::new(&config);
+ let response =
+ r#"{"items": [{"name": "first", "value": 10}, {"name": "second", "value": 20}]}"#;
+
+ let result: Result<TestData, _> = extractor.extract(response);
+ assert!(result.is_ok());
+ assert_eq!(
+ result.unwrap(),
+ TestData {
+ name: "first".to_string(),
+ value: 10
+ }
+ );
+ }
+}
diff --git a/core/connectors/runtime/src/configs/connectors/http_provider/url_builder.rs b/core/connectors/runtime/src/configs/connectors/http_provider/url_builder.rs
new file mode 100644
index 000000000..0ca4bb24f
--- /dev/null
+++ b/core/connectors/runtime/src/configs/connectors/http_provider/url_builder.rs
@@ -0,0 +1,263 @@
+/*
+ * 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.
+ */
+
+use std::collections::HashMap;
+
+/// Default RESTful URL templates for connector operations
+pub struct DefaultUrls;
+
+impl DefaultUrls {
+ pub const CREATE_SINK: &'static str = "/sinks/{key}/configs";
+ pub const CREATE_SOURCE: &'static str = "/sources/{key}/configs";
+ pub const GET_ACTIVE_CONFIGS: &'static str = "/configs/active";
+ pub const GET_ACTIVE_VERSIONS: &'static str = "/configs/active/versions";
+ pub const SET_ACTIVE_SINK: &'static str = "/sinks/{key}/configs/active";
+ pub const SET_ACTIVE_SOURCE: &'static str = "/sources/{key}/configs/active";
+ pub const GET_SINK_CONFIGS: &'static str = "/sinks/{key}/configs";
+ pub const GET_SINK_CONFIG_BY_VERSION: &'static str = "/sinks/{key}/configs/{version}";
+ pub const GET_ACTIVE_SINK_CONFIG: &'static str = "/sinks/{key}/configs/active";
+ pub const GET_SOURCE_CONFIGS: &'static str = "/sources/{key}/configs";
+ pub const GET_SOURCE_CONFIG_BY_VERSION: &'static str = "/sources/{key}/configs/{version}";
+ pub const GET_ACTIVE_SOURCE_CONFIG: &'static str = "/sources/{key}/configs/active";
+ pub const DELETE_SINK_CONFIG: &'static str = "/sinks/{key}/configs";
+ pub const DELETE_SOURCE_CONFIG: &'static str = "/sources/{key}/configs";
+}
+
+/// Template operation names used as keys in the url_templates configuration
+pub struct TemplateKeys;
+
+impl TemplateKeys {
+ pub const CREATE_SINK: &'static str = "create_sink";
+ pub const CREATE_SOURCE: &'static str = "create_source";
+ pub const GET_ACTIVE_CONFIGS: &'static str = "get_active_configs";
+ pub const GET_ACTIVE_VERSIONS: &'static str = "get_active_versions";
+ pub const SET_ACTIVE_SINK: &'static str = "set_active_sink";
+ pub const SET_ACTIVE_SOURCE: &'static str = "set_active_source";
+ pub const GET_SINK_CONFIGS: &'static str = "get_sink_configs";
+ pub const GET_SINK_CONFIG: &'static str = "get_sink_config";
+ pub const GET_ACTIVE_SINK_CONFIG: &'static str = "get_active_sink_config";
+ pub const GET_SOURCE_CONFIGS: &'static str = "get_source_configs";
+ pub const GET_SOURCE_CONFIG: &'static str = "get_source_config";
+ pub const GET_ACTIVE_SOURCE_CONFIG: &'static str = "get_active_source_config";
+ pub const DELETE_SINK_CONFIG: &'static str = "delete_sink_config";
+ pub const DELETE_SOURCE_CONFIG: &'static str = "delete_source_config";
+}
+
+/// URL builder for constructing HTTP endpoints from templates
+pub struct UrlBuilder {
+ base_url: String,
+ templates: HashMap<String, String>,
+}
+
+impl UrlBuilder {
+ /// Creates a new UrlBuilder with optional custom templates
+ ///
+ /// If templates are not provided, uses RESTful defaults
+ pub fn new(base_url: &str, custom_templates: &HashMap<String, String>) -> Self {
+ let mut templates = HashMap::new();
+
+ // Initialize with defaults
+ templates.insert(
+ TemplateKeys::CREATE_SINK.to_string(),
+ DefaultUrls::CREATE_SINK.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::CREATE_SOURCE.to_string(),
+ DefaultUrls::CREATE_SOURCE.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::GET_ACTIVE_CONFIGS.to_string(),
+ DefaultUrls::GET_ACTIVE_CONFIGS.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::GET_ACTIVE_VERSIONS.to_string(),
+ DefaultUrls::GET_ACTIVE_VERSIONS.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::SET_ACTIVE_SINK.to_string(),
+ DefaultUrls::SET_ACTIVE_SINK.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::SET_ACTIVE_SOURCE.to_string(),
+ DefaultUrls::SET_ACTIVE_SOURCE.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::GET_SINK_CONFIGS.to_string(),
+ DefaultUrls::GET_SINK_CONFIGS.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::GET_SINK_CONFIG.to_string(),
+ DefaultUrls::GET_SINK_CONFIG_BY_VERSION.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::GET_ACTIVE_SINK_CONFIG.to_string(),
+ DefaultUrls::GET_ACTIVE_SINK_CONFIG.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::GET_SOURCE_CONFIGS.to_string(),
+ DefaultUrls::GET_SOURCE_CONFIGS.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::GET_SOURCE_CONFIG.to_string(),
+ DefaultUrls::GET_SOURCE_CONFIG_BY_VERSION.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::GET_ACTIVE_SOURCE_CONFIG.to_string(),
+ DefaultUrls::GET_ACTIVE_SOURCE_CONFIG.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::DELETE_SINK_CONFIG.to_string(),
+ DefaultUrls::DELETE_SINK_CONFIG.to_string(),
+ );
+ templates.insert(
+ TemplateKeys::DELETE_SOURCE_CONFIG.to_string(),
+ DefaultUrls::DELETE_SOURCE_CONFIG.to_string(),
+ );
+
+ // Override with custom templates if provided
+ for (key, value) in custom_templates.clone().into_iter() {
+ templates.insert(key, value);
+ }
+
+ Self {
+ base_url: base_url.to_owned(),
+ templates,
+ }
+ }
+
+ /// Builds a URL for the specified operation with variable substitution
+ pub fn build(&self, operation: &str, vars: &HashMap<&str, &str>) -> String {
+ let template = self
+ .templates
+ .get(operation)
+ .map(|s| s.as_str())
+ .unwrap_or("");
+
+ let mut url = template.to_string();
+
+ // Replace template variables: {key}, {version}, {type}
+ for (var_name, var_value) in vars {
+ let placeholder = format!("{{{}}}", var_name);
+ url = url.replace(&placeholder, var_value);
+ }
+
+ format!("{}{}", self.base_url, url)
+ }
+
+ /// Builds a URL with query parameters appended
+ pub fn build_with_query(
+ &self,
+ operation: &str,
+ vars: &HashMap<&str, &str>,
+ query_params: &HashMap<&str, &str>,
+ ) -> String {
+ let base = self.build(operation, vars);
+
+ if query_params.is_empty() {
+ return base;
+ }
+
+ let query_string: Vec<String> = query_params
+ .iter()
+ .map(|(k, v)| format!("{}={}", k, v))
+ .collect();
+
+ format!("{}?{}", base, query_string.join("&"))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_default_templates() {
+ let builder = UrlBuilder::new("http://localhost:8080", &HashMap::new());
+
+ let mut vars = HashMap::new();
+ vars.insert("key", "my-sink");
+
+ let url = builder.build(TemplateKeys::CREATE_SINK, &vars);
+ assert_eq!(url, "http://localhost:8080/sinks/my-sink/configs");
+ }
+
+ #[test]
+ fn test_custom_template() {
+ let mut custom = HashMap::new();
+ custom.insert(
+ TemplateKeys::CREATE_SINK.to_string(),
+ "/api/v2/sinks/{key}/config".to_string(),
+ );
+
+ let builder = UrlBuilder::new("http://localhost:8080", &custom);
+
+ let mut vars = HashMap::new();
+ vars.insert("key", "my-sink");
+
+ let url = builder.build(TemplateKeys::CREATE_SINK, &vars);
+ assert_eq!(url, "http://localhost:8080/api/v2/sinks/my-sink/config");
+ }
+
+ #[test]
+ fn test_version_substitution() {
+ let builder = UrlBuilder::new("http://localhost:8080", &HashMap::new());
+
+ let mut vars = HashMap::new();
+ vars.insert("key", "my-sink");
+ vars.insert("version", "42");
+
+ let url = builder.build(TemplateKeys::GET_SINK_CONFIG, &vars);
+ assert_eq!(url, "http://localhost:8080/sinks/my-sink/configs/42");
+ }
+
+ #[test]
+ fn test_query_parameters() {
+ let builder = UrlBuilder::new("http://localhost:8080", &HashMap::new());
+
+ let mut vars = HashMap::new();
+ vars.insert("key", "my-sink");
+
+ let mut query_params = HashMap::new();
+ query_params.insert("version", "5");
+
+ let url = builder.build_with_query(TemplateKeys::DELETE_SINK_CONFIG, &vars, &query_params);
+ assert_eq!(url, "http://localhost:8080/sinks/my-sink/configs?version=5");
+ }
+
+ #[test]
+ fn test_query_based_custom_template() {
+ let mut custom = HashMap::new();
+ custom.insert(
+ TemplateKeys::GET_SINK_CONFIG.to_string(),
+ "/api?action=get&key={key}&version={version}".to_string(),
+ );
+
+ let builder = UrlBuilder::new("http://localhost:8080", &custom);
+
+ let mut vars = HashMap::new();
+ vars.insert("key", "my-sink");
+ vars.insert("version", "10");
+
+ let url = builder.build(TemplateKeys::GET_SINK_CONFIG, &vars);
+ assert_eq!(
+ url,
+ "http://localhost:8080/api?action=get&key=my-sink&version=10"
+ );
+ }
+}
diff --git a/core/connectors/runtime/src/configs/runtime.rs b/core/connectors/runtime/src/configs/runtime.rs
index 1375a3a55..720204f07 100644
--- a/core/connectors/runtime/src/configs/runtime.rs
+++ b/core/connectors/runtime/src/configs/runtime.rs
@@ -22,9 +22,11 @@ use figment::providers::{Format, Toml};
use figment::value::Dict;
use figment::{Metadata, Profile, Provider};
use iggy_common::defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
-use iggy_common::{CustomEnvProvider, FileConfigProvider};
+use iggy_common::{CustomEnvProvider, FileConfigProvider, IggyDuration};
use serde::{Deserialize, Serialize};
-use std::fmt::Formatter;
+use serde_with::{DisplayFromStr, serde_as};
+use std::collections::HashMap;
+use std::fmt::{Display, Formatter};
#[derive(Debug, Default, Clone, Deserialize, Serialize)]
#[serde(default)]
@@ -57,10 +59,51 @@ pub struct LocalConnectorsConfig {
pub config_dir: String,
}
+#[serde_as]
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+pub struct HttpConnectorsConfig {
+ pub base_url: String,
+ #[serde_as(as = "DisplayFromStr")]
+ #[serde(default = "default_from_secs")]
+ pub timeout: IggyDuration,
+ #[serde(default)]
+ pub request_headers: HashMap<String, String>,
+ #[serde(default)]
+ pub url_templates: HashMap<String, String>,
+ #[serde(default)]
+ pub response: ResponseConfig,
+}
+
+impl Display for HttpConnectorsConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ type: \"http\", base_url: {:?}, request_headers: {:?}, timeout: {}, url_templates: {:?}, response: {:?} }}",
+ self.base_url,
+ self.request_headers.keys(),
+ self.timeout,
+ self.url_templates,
+ self.response
+ )
+ }
+}
+
+fn default_from_secs() -> IggyDuration {
+ IggyDuration::new_from_secs(10)
+}
+
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[serde(default)]
+pub struct ResponseConfig {
+ pub data_path: Option<String>,
+ pub error_path: Option<String>,
+}
+
#[derive(Debug, Clone, Deserialize, Serialize)]
#[serde(tag = "config_type", rename_all = "lowercase")]
pub enum ConnectorsConfig {
Local(LocalConnectorsConfig),
+ Http(HttpConnectorsConfig),
}
impl Default for ConnectorsConfig {
@@ -74,13 +117,13 @@ pub struct StateConfig {
pub path: String,
}
-impl std::fmt::Display for StateConfig {
+impl Display for StateConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "{{ path: {} }}", self.path)
}
}
-impl std::fmt::Display for ConnectorsRuntimeConfig {
+impl Display for ConnectorsRuntimeConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
@@ -90,7 +133,7 @@ impl std::fmt::Display for ConnectorsRuntimeConfig {
}
}
-impl std::fmt::Display for IggyConfig {
+impl Display for IggyConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
@@ -108,7 +151,7 @@ impl std::fmt::Display for IggyConfig {
}
}
-impl std::fmt::Display for IggyTlsConfig {
+impl Display for IggyTlsConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
@@ -118,7 +161,7 @@ impl std::fmt::Display for IggyTlsConfig {
}
}
-impl std::fmt::Display for ConnectorsConfig {
+impl Display for ConnectorsConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
ConnectorsConfig::Local(config) => write!(
@@ -126,6 +169,7 @@ impl std::fmt::Display for ConnectorsConfig {
"{{ type: \"file\", config_dir: {:?} }}",
config.config_dir
),
+ ConnectorsConfig::Http(config) => write!(f, "{config}",),
}
}
}
diff --git a/core/connectors/runtime/src/error.rs b/core/connectors/runtime/src/error.rs
index 8378e702b..b392b60b6 100644
--- a/core/connectors/runtime/src/error.rs
+++ b/core/connectors/runtime/src/error.rs
@@ -55,6 +55,8 @@ pub enum RuntimeError {
CannotConvertConfiguration,
#[error("IO operation failed with error: {0:?}")]
IoError(#[from] std::io::Error),
+ #[error("HTTP request failed: {0}")]
+ HttpRequestFailed(String),
}
impl RuntimeError {
@@ -66,6 +68,7 @@ impl RuntimeError {
RuntimeError::SourceConfigNotFound(_, _) => "source_config_not_found",
RuntimeError::MissingIggyCredentials => "invalid_configuration",
RuntimeError::InvalidConfiguration(_) => "invalid_configuration",
+ RuntimeError::HttpRequestFailed(_) => "http_request_failed",
_ => "error",
}
}