InfluxDB Source

The InfluxDB source connector polls InfluxDB on an interval and produces the resulting rows as messages into an Iggy stream. It supports both InfluxDB V2 (OSS 2.x / Cloud, Flux queries) and InfluxDB V3 (Core / Enterprise, SQL queries), selected with the version option. An acknowledged timestamp cursor and row offset are stored in connector state. Restarts resume from that checkpoint, subject to the ordering and late-arrival limits below.

This page is a curated subset of the documentation. The canonical reference, including the full cursor semantics and stuck-timestamp handling, is the upstream influxdb_source README in the apache/iggy repository.

Configuration

Use the broker credentials and main runtime configuration from the source guide. Build the plugin from the matching 0.9.0/edge checkout root:

cargo build --release -p iggy_connector_influxdb_source

The V2 example requires an existing telemetry bucket in iggy_org, a token with read access, and cpu points with a numeric usage field and a host tag. Save this entry in the runtime's connector directory and start the runtime from the checkout root.

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

[[streams]]
stream = "events"
topic = "influx_events"
schema = "json"
batch_length = 100

[plugin_config]
version = "v2"
url = "http://localhost:8086"
org = "iggy_org"
token = "replace_with_secret_token"
query = '''
from(bucket: "telemetry")
  |> range(start: time(v: "$cursor"))
  |> filter(fn: (r) => r._measurement == "cpu" and r._field == "usage")
  |> group(columns: [])
  |> sort(columns: ["_time", "host"])
  |> limit(n: $limit)
'''
poll_interval = "5s"
batch_size = 500

Create the destination before starting the runtime:

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

InfluxDB V2 takes an org and a Flux query. For V3, replace the entire [plugin_config] section below, including the V2-only org. The my-db database must contain a cpu table; this example assumes host is its only tag.

# V3 variant
[plugin_config]
version = "v3"
url = "http://localhost:8181"
db = "my-db"
token = "replace_with_secret_token"
query = '''
SELECT * FROM cpu
WHERE time > '$cursor'
ORDER BY time, host
LIMIT $limit OFFSET $offset
'''

Omitting the plugin version defaults to "v2" for backward compatibility; it is separate from the connector entry's numeric configuration version. Unknown plugin keys and unsupported versions are rejected. The query must contain $cursor outside comments. Keep $limit in the query so the connector can control its size. With stuck-batch detection enabled, V3 also requires $offset and an ascending ORDER BY. Startup checks use text matching, so they do not prove that a query implements the required ordering.

Start the runtime as described in the source guide. After a polling cycle, read the produced records:

./target/release/iggy --username iggy --password iggy message poll --offset 0 events influx_events 0

Common Options

payload_column extracts one column. Without it, every payload_format setting emits whole-row JSON. With it, the selected json, text or raw format applies, and the destination streams.schema should match. Selecting a column that is missing from a row causes a poll error.

Resilience defaults are timeout = "10s", max_retries = 3, retry_delay = "1s", retry_max_delay = "5s", max_open_retries = 10, open_retry_max_delay = "60s", circuit_breaker_threshold = 5, and circuit_breaker_cool_down = "30s". Attempt counts include the first attempt and are clamped to at least one. Query retries cover network errors, HTTP 429 and 5xx, with exponential backoff and jitter; an integer-seconds Retry-After on 429 can override the delay cap. Other HTTP failures and malformed successful responses return poll errors. Invalid duration strings warn and fall back to 1s; zero durations are accepted. verbose_logging defaults to false.

Cursor-Based Polling

The connector substitutes its RFC3339 cursor into the query. Both versions require ascending timestamps and a stable, unique order within each timestamp group. Include all necessary tag or key columns after the timestamp in the sort. The examples use host because their data model has one point per host and timestamp.

• V2 uses an inclusive cursor, such as range(start: time(v: "$cursor")). Flux sorts and limits each table separately, so the example groups its selected numeric field into one table before sorting. A persisted count skips rows already acknowledged at the cursor timestamp. $limit is the base batch size plus that count, capped at eleven times the base size. Dense timestamp groups can exhaust the skip allowance and produce a stuck-cursor error without advancing the cursor. A strict > query can lose unseen ties and is unsuitable for this skip-count scheme.
• V3 requires strict WHERE time > '$cursor'; inclusive >= templates are rejected. A full batch containing one timestamp retains the previous cursor, advances $offset, and doubles the next query size up to the configured cap. A full batch containing several timestamps defers the final timestamp group to the next poll. Reaching the inflation cap emits no messages, resets the effective size to the base size, preserves the offset, and records a circuit-breaker failure. The breaker opens only when its failure threshold is reached. Setting the cap factor to 0 disables these guards and can skip unseen tied rows.

A timestamp watermark is not change-data capture. Late inserts, edits and backdated points at or below an acknowledged watermark can be missed, and changing the order of tied rows invalidates offset-based progress. Neither version tracks deletions. The connector's timestamp-plus-position message IDs can also collide for distinct rows across batches; they are not unique database-row keys.

Payload Formats

• Whole-row json: V3 emits a flat object with native JSON types; include_metadata = false removes only the cursor column. V2 emits an envelope with measurement, field, timestamp, value and row. Its parser reads CSV cells as strings and ignores the #datatype and #default values. Payload conversion tries bool, integer and finite float, then string; an empty cell becomes null. Thus a string field containing "42" becomes the number 42 in V2 but stays a string in V3. With metadata disabled, V2's row retains only _time and _value; the envelope remains.
• Selected-column json: V2 parses the cell as JSON. V3 serializes the column's existing JSON value, so a string containing JSON remains a JSON string.
• Selected-column text: V2 emits the cell as UTF-8. V3 emits strings directly and serializes other JSON values as text.
• Selected-column raw: the column must contain standard base64 text, which is decoded into raw bytes.

Missing selected columns, invalid JSON cells in V2, and invalid base64 reject the poll. V3 also rejects any row with a missing, non-string or invalid cursor. V2 can emit rows without valid cursors if another row supplies a usable watermark; keep the cursor column in every row. V3 timestamps without a timezone suffix are treated as UTC, preserving nanoseconds. Keep database timestamps in the payload when consumers need them; the runtime does not transfer the plugin's timestamp fields into broker message metadata.

State and Failures

Successful polls return a versioned MessagePack checkpoint, including empty polls. After forwarding messages, the runtime saves it and sends Ack; only then does the plugin apply its candidate cursor and row count. Nack discards that candidate so the next poll can fetch the records again. A failure after forwarding but before acknowledgement can produce duplicates.

The default file backend stores source_influxdb.state under the runtime's state path (default local_state). The runtime also supports an HTTP state backend. V2 restores versioned state or its legacy unversioned state; V3 requires versioned V3 state. Corrupt checkpoints, invalid saved timestamps and mismatched versions prevent startup. To reset file-backed progress, stop the runtime and remove the source's checkpoint, then restart with the desired initial_offset.

Query and parsing failures leave the acknowledged cursor unchanged and are logged as poll errors. They do not increment the runtime's forwarding-error counter or change its running status. While its circuit breaker is open, the plugin waits and returns empty batches without checkpoints. On cooldown expiry it permits another query. A V3 response reporting that the database was not found is treated as an empty result, so create the database before starting the source.