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/06/04 12:51:55 UTC
(iggy) branch connectors_docs updated (ccb52297 -> d6d7a7ea)
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a change to branch connectors_docs
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard ccb52297 Add connectors docs, WiP
new d6d7a7ea Add connectors docs, WiP
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (ccb52297)
\
N -- N -- N refs/heads/connectors_docs (d6d7a7ea)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/connectors/README.md | 2 +
core/connectors/runtime/README.md | 24 ++++
core/connectors/runtime/src/main.rs | 2 +-
core/connectors/sdk/README.md | 18 +++
core/connectors/sdk/src/transforms/README.md | 0
core/connectors/sinks/README.md | 4 +
core/connectors/sources/README.md | 173 +++++++++++++++++++++++++--
7 files changed, 209 insertions(+), 14 deletions(-)
delete mode 100644 core/connectors/sdk/src/transforms/README.md
(iggy) 01/01: Add connectors docs, WiP
Posted by pi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch connectors_docs
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit d6d7a7ea403d3035d5d7d099d2fcb5bcc2b7d9f2
Author: spetz <pi...@gmail.com>
AuthorDate: Tue Jun 3 21:29:48 2025 +0200
Add connectors docs, WiP
---
core/connectors/README.md | 47 ++------
core/connectors/runtime/README.md | 25 ++++
core/connectors/runtime/src/main.rs | 2 +-
core/connectors/sdk/README.md | 19 ++++
core/connectors/sinks/README.md | 5 +
core/connectors/sources/README.md | 220 ++++++++++++++++++++++++++++++++++++
6 files changed, 280 insertions(+), 38 deletions(-)
diff --git a/core/connectors/README.md b/core/connectors/README.md
index 3bbb1d1b..68051e50 100644
--- a/core/connectors/README.md
+++ b/core/connectors/README.md
@@ -34,58 +34,31 @@ iggy --username iggy --password iggy topic create qw records 1 none 1d
6. Start the connector runtime `cargo r --bin iggy_connector_runtime -r` - you should be able to browse Quickwit UI with records being constantly added to the `events` index. At the same time, you should see the new messages being added to the `example` stream and `topic1` topic by the test source connector - you can use Iggy Web UI to browse the data. The messages will have applied the basic fields transformations.
-## Building the connectors
+## Runtime
-New connector can be built simply by implementing either `Sink` or `Source` trait. Please check the existing examples under `core/connectors/sinks` and `core/connectors/sources` directories.
+All the connectors are implemented as Rust libraries and can be used as a part of the connector runtime. The runtime is responsible for managing the lifecycle of the connectors and providing the necessary infrastructure for the connectors to run. For more information, please refer to the **[runtime documentation](https://github.com/apache/iggy/tree/master/core/connectors/runtime)**.
-### Sink
+## Sink
Sinks are responsible for consuming the messages from the configured stream(s) and topic(s) and sending them further to the specified destination. For example, the Quickwit sink connector is responsible for sending the messages to the Quickwit indexer.
-```rust
-#[async_trait]
-pub trait Sink: Send + Sync {
- /// Invoked when the sink is initialized, allowing it to perform any necessary setup.
- async fn open(&mut self) -> Result<(), Error>;
-
- /// Invoked every time a batch of messages is received from the configured stream(s) and topic(s).
- async fn consume(
- &self,
- topic_metadata: &TopicMetadata,
- messages_metadata: MessagesMetadata,
- messages: Vec<ConsumedMessage>,
- ) -> Result<(), Error>;
-
- /// Invoked when the sink is closed, allowing it to perform any necessary cleanup.
- async fn close(&mut self) -> Result<(), Error>;
-}
-```
+Please refer to the **[Sink documentation](https://github.com/apache/iggy/tree/master/core/connectors/sinks)** for the details about the configuration and the sample implementation.
When implementing `Sink`, make sure to use the `sink_connector!` macro to expose the FFI interface and allow the connector runtime to register the sink with the runtime.
Each sink should have its own, custom configuration, which is passed along with the unique plugin ID via expected `new()` method.
-### Source
+## Source
Sources are responsible for producing the messages to the configured stream(s) and topic(s). For example, the Test source connector will generate the random messages that will be then sent to the configured stream and topic.
-```rust
-#[async_trait]
-pub trait Source: Send + Sync {
- /// Invoked when the source is initialized, allowing it to perform any necessary setup.
- async fn open(&mut self) -> Result<(), Error>;
+Please refer to the **[Source documentation](https://github.com/apache/iggy/tree/master/core/connectors/sources)** for the details about the configuration and the sample implementation.
- /// Invoked every time a batch of messages is produced to the configured stream and topic.
- async fn poll(&self) -> Result<ProducedMessages, Error>;
-
- /// Invoked when the source is closed, allowing it to perform any necessary cleanup.
- async fn close(&mut self) -> Result<(), Error>;
-}
-```
-
-When implementing `Source`, make sure to use the `source_connector!` macro to expose the FFI interface and allow the connector runtime to register the source with the runtime.
-Each source should have its own, custom configuration, which is passed along with the unique plugin ID via expected `new()` method.
+## Building the connectors
+New connector can be built simply by implementing either `Sink` or `Source` trait. Please check the **[sink](https://github.com/apache/iggy/tree/master/core/connectors/sinks)** or **[source](https://github.com/apache/iggy/tree/master/core/connectors/sources)** documentation, as well as the existing examples under `/sinks` and `/sources` directories.
## Transformations
Field transformations (depending on the supported payload formats) can be applied to the messages either before they are sent to the specified topic (e.g. when produced by the source connectors), or before consumed by the sink connectors. To add the new transformation, simply implement the `Transform` trait and extend the existing `load` function. Each transform may have its own, custom configuration.
+
+To find out more about the transforms, stream decoders or encoders, please refer to the **[SDK documentation](https://github.com/apache/iggy/tree/master/core/connectors/sdk)**.
diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md
new file mode 100644
index 00000000..23a35c11
--- /dev/null
+++ b/core/connectors/runtime/README.md
@@ -0,0 +1,25 @@
+# Apache Iggy Connectors - Runtime
+
+Runtime is responsible for managing the lifecycle of the connectors and providing the necessary infrastructure for the connectors to run.
+
+The runtime uses a shared [Tokio runtime](https://tokio.rs) to manage the asynchronous tasks and events across all connectors. Additionally, it has built-in support for logging via [tracing](https://docs.rs/tracing/latest/tracing/) crate.
+
+The connector are implemented as Rust libraries, and these are loaded dynamically during the runtime initialization process.
+
+Internally, [dlopen2](https://github.com/OpenByteDev/dlopen2) provides a safe and efficient way of loading the plugins via C FFI.
+
+By default, runtime will look for the configuration file, to decide which connectors to load and how to configure them.
+
+The minimal viable configuration requires at least the Iggy credentials, to create 2 separate instances of producer & consumer connections.
+
+```toml
+[iggy]
+address = "localhost:8090"
+username = "iggy"
+password = "iggy"
+# token = "secret" # Personal Access Token (PAT) can be used instead of username and password
+```
+
+All the other config sections start either with `sources` or `sinks` depending on the connector type.
+
+Keep in mind that either of `toml`, `yaml`, or `json` formats are supported for the configuration file. The path to the configuration can be overriden by `IGGY_CONNECTORS_RUNTIME_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.
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
index cbffc619..f87057bf 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -83,7 +83,7 @@ async fn main() -> Result<(), RuntimeError> {
info!("Starting Iggy Connector Runtime, loading configuration from: {config_path}...");
let builder = Config::builder()
.add_source(File::with_name(&config_path))
- .add_source(Environment::with_prefix("IGGY").separator("_"));
+ .add_source(Environment::with_prefix("IGGY_CONNECTORS").separator("_"));
let config: RuntimeConfig = builder
.build()
diff --git a/core/connectors/sdk/README.md b/core/connectors/sdk/README.md
new file mode 100644
index 00000000..13ee85fa
--- /dev/null
+++ b/core/connectors/sdk/README.md
@@ -0,0 +1,19 @@
+# Apache Iggy Connectors - SDK
+
+SDK provides the commonly used structs and traits such as `Sink` and `Source`, along with the `sink_connector` and `source_connector` macros to be used when developing connectors.
+
+Moreover, it contains both, the `decoders` and `encoders` modules, implementing either `StreamDecoder` or `StreamEncoder` traits, which are used when consuming or producing data from/to Iggy streams.
+
+SDK is WiP, and it'd certainly benefit from having the support of multiple format schemas, such as Protobuf, Avro, Flatbuffers etc. including decoding/encoding the data between the different formats (when applicable) and supporting the data transformations whenever possible (easy for JSON, but complex for Bincode for example).
+
+
+Last but not least, the different `transforms` are available, to transform (add, update, delete etc.) the particular fields of the data being processed via external configuration. It's as simple as adding a new transform to the `transforms` section of the particular connector configuration:
+
+```toml
+[sources.random.transforms.add_fields]
+enabled = true
+
+[[sources.random.transforms.add_fields.fields]]
+key = "message"
+value.static = "hello"
+```
diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md
new file mode 100644
index 00000000..310ee8d4
--- /dev/null
+++ b/core/connectors/sinks/README.md
@@ -0,0 +1,5 @@
+# Apache Iggy Connectors - Sink
+
+And that's all, enjoy using the sink connector!
+
+On a side note, if you'd like to produce the messages to the Iggy stream instead, you can implement your own **[Source connector](https://github.com/apache/iggy/tree/master/core/connectors/sources)** too :)
diff --git a/core/connectors/sources/README.md b/core/connectors/sources/README.md
new file mode 100644
index 00000000..d7add0d7
--- /dev/null
+++ b/core/connectors/sources/README.md
@@ -0,0 +1,220 @@
+# Apache Iggy Connectors - Source
+
+## Overview
+
+Source connectors are responsible for ingesting data from external sources into Apache Iggy. They provide a way to integrate Apache Iggy with various data sources, such as databases, message queues, or file systems.
+
+The source is represented by the single `Source` trait, which defines the basic interface for all source connectors. It provides methods for initializing the source, reading data from it, and closing the source.
+
+```rust
+#[async_trait]
+pub trait Source: Send + Sync {
+ /// Invoked when the source is initialized, allowing it to perform any necessary setup.
+ async fn open(&mut self) -> Result<(), Error>;
+
+ /// Invoked every time a batch of messages is produced to the configured stream and topic.
+ async fn poll(&self) -> Result<ProducedMessages, Error>;
+
+ /// Invoked when the source is closed, allowing it to perform any necessary cleanup.
+ async fn close(&mut self) -> Result<(), Error>;
+}
+```
+
+## Configuration
+
+Source is configured in the default `config` file used by runtime. Each source configuration, is part of the map of <String, SourceConfig>, which can be represented using toml, json, or yaml.
+
+```rust
+pub struct SourceConfig {
+ pub enabled: bool,
+ pub name: String,
+ pub path: String,
+ pub transforms: Option<TransformsConfig>,
+ pub streams: Vec<StreamProducerConfig>,
+ pub config: Option<serde_json::Value>,
+}
+```
+
+Below is the example configuration for a source connector, using `random` as it's unique identifier:
+
+```toml
+# Required configuration for a source connector
+[sources.random]
+enabled = true # Toggle source on/off
+name = "Random source" # Name of the source
+path = "libiggy_connector_random_source" # Path to the source connector
+
+# Collection of the streams to which the produced messages are sent
+[[sources.random.streams]]
+stream = "example_stream"
+topic = "example_topic"
+schema = "json"
+batch_size = 100
+send_interval = "5ms"
+
+# Custom configuration for the source connector, deserialized to type T from `config` field
+[sources.random.config]
+messages_count = 10
+
+# Optional data transformation(s) to be applied before sending messages to the stream
+[sources.random.transforms.add_fields]
+enabled = true
+
+# Collection of the fields transforms to be applied before sending messages to the stream
+[[sources.random.transforms.add_fields.fields]]
+key = "message"
+value.static = "hello"
+```
+
+## Sample implementation
+
+Let's implement the example source connector, which will simply generate the N random messages depending on the count specified in the configuration.
+
+Additionally, our source connector will have its own state, which can be used e.g. to track the overall progress or store some relevant information when producing the data from the actual external sources or tooling.
+
+Keep in mind, that the produces messages will be sent further to the specified stream, but it's already the responsibility of the runtime to handle the delivery.
+
+Alo, when implementing the source connector, make sure to use the `source_connector!` macro to expose the FFI interface and allow the connector runtime to register the source with the runtime.
+
+And finally, each source should have its own, custom configuration, which is passed along with the unique plugin ID via expected `new()` method.
+
+Let's start by defining the internal state and the public source connector along with its own configuration.
+
+```rust
+#[derive(Debug)]
+struct State {
+ current_id: usize,
+}
+```
+
+```rust
+#[derive(Debug)]
+pub struct RandomSource {
+ id: u32,
+ messages_count: u32,
+ state: Mutex<State>,
+}
+```
+
+```rust
+#[derive(Debug, Serialize, Deserialize)]
+pub struct RandomSourceConfig {
+ messages_count: Option<u32>,
+}
+```
+
+At this point, we can expose the expected `new()` method, which will be used by the runtime to create a new instance of the source connector. The `id` is assigned by the runtime, and represents the unique identifier of the source connector.
+
+```rust
+impl RandomSource {
+ pub fn new(id: u32, config: RandomSourceConfig) -> Self {
+ RandomSource {
+ id,
+ payload_size: config.payload_size.unwrap_or(100),
+ state: Mutex::new(State { current_id: 0 }),
+ }
+ }
+}
+```
+
+And we can invoke the expected macro to expose the FFI interface and allow the connector runtime to register the source within the runtime.
+
+```rust
+source_connector!(TestSource);
+```
+
+At a bare minimum, we need to add the following dependencies to the `Cargo.toml` file to compile the plugin at all:
+
+- dashmap
+- once_cell
+- tracing
+
+Before we make use of the `Source` trait, let's define the internal payload of the message that will be produced (e.g. as if it was pulled from some external database or so).
+
+```rust
+#[derive(Debug, Serialize, Deserialize)]
+struct Record {
+ id: u64,
+ text: String,
+}
+```
+
+Now, let's implement the `Source` trait for our `RandomSource` struct. We'll assume that the amount of messages (provided in the config), will be generated every 100ms to mimic the behavior of a real-world external source.
+
+```rust
+#[async_trait]
+impl Source for RandomSource {
+ async fn open(&mut self) -> Result<(), iggy_connector_sdk::Error> {
+ info!(
+ "Opened random source connector with ID: {}, messages count: {}",
+ self.id,
+ self.messages_count
+ );
+ Ok(())
+ }
+
+ async fn poll(&self) -> Result<ProducedMessages, iggy_connector_sdk::Error> {
+ sleep(Duration::from_millis(100)).await;
+ let mut state = self.state.lock().await;
+ let current_id = state.current_id;
+
+ let mut messages = Vec::new();
+ for _ in 0..self.messages_count {
+ current_id += 1;
+ let record = Record {
+ id: current_id,
+ text: format!("Hello from Random Source Connector: #{current_id}")
+ };
+ let Ok(payload) = simd_json::to_vec(&record) else {
+ panic!("Failed to serialize record");
+ };
+
+ let message = ProducedMessage {
+ id: None,
+ headers: None,
+ checksum: None,
+ timestamp: None,
+ origin_timestamp: None,
+ payload,
+ };
+ messages.push(message);
+ }
+
+ state.current_id += current_id;
+ info!(
+ "Generated {} messages by random source with ID: {}"
+ messages.len(),
+ self.id,
+ );
+ Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages,
+ })
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ info!("Random source connector with ID: {} is closed.", self.id);
+ Ok(())
+ }
+}
+```
+
+As you can see, the `ProducedMessage` can be customized to fit your needs, as all the fields will be directly mapped to the existing Iggy message struct.
+
+It's also important to note, that the supported format(s) might vary depending on the connector implementation. For example, you might use `JSON` as the payload format, which can be then easily parsed and processed by downstream components such as data transforms, but at the same time, you could support the other formats and let the user decide which one to use.
+
+While the final schema of messages (that will be appended to the Iggy stream), can be controlled with the the built-in configuration (the particular `StreamEncoder` will be used), keep in mind, that it might be sometimes difficult/impossible e.g. to transform one format to another e.g. JSON to SBE or so, and in such a case, the produced messages will be ignored.
+
+Eventually, compile the source code and update the runtime configuration file using the example config above (`config.toml` file by default, unless you prefer `yaml` or `json` format instead - just make sure that `path` points to the existing plugin).
+
+And before starting the runtime, do not forget to create the specified stream and topic e.g. via Iggy CLI.
+
+```
+iggy --username iggy --password iggy stream create example_stream
+
+iggy --username iggy --password iggy topic create example_stream example_topic 1 none 1d
+```
+
+And that's all, enjoy using the source connector!
+
+On a side note, if you'd like to process the messages consumed from the Iggy stream instead, you can implement your own **[Sink connector](https://github.com/apache/iggy/tree/master/core/connectors/sinks)** too :)