Threshold-based alerting with Flux and Python hooks
A data-centre floor with 48 racks streams inlet temperature and humidity every ten seconds into a dc-environmental bucket, and the on-call team wants a page the moment a rack inlet crosses 32 °C — but not the 400 pages a night that a naive _value > 32 check produces when a rack hovers right on the line and every reading flips the alert on and off. The fix is two cooperating pieces: a Flux check task that assigns a severity level to each window and records it as a status, with a deliberate gap between the trip point and the clear point so a rack sitting on the boundary does not flap, and a small Python hook that reads only genuine state changes and delivers them to a webhook exactly once. This page builds both against the Anomaly Detection & Alerting workflow, using InfluxDB’s native monitor package for the detection tier and Python for the delivery tier where retries, secrets, and third-party endpoints belong.
Prerequisites
Solution walkthrough
1. Build the threshold check task with hysteresis
monitor.check evaluates each incoming record against ordered crit / warn / ok predicates, stamps it with a _level, and writes a row into the statuses measurement of _monitoring. The trick that kills flapping is to make the clear threshold lower than the trip threshold, so a rack must genuinely cool down before it returns to ok rather than oscillating around a single line. Here a rack goes critical above 32 °C but only clears once it falls to 25 °C or below — the 25–32 °C band is a deliberate dead zone.
import "influxdata/influxdb/monitor"
option task = {name: "rack-inlet-temp-check", every: 1m, offset: 15s}
check = {
_check_id: "0af1b2c3d4e50001",
_check_name: "Rack inlet temperature",
_type: "threshold",
tags: {domain: "thermal", floor: "dc-1"},
}
// Hysteresis: trip high, clear low. The 25-32C gap holds the previous level.
crit = (r) => r.inlet_temp_c > 32.0
warn = (r) => r.inlet_temp_c > 27.0
ok = (r) => r.inlet_temp_c <= 25.0
messageFn = (r) =>
"Rack ${r.rack}: inlet ${string(v: r.inlet_temp_c)}C is ${r._level}"
from(bucket: "dc-environmental")
|> range(start: -task.every)
|> filter(fn: (r) => r._measurement == "rack_environment")
|> filter(fn: (r) => r._field == "inlet_temp_c")
|> aggregateWindow(every: 1m, fn: mean, createEmpty: false)
|> monitor.check(
data: check,
messageFn: messageFn,
crit: crit,
warn: warn,
ok: ok,
)
The aggregateWindow step is not cosmetic: checking a one-minute mean rather than raw ten-second samples smooths transient spikes that would otherwise trip crit for a single reading. monitor.check requires the predicates to reference fields as columns, which is why the query pivots implicitly through aggregateWindow keeping inlet_temp_c as a column. Every window now carries a _level; a value landing in the 25–32 °C gap matches none of the three predicates and is left as unknown, which monitor.stateChanges treats as “hold the last real level” — that is the hysteresis working. Choosing the aggregate function itself (mean here, but median for spiky signals) is its own decision covered under the aggregation section; the offset that lets late edge batches land before the check fires follows the same reasoning as interval scheduling for tasks.
The flap-free guarantee is just an ordering constraint on the three thresholds, with the hysteresis margin $H$ being the width of the dead-band:
$$ T_{clear} < T_{warn} < T_{crit}, \qquad H = T_{crit} - T_{clear} $$
Size $H$ to exceed the sensor’s short-term noise amplitude; here $H = 32 - 25 = 7,^{\circ}\mathrm{C}$ comfortably clears the ±1 °C jitter of the rack probes.
2. Read only new CRIT state changes
The hook should never re-scan the whole statuses history — it needs just the transitions into crit since it last ran. monitor.from reads statuses back out, and monitor.stateChanges collapses them to the rows where the level actually changed, so a rack that stays critical for an hour yields one alert, not sixty.
import "influxdata/influxdb/monitor"
monitor.from(
start: -2m,
fn: (r) => r._check_name == "Rack inlet temperature",
)
|> monitor.stateChanges(fromLevel: "any", toLevel: "crit")
|> keep(columns: ["_time", "_check_id", "rack", "_level", "_message"])
The start: -2m window overlaps the hook’s one-minute cadence on purpose: it guarantees no transition slips through a gap between runs, at the cost of occasionally seeing the same change twice — which the idempotency layer in the next step absorbs. stateChanges(fromLevel: "any", toLevel: "crit") is what turns a noisy status stream into discrete, actionable events.
3. Post to the webhook idempotently from a Python hook
Delivery lives in Python because that is where retries, the webhook secret, and endpoint-specific formatting belong — not inside a Flux task. Idempotency comes from a deterministic dedupe key (check + rack + level + timestamp) recorded in a short-retention bucket; the hook skips any key it has already delivered, so overlapping query windows and hook re-runs never double-page the on-call engineer.
import hashlib
import os
import requests
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
ORG = "dc-ops"
DEDUPE_BUCKET = "alert-dedupe"
WEBHOOK_URL = os.environ["ALERT_WEBHOOK_URL"]
NEW_CRIT = '''
import "influxdata/influxdb/monitor"
monitor.from(start: -2m, fn: (r) => r._check_name == "Rack inlet temperature")
|> monitor.stateChanges(fromLevel: "any", toLevel: "crit")
|> keep(columns: ["_time", "_check_id", "rack", "_level", "_message"])
'''
RECENT_KEYS = '''
from(bucket: "alert-dedupe")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "sent")
|> keep(columns: ["dedupe_key"])
|> group()
|> distinct(column: "dedupe_key")
'''
def dedupe_key(rec) -> str:
raw = f'{rec["_check_id"]}|{rec["rack"]}|{rec["_level"]}|{rec["_time"].isoformat()}'
return hashlib.sha1(raw.encode()).hexdigest()[:16]
def run_hook() -> None:
client = InfluxDBClient(
url=os.environ["INFLUX_URL"],
token=os.environ["INFLUX_TOKEN"],
org=ORG,
)
query_api = client.query_api()
write_api = client.write_api(write_options=SYNCHRONOUS)
try:
already = {
r["dedupe_key"]
for tbl in query_api.query(RECENT_KEYS)
for r in tbl.records
}
for table in query_api.query(NEW_CRIT):
for rec in table.records:
values = rec.values
key = dedupe_key(values)
if key in already:
continue # already delivered — idempotent skip
resp = requests.post(
WEBHOOK_URL,
json={
"rack": values["rack"],
"level": values["_level"],
"message": values["_message"],
"ts": values["_time"].isoformat(),
"dedupe_key": key,
},
timeout=5,
)
resp.raise_for_status()
# Only record the marker AFTER a confirmed 2xx delivery.
write_api.write(
bucket=DEDUPE_BUCKET,
record=Point("sent")
.tag("dedupe_key", key)
.tag("rack", values["rack"])
.field("delivered", 1),
)
already.add(key)
finally:
client.close()
if __name__ == "__main__":
run_hook()
Order matters: the marker is written only after the webhook returns 2xx, so a crash mid-delivery leaves the key unrecorded and the next run retries rather than silently dropping the page. Run the hook on a one-minute cron alongside the check task; wrapping the requests.post in exponential backoff and bounding concurrency across all 48 racks are extensions covered in Python client orchestration patterns.
Gotchas and edge cases
No hysteresis means an alert storm. With a single _value > 32 predicate and no lower clear line, a rack oscillating between 31.9 °C and 32.1 °C emits a fresh crit and ok on nearly every window, and stateChanges faithfully forwards each flip. The dead-band between T_clear and T_crit is not optional polish — it is the mechanism that makes threshold alerting usable. Widen $H$ until the flapping stops in a replay of your noisiest rack’s history.
monitor.stateChanges needs the status history to exist. If you delete or fail to provision the _monitoring bucket, or scope the check task’s token without write access to it, monitor.check silently writes nothing and the hook queries an empty stream — no error, no alerts. Confirm the statuses measurement is being populated before trusting the pipeline, and grant the check task write access to _monitoring explicitly.
A too-short overlap window drops transitions. The hook queries start: -2m against a one-minute cadence for a reason. If you tighten the window to -1m and a run is delayed by scheduler jitter or a slow webhook, a transition that landed in the uncovered gap is never seen and the page is lost. Keep the query window at roughly twice the run interval and let idempotency absorb the resulting duplicates. Absence of data entirely — a sensor that goes silent rather than hot — is a different failure that threshold checks cannot catch; that gap is closed by deadman checks for detecting silent IoT sensors.
Verification
Confirm the check task is recording statuses and that critical transitions are present before you trust delivery:
import "influxdata/influxdb/monitor"
monitor.from(start: -15m, fn: (r) => r._check_name == "Rack inlet temperature")
|> group(columns: ["_level"])
|> count(column: "_message")
|> yield(name: "status_counts_by_level")
A healthy pipeline shows ok counts dominating with occasional warn/crit rows during real thermal events. Then prove idempotency by counting delivery markers per key — every key must appear exactly once no matter how many overlapping runs saw it:
from(bucket: "alert-dedupe")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "sent")
|> group(columns: ["dedupe_key"])
|> count()
|> filter(fn: (r) => r._value > 1)
|> yield(name: "duplicates_should_be_empty")
An empty result confirms no crit transition was delivered twice; any row here means the marker write is racing the webhook post and needs the ordering tightened.
FAQ
Why use monitor.check instead of a plain Flux comparison?
monitor.check gives you a first-class status stream with per-level severity, structured messages, and a statuses history that monitor.from and monitor.stateChanges can query, so notifications key off genuine transitions rather than raw values. A hand-rolled filter(fn: (r) => r._value > 32) has no memory of the previous state, which is exactly what makes hysteresis and change-detection impossible to express cleanly.
Should the webhook call live in the Flux task instead of Python?
Keep delivery in Python. Flux tasks can call http.post, but retries, honouring Retry-After, secret management, and per-endpoint payload shaping are awkward there and block the task engine on a slow third party. Detecting the condition in Flux and delivering it from a Python hook cleanly separates the fast, deterministic detection tier from the messy, network-bound delivery tier.
How do I add a warning-level notification without doubling the noise?
Run a second monitor.stateChanges(fromLevel: "any", toLevel: "warn") query in the same hook and route it to a lower-urgency channel, but reuse the same dedupe-key scheme so a rack climbing ok → warn → crit produces one warn notice and one crit page, not a cascade. Because the hysteresis dead-band holds levels steady, warn transitions are already de-flapped by the check task.
Related
- Anomaly Detection & Alerting — the parent workflow covering checks, notifications, and deadman detection for automated pipelines.
- Deadman Checks for Detecting Silent IoT Sensors — the complementary check for sensors that stop reporting entirely rather than crossing a threshold.
- Python Client Orchestration Patterns — reusable client, batching, and backoff patterns for hardening the delivery hook.
Up one level: Anomaly Detection & Alerting