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 to manage the asynchronous tasks and events across all connectors. Additionally, it has built-in support for logging via tracing crate.

The connector are implemented as Rust libraries, and these are loaded dynamically during the runtime initialization process.

Internally, 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.

To start the connector runtime, simply run cargo run --bin iggy-connectors.

The docker image 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 configuration provider settings.

[iggy]
address = "localhost:8090"
username = "iggy"
password = "iggy"
token = "" # Personal Access Token (PAT) can be used instead of username and password

[iggy.tls] # Optional TLS configuration for Iggy TCP connection
enabled = false
ca_file = "core/certs/iggy_ca_cert.pem"
domain = "" # Optional domain for TLS connection

[state]
path = "local_state"

[connectors]
config_type = "local"
config_dir = "path/to/connectors"

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.

[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.

[connectors]
config_type = "http"
base_url = "http://localhost:8080/api"
timeout = "10s"

[connectors.request_headers]
api-key = "your-api-key"

[connectors.retry]
enabled = true
max_attempts = 3
initial_backoff = "1 s"
max_backoff = "30 s"
backoff_multiplier = 2

[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.

[http] # Optional HTTP API configuration
enabled = true
address = "127.0.0.1:8081"
api_key = "" # Optional API key for authentication to be passed as `api-key` header

[http.cors] # Optional CORS configuration for HTTP API
enabled = false
allowed_methods = ["GET", "POST", "PUT", "DELETE"]
allowed_origins = ["*"]
allowed_headers = ["content-type"]
exposed_headers = [""]
allow_credentials = false
allow_private_network = false

[http.metrics] # Optional Prometheus metrics configuration
enabled = false
endpoint = "/metrics"

[http.tls] # Optional TLS configuration for HTTP API
enabled = false
cert_file = "core/certs/iggy_cert.pem"
key_file = "core/certs/iggy_key.pem"

Currently, it does expose the following endpoints:

• GET /: welcome message.
• GET /health: health status of the runtime.
• GET /stats: runtime statistics including process info, memory/CPU usage, and connector status.
• GET /metrics: Prometheus-formatted metrics (when http.metrics.enabled is true).
• GET /sinks: list of sinks.
• GET /sinks/{key}: sink details.
• 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}/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.

Telemetry

The connector runtime supports OpenTelemetry for logs and traces. To enable telemetry, add the following configuration:

[telemetry]
enabled = true
service_name = "iggy-connectors"

[telemetry.logs]
transport = "grpc" # Options: "grpc", "http"
endpoint = "http://localhost:4317"

[telemetry.traces]
transport = "grpc" # Options: "grpc", "http"
endpoint = "http://localhost:4317"

Metrics

The runtime exposes Prometheus-compatible metrics via the /metrics endpoint when enabled. The following metrics are available:

Runtime Gauges

• iggy_connectors_sources_total: Total configured source connectors
• iggy_connectors_sources_running: Sources currently in Running status
• iggy_connectors_sinks_total: Total configured sink connectors
• iggy_connectors_sinks_running: Sinks currently in Running status

Per-Connector Counters (labeled with connector_key and connector_type)

• iggy_connector_messages_produced_total: Messages received from source plugin
• iggy_connector_messages_sent_total: Messages sent to Iggy (source)
• iggy_connector_messages_consumed_total: Messages consumed from Iggy (sink)
• iggy_connector_messages_processed_total: Messages processed and sent to sink plugin
• iggy_connector_errors_total: Errors encountered

Stats

The /stats endpoint provides runtime and per-connector statistics in JSON format. This includes system resource usage, connector counts, and detailed per-connector metrics with version information.

Example Response

{
  "connectors_runtime_version": "0.3.0",
  "connectors_runtime_version_semver": 203,
  "process_id": 12345,
  "cpu_usage": 2.5,
  "total_cpu_usage": 15.3,
  "memory_usage": 52428800,
  "total_memory": 17179869184,
  "available_memory": 8589934592,
  "run_time": 3600000000,
  "start_time": 1706889600000000,
  "sources_total": 1,
  "sources_running": 1,
  "sinks_total": 1,
  "sinks_running": 1,
  "connectors": [
    {
      "key": "random-source",
      "name": "Random Source",
      "connector_type": "source",
      "version": "0.3.0",
      "version_semver": 202,
      "status": "running",
      "enabled": true,
      "messages_produced": 1000,
      "messages_sent": 1000,
      "errors": 0
    },
    {
      "key": "stdout-sink",
      "name": "Stdout Sink",
      "connector_type": "sink",
      "version": "0.3.0",
      "version_semver": 202,
      "status": "running",
      "enabled": true,
      "messages_consumed": 1000,
      "messages_processed": 1000,
      "errors": 0
    }
  ]
}