InfluxDB Sink

The InfluxDB sink connector consumes messages from Iggy streams and writes them to InfluxDB as line-protocol points. It supports both InfluxDB V2 (OSS 2.x / Cloud) and InfluxDB V3 (Core / Enterprise), selected with the version option.

This page is a curated subset of the documentation. The canonical reference is the upstream influxdb_sink README in the apache/iggy repository.

Configuration

Use the broker credentials and runtime setup from the sink guide. From the matching 0.9.0/edge repository root, build the plugin:

cargo build --release -p iggy_connector_influxdb_sink

Create the InfluxDB organization and bucket (V2), or database (V3), and provide a token with write access. Save the following connector file in the runtime's connector directory, replacing the URL and token for your deployment. Startup checks GET /health with the configured token, using Token authentication for V2 and Bearer for V3. A successful health check does not verify write permissions.

type = "sink"
key = "influxdb"
enabled = true
version = 0
name = "InfluxDB sink"
path = "target/release/libiggy_connector_influxdb_sink"
plugin_config_format = "toml"

[[streams]]
stream = "events"
topics = ["influx_events"]
schema = "json"
batch_length = 100
poll_interval = "5ms"
consumer_group = "influxdb_sink"

[plugin_config]
version = "v2"
url = "http://localhost:8086"
org = "iggy"
bucket = "events"
token = "replace_with_secret_token"
measurement = "iggy_messages"
precision = "us"
batch_size = 500

Create the Iggy resources and send a row:

./target/release/iggy --username iggy --password iggy stream create events
./target/release/iggy --username iggy --password iggy topic create events influx_events 1 none 1d
./target/release/iggy --username iggy --password iggy message send --partition-id 0 events influx_events '{"sensor":"temperature","value":21.5}'

InfluxDB V2 organizes data as org + bucket. V3 uses a single db field instead. Omitting version defaults to "v2" for backward compatibility. The write body, batching, and retry behavior are identical between versions. The connector handles the differing endpoints, auth header styles, and precision spellings internally.

For V3, replace the entire [plugin_config] section. Unknown options are rejected, so remove the V2 org and bucket fields:

# V3 variant: org/bucket are replaced by db
[plugin_config]
version = "v3"
url = "http://localhost:8181"
db = "my-db"
token = "replace_with_secret_token"

Common Options

The include_checksum, include_origin_timestamp, include_stream_tag, include_topic_tag, and include_partition_tag flags also default to true.

• message_id is always a string field and offset is always a tag.
• With include_metadata = true, stream, topic and partition are tags by default. Disabling an individual include_*_tag flag writes that value as an iggy_stream, iggy_topic, or iggy_partition field instead.
• include_metadata = false omits those three values. It does not disable the independently controlled iggy_checksum and iggy_origin_timestamp fields.

Point timestamps come from the Iggy message timestamp in microseconds, converted to precision. A zero timestamp uses the current wall-clock time. Millisecond and second precision discard finer timestamp digits.

InfluxDB identifies points by measurement/table, tags and timestamp. Removing stream, topic or partition tags can merge distinct messages that share an offset and timestamp; moving those values to fields does not preserve their identity. See the V2 and V3 duplicate-point rules.

Resilience options include timeout (30s), max_retries (3 total write attempts), retry_delay (1s), retry_max_delay (5s), max_open_retries (10 total health-check attempts), open_retry_max_delay (60s), circuit_breaker_threshold (5), and circuit_breaker_cool_down (30s). Invalid duration strings warn and fall back to 1s. See the upstream influxdb_sink README for the full list.

Payload Formats

• json (default): the message payload is validated as JSON, compact-serialized, and written as a single payload_json string field (individual JSON fields do not become separate field entries).
• text: the payload must be valid UTF-8 and is written as a single payload_text string field.
• base64: the payload bytes are base64-encoded into a payload_base64 string field.

payload_format selects the stored representation after stream decoding and transforms. For arbitrary bytes, use schema = "raw" with payload_format = "base64" and no payload-changing transform. For plain text, use schema = "text" with payload_format = "text". JSON stream decoding can change the original byte representation.

Text carriage returns and newlines become the literal sequences \r and \n in the stored string. Use base64 if those original bytes must round-trip. Tabs in measurement names or tag values are rejected; quoted text fields allow tabs.

Format names are case-insensitive; utf8 aliases text and raw aliases base64. An unrecognized payload_format logs a warning and falls back to json.

Reliability

Messages are serialized to line protocol and each polled batch is split into chunks of at most batch_size messages, one HTTP POST per chunk. The final partial chunk is written immediately; the connector does not accumulate messages across polls. A serialization error rejects its entire chunk. Later chunks are still attempted after a failure, and the first error is returned after the loop.

HTTP 429, all 5xx responses, and network errors are retried. max_retries = 3 allows the initial attempt plus two retries; 0 and 1 both allow one attempt. Backoff starts at retry_delay, doubles, adds ±20% jitter, and is capped by retry_max_delay. An integer-seconds Retry-After on a 429 overrides that cap; HTTP-date values are ignored. Startup health checks use a separate attempt budget and cap, retrying any failed check.

The circuit breaker counts at most one failure per consumed batch, based on its first error. A permanent HTTP error does not increment the counter; a fully successful batch resets it. While the breaker is open, incoming batches fail without a write. After the cool-down window, writes resume and the failure counter resets.

The runtime records a plugin error and continues polling with consumer auto-commit. Failed chunks and batches skipped by the circuit breaker are not queued for replay, so retries and the circuit breaker do not provide an end-to-end at-least-once guarantee. Other chunks from the same batch may already have reached InfluxDB.

Transforms

Transforms can be applied before writing to InfluxDB. See the transforms documentation for the available types and their configuration.