Automating bucket expiration with scheduled purge tasks
You run home_telemetry, a single bucket collecting thermostat, energy, and door measurements from a consumer smart-home fleet with a 13-month retention window agreed for year-over-year comparisons. Then two things land in the same week: a customer files an erasure request for an RMA’d hub with device_id="d29f-rma" that must be gone within 24 hours, and a beta firmware is found to have shipped a junk debug_beacon measurement into production. Bucket retention cannot help with either — it only expires whole time windows once they age past the boundary, identically for every series. It has no way to drop one measurement early, or to erase one device’s history while keeping everyone else’s. The escape hatch is the delete API: predicate-based deletion that removes data by measurement or tag value, at any age, on a schedule you control. This page builds a safe, idempotent purge job around it, sizing the blast radius before deleting and running under a least-privilege token. It extends the whole-bucket expiry model from Retention Policy Design with selective, predicate-driven purges.
The gap is worth stating precisely. Retention answers “how long”; a purge answers “which rows”. Left to retention alone, the worst-case erasure latency for a decommissioned device equals the full window, whereas a scheduled purge bounds it to the run interval:
$$ T_{erasure} \le P_{purge} + \delta_{run} \ll R_{bucket} $$
For a GDPR-style deletion SLA, that difference is the difference between hours and thirteen months.
Prerequisites
How predicate purge differs from retention
Retention and predicate purge act on two orthogonal axes. Retention sweeps the time axis, dropping the oldest shard groups for every series at once. A predicate purge sweeps the identity axis, dropping a chosen measurement or tag value across every age. Understanding that they are independent is what stops engineers from trying to force one to do the other’s job.
Solution walkthrough
1. Measure the blast radius with a Flux dry run
Never issue a delete you have not counted first. A predicate delete is irreversible and has no built-in confirmation, so the discipline is to run the same selection as a read query, see exactly how many rows match, and only then delete. Bound the range explicitly — an unbounded read here masks how much the matching delete would remove.
from(bucket: "home_telemetry")
|> range(start: 2025-02-01T00:00:00Z, stop: now())
|> filter(fn: (r) => r.device_id == "d29f-rma")
|> group()
|> count(column: "_value")
|> yield(name: "rows_to_purge")
The group() collapses every series into one table so count returns a single fleet-wide total rather than a per-series breakdown. For the deprecated-measurement case, swap the filter to r._measurement == "debug_beacon". Record this number; the scheduled job in Step 3 aborts if the live count ever exceeds a sane ceiling, which catches a fat-fingered predicate before it erases the wrong device.
2. Purge by predicate with the CLI
For a one-off erasure — a support ticket, an incident — influx delete is the fastest safe path. It takes an explicit time range and a predicate expressed in the delete grammar: equality (=) and inequality (!=) on _measurement, tag keys, and _field, combined only with AND.
influx delete \
--bucket home_telemetry \
--org smart-home-prod \
--token "$INFLUX_PURGE_TOKEN" \
--start '2025-02-01T00:00:00Z' \
--stop '2026-02-24T00:00:00Z' \
--predicate 'device_id="d29f-rma"'
Quote the whole predicate in single quotes and the tag value in double quotes; that is the one syntax detail that trips people up. --start and --stop are mandatory and must be RFC 3339 timestamps — there is no “everything” default, which is a deliberate safety feature. Widen the range to cover the device’s entire lifetime (a 1970-01-01T00:00:00Z start is idiomatic for “from the beginning”) so no stray early points survive. The token here is the scoped purge token, not an operator token; scoping delete rights tightly is covered in Data Ingestion Security Frameworks.
3. Automate it as a scheduled purge job
Recurring obligations — nightly cleanup of test-device data, weekly purges of a deprecated measurement, a queue of erasure requests — belong in a repeatable job. The influxdb-client library exposes the same delete through delete_api().delete(...). Wrap it in the dry-run count from Step 1 and a blast-radius guard so the job refuses to run away.
import os
import sys
from influxdb_client import InfluxDBClient
BUCKET = "home_telemetry"
ORG = "smart-home-prod"
MAX_ROWS = 5_000_000 # blast-radius guard: refuse deletes larger than this
def dry_run_count(client, predicate_flux, start, stop):
query = f'''
from(bucket: "{BUCKET}")
|> range(start: {start}, stop: {stop})
|> filter(fn: (r) => {predicate_flux})
|> group() |> count(column: "_value")
'''
tables = client.query_api().query(query, org=ORG)
return sum(rec.get_value() for tbl in tables for rec in tbl.records)
def purge(predicate_delete, predicate_flux, start, stop):
with InfluxDBClient(url=os.environ["INFLUX_URL"],
token=os.environ["INFLUX_PURGE_TOKEN"], org=ORG) as client:
n = dry_run_count(client, predicate_flux, start, stop)
print(f"[purge] {n} rows match {predicate_delete} in [{start}, {stop})")
if n == 0:
print("[purge] nothing to delete"); return
if n > MAX_ROWS:
sys.exit(f"[abort] {n} rows exceeds guard {MAX_ROWS}; refusing")
client.delete_api().delete(start, stop, predicate_delete,
bucket=BUCKET, org=ORG)
print(f"[purge] deleted {n} rows for {predicate_delete}")
if __name__ == "__main__":
purge(
predicate_delete='device_id="d29f-rma"',
predicate_flux='r.device_id == "d29f-rma"',
start="2025-02-01T00:00:00Z",
stop="2026-02-24T00:00:00Z",
)
delete_api().delete(start, stop, predicate, bucket, org) takes RFC 3339 strings (or datetime objects) and the same predicate grammar as the CLI. The MAX_ROWS gate turns a mistyped predicate into an aborted run instead of a data-loss incident. Because deletes are naturally idempotent — a second run over data that is already gone matches zero rows and is a no-op — this job is safe to retry after a transient failure. Drive it from cron or an orchestrator; the reusable-client, retry, and scheduling patterns live in Python client orchestration patterns. See the InfluxDB delete API reference for the endpoint schema behind this call.
Gotchas and edge cases
A native InfluxDB task cannot delete — Flux has no delete function. The tempting instinct is to schedule the purge as a native task, the way you would a downsampling rollup. Flux can from, filter, and to, but it has no primitive that removes existing points; deletion lives only in the delete API. So the destructive step must run from an external scheduler that calls the API, not from option task. If purging and migrating are two halves of the same lifecycle, drive both from the same job and reach for the migration side via Storage Tier Migration Automation so warm copies exist before hot rows are purged.
Predicates are AND-only equality — no OR, no regex, no _value. One predicate cannot express “device A or device B”, cannot match device_id =~ /^test-/, and cannot filter on the field value itself. Attempting OR or a regex returns an error, not a partial match. To purge several devices, loop and issue one delete per device (or per predicate); to purge a naming pattern, resolve the concrete tag values with a Flux schema.tagValues query first, then delete each. Keeping each predicate a single equality also keeps the dry-run count honest.
A delete is immediate to queries but reclaims disk on compaction. After a successful delete the rows vanish from query results at once, so verification returns zero straight away. The physical storage, however, is tombstoned and only reclaimed when the affected shards next compact — disk usage can lag the delete by minutes to hours. Do not read a still-elevated disk figure as a failed purge; confirm success by querying for the data, never by watching the volume gauge.
Verification
Re-run the selection as a read. An empty result (or a zero count) proves the purge landed, while any surviving rows point straight at a range or predicate that missed some data:
from(bucket: "home_telemetry")
|> range(start: 2025-02-01T00:00:00Z, stop: now())
|> filter(fn: (r) => r.device_id == "d29f-rma")
|> group()
|> count(column: "_value")
|> yield(name: "should_be_empty")
Then confirm the blast stayed contained by counting a sibling device that should be untouched — a non-zero result there is the reassurance that the predicate hit only its target and nothing adjacent was collateral.
FAQ
Can a scheduled InfluxDB task run the purge for me?
No. Native tasks execute Flux, and Flux has no delete operation — it can only read, transform, and write. Schedule the purge from cron, a systemd timer, a Kubernetes CronJob, or an orchestrator that calls the delete API, exactly as the Python job in Step 3 does.
Does purging free disk space right away?
Not instantly. The delete removes rows from query results immediately, but the underlying storage is tombstoned and physically reclaimed at the next compaction of the affected shards, which can be minutes to hours later. Verify by querying the data, not by watching disk usage fall.
How do I purge two devices in one run?
Issue one delete per device. The predicate grammar supports only AND of equalities, so it cannot express “device A OR device B”. Loop over the list in your purge job — each delete is independently counted, guarded, and idempotent — rather than trying to combine them into a single predicate.
Related
- Retention Policy Design — the whole-bucket expiry model these predicate purges complement for selective, early deletion.
- Data Ingestion Security Frameworks — how to scope the least-privilege purge token that the delete job runs under.
- Storage Tier Migration Automation — migrating data to a warm tier before purging it from the hot bucket.