Google Git
Sign in
apache/iggy/refs/heads/fix_encryption/./core/connectors/runtime
tree: ca5eef716030b40277d0497620c9d4f9142b7eeb
  1. example_config/
  2. src/
  3. .dockerignore
  4. Cargo.toml
  5. config.toml
  6. Dockerfile
  7. README.md
  8. runtime.http
core/connectors/runtime/README.md

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"

[logging] # Optional logging configuration
format = "text" # Options: "text" (default), "json"

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.

Logging

By default, the runtime emits human-readable text logs via the tracing crate. Switching logging.format to json produces structured JSON lines (one event per line, machine-parseable). When telemetry is enabled, the chosen format applies to the local stdout layer; OpenTelemetry export is unaffected.

[logging]
format = "json"

RUST_LOG controls verbosity and per-target filtering (e.g. RUST_LOG=info,iggy_connectors::benchmark=info).

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"

Benchmark Mode

Each connector configuration accepts an optional benchmark flag. When set to true, the runtime emits a per-batch info! event on the iggy_connectors::benchmark tracing target with stage timings in microseconds. This is opt-in and adds a single tracing call per processed batch.

type = "sink"
key = "stdout"
# ... other fields ...
benchmark = true

Emitted fields:

  • sink: connector_key, stream, topic, partition_id, current_offset, batch_size, processed_count, decode_us, prepare_us, ffi_us, total_us
  • source: connector_key, stream, topic, batch_size, sent_count, decode_us, prepare_us, iggy_send_us, state_saved, state_save_us, total_us (state_save_us is 0 when state_saved is false)

Filter the stream via RUST_LOG=iggy_connectors::benchmark=info. The corresponding stage durations are also recorded in the iggy_connector_stage_duration_seconds histogram regardless of this flag, so Prometheus dashboards remain available without enabling text events.

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_messages_filtered_total: Messages intentionally dropped by transforms returning Ok(None)
  • iggy_connector_errors_total: Errors encountered

Stage Duration Histograms (labeled with connector_key, connector_type, stage)

  • iggy_connector_stage_duration_seconds: Per-batch processing stage duration

Stage label values (snake_case): decode, prepare, total on both sides; ffi on sinks; iggy_send, state_save on sources. The connector_type label is also snake_case (source / sink). Histograms are always exposed (independent of benchmark); buckets cover 50us to 5s.

Cardinality: each (connector_key, stage) pair yields ~16 series (13 explicit buckets + +Inf + _sum + _count), times 3 stages per sink or 5 per source, times the number of distinct connector_key values. connector_key comes from operator config, so keep it bounded. Do not template per-user or per-request keys into it or the time-series database will blow up.

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,
      "messages_filtered": 0,
      "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,
      "messages_filtered": 0,
      "errors": 0
    }
  ]
}