Edge Time Series Processing

Orientation

Edge time series processing involves collecting, filtering, and aggregating timestamped data generated by sensors, devices, or system monitors at or near the source of production, before transmitting summaries to central backends. It matters when latency, bandwidth, or connectivity constraints make round‑trip central processing impractical—such as in remote installations, high‑frequency industrial telemetry, or mobile field deployments. This page walks through the practical workflow of setting up, configuring, and operating time series pipelines at the edge, with concrete commands, configuration keys, and the failure modes you’ll actually encounter in production.

1. Ingest time series at the edge

Most edge deployments begin with a protocol-level collector. The two most common entry points are MQTT (for constrained sensors) and HTTP/HTTPS (for local services or telemetry endpoints). Below are the exact commands and configuration patterns you’ll use to get data into a local time series store.

MQTT intake with Mosquitto

Start a lightweight Mosquitto broker tuned for edge retention policies:

mosquitto -c /etc/mosquitto/edge.conf -v

/etc/mosquitto/edge.conf:

listener 1883
log_type all both
log_destination /var/log/mosquitto/edge.log

Subscribe to a sensor topic and translate JSON payloads to InfluxDB line protocol in a single pipeline:

mosquitto_sub -h localhost -t "sensors/#" -C 50 | python3 -c "
import sys, json
for line in sys.stdin:
    try:
        payload = json.loads(line)
        ts = payload.get('ts') or int(payload.get('t', 0))
        val = float(payload.get('v', 0))
        print(f'sensor,location={payload.get(\"loc\",\"unknown\")} value={val} {ts}')
    except Exception:
        pass
" | influxd write -b edge-bucket -o http://localhost:8086

Sharp edge: If a payload omits the ts field, InfluxDB assigns the receive timestamp silently. You’ll see the runtime warning w! [inputs] timestamp out of order, adjusting when later querying by time range. This silently corrupts windowed aggregations downstream.

HTTP intake with Telegraf

When edge nodes expose metrics via HTTP, Telegraf’s inputs.http plugin pulls them at a configurable interval. Minimal telegraf.conf section:

[[outputs.influxdb_v2]]
  urls = ["http://edge-influx:8086"]
  token = "edge-token"
  organization = "edge-org"
  bucket = "telemetry"

[[inputs.http]]
  urls = ["http://edge-node:8080/metrics"]
  response_format = "influxline"
  data_format = "influxline"

Run with:

telegraf --config /etc/telegraf/edge.conf --debug

The --debug flag prints each received line. If you encounter error: invalid field type mapping, the field name collides with an InfluxDB reserved keyword (e.g., type, time, measurement). This collision produces a 400 Bad Request: field type conflict that is easy to miss in quiet log mode.

Exact flag reference:

telegraf --config <path> — explicit config path (required if not at default /etc/telegraf/telegraf.conf)
telegraf --debug — streams received line protocol to stdout for validation
influxd write -b <bucket> -o <url> — writes line protocol to a specified bucket/URL pair

2. Preprocess & aggregate locally

Once data is ingested, the next practical step is reducing volume before downstream shipment. Edge aggregation saves bandwidth and protects downstream stores from write spikes.

Continuous queries in InfluxDB

If you’re running InfluxDB at the edge, continuous queries (CQs) automate per‑interval downsampling. Create a CQ with exact keys:

CREATE CONTINUOUS QUERY cpu_avg_1m
  ON edge-bucket
  BEGIN
    SELECT MEAN(value) INTO cpu_avg_1m FROM /cpu.*/ GROUP BY time(1m) END

Configuration keys that control CQ behavior are set at the store level, not per‑query:

max_series_per_query — defaults to 1000; raise only if your downsampling pattern hits the series limit.
retention_policy — CQs target a specific RP; mis‑aligned RP names cause silent no‑op drops.

Exact error text: If the target RP doesn’t exist, CREATE CONTINUOUS QUERY fails with error: retention policy "autogen" does not exist. The query silently does nothing, and no downsampled data appears—you’ll only notice after a missing bucket in a dashboard.

Telegraf processors for on‑the‑fly transforms

Telegraf’s processor pipeline lets you drop, rename, or math‑transform fields before write. A typical processor block in telegraf.conf:

[[processors.rename]]
  tagdrop = ["__name_excluded"]
  fielddrop = ["raw_payload"]
  field_rename = { "temp_c": "temperature" }

Concrete code snippet (inline Python processor, rare but useful for edge kernels without Telegraf):

# /opt/edge/process_ts.py — executed by a cron or systemd timer
import sys, json
for line in sys.stdin:
    try:
        point = json.loads(line.strip())
        # Drop points older than 5m relative to now
        if int(point.get('ts', 0)) < int(time.time()) - 300:
            continue
        # Normalize field name
        point['value'] = round(float(point['value']), 2)
        print(json.dumps(point))
    except Exception:
        pass

Run the filter:

cat /var/log/mosquitto/edge.log | python3 /opt/edge/process_ts.py | influxd write -b edge-bucket

This page was rewritten on 10 October 2026. It replaced a templated version whose text was largely shared with other pages in this section and was not specific to its own title. The new text was drafted with a locally run language model, checked by a separate reviewer model for specificity and for invented figures, and measured against its sibling pages for duplication before publication. If anything here is wrong, tell us at [email protected] and we will correct it.