Live streams

Collector: InfluxDB

Read measurements from InfluxDB 2.x (Flux) or 1.x (InfluxQL), incrementally and for any past range.

InfluxDB

Reads time series out of an InfluxDB that already holds them (Telegraf, Node-RED and historian-style stacks), with the Flux API of InfluxDB 2.x or the InfluxQL API of 1.x. Every few seconds it asks for the points newer than the last one it sent, so nothing is repeated or lost across restarts, and backfill reads any past range.

collector.yaml

sources:
  - name: Customer InfluxDB 2
    type: influx
    url: https://influx.customer.local:8086
    api: flux                      # InfluxDB 2.x
    token: env:INFLUX_TOKEN        # a read-only API token
    org: customer
    bucket: telemetry
    series:
      - measurement: cpu
        fields: [usage, temp]
        where: {dc: line1}         # tag = value filters
        group_by: [host]           # one channel per host

  - name: Legacy InfluxDB 1
    type: influx
    url: https://influx1.customer.local:8086
    api: influxql                  # InfluxDB 1.x (and the v1 endpoint of 3.x)
    database: telemetry
    user: reader
    password: env:INFLUX_PASSWORD
    series:
      - {measurement: pump, fields: [flow, pressure], group_by: [unit]}

Name	Type	Description
`url`required	`string`	The server. https:// by default; http:// needs allow_http.
`api`	`flux \| influxql` Default: `flux`	flux: POST /api/v2/query. influxql: GET /query.
`token, org, bucket`	`string`	flux: all three required. Use a read-only token (env:VAR or file:path).
`database, rp, user, password`	`string`	influxql: database required; rp is the retention policy; user and password when authentication is on.
`series`required	`list`	What to read. Each entry: measurement, fields (numbers or booleans), where (tag filters, equality) and group_by (tags that split the measurement into channels).
`poll_ms`	`number` Default: `5000`	Polling interval.
`overlap_s`	`number` Default: `60`	Each series keeps its own watermark, which trails its newest point by this long, so a writer that flushes later than another (Telegraf flushes every 10 s by default, plus jitter) is still read; duplicates are dropped. A larger value tolerates later writers but re-reads more on every poll; a point later than this is lost.
`lookback_s`	`number` Default: `0`	On the first start, how far back to read (otherwise just the last poll interval).
`history_chunk_s`	`number` Default: `3600`	No query asks for more than this: backfill and the catch-up after downtime walk the range in windows of this size, through empty ones, until they reach now.
`discover_s, max_tags`	`number` Default: `3600, 5000`	discover lists the channels that have points in the last discover_s seconds; channels beyond max_tags are dropped and counted.
`ca, insecure, allow_http, timeout_s`	`mixed`	As in the REST source.

Channels are named measurement.field, plus ,tag=value for each group_by tag (cpu.usage,host=a). String fields are dropped and counted (strings_dropped). An InfluxQL answer cut short by the server's max-row-limit ("partial": true) is never taken as complete: the window is halved and asked again (partial_results); if even one second is cut, the log says rows were lost. Health reports connected (1 after a successful query, 0 after a failed one). Without group_by, series that differ only by tags share one channel, so list the tags that tell them apart.