Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.
from typing import Optional, Sequence
from .monitor_series_window_types import MetricSample
from .monitor_series_window import series_window ← from monitor.series-window ^1.0.0 · built alongside by fune
from .monitor_alert_rule_types import CompareOp, RuleEvaluation, ThresholdRule
def _beyond(op: CompareOp, value: int, limit: int) -> bool:
if op == "gt":
return value > limit
if op == "gte":
return value >= limit
if op == "lt":
return value < limit
if op == "lte":
return value <= limit
raise ValueError("unknown comparison: %s" % (op,))
def evaluate_alert_rule(samples: Sequence[MetricSample], rule: ThresholdRule, now: int) -> RuleEvaluation:
"""The state of a threshold rule at `now`, Prometheus style: a breach
must hold for `for_seconds` before it fires, and until then it is pending.
With a clear_threshold a breach clears only once the value passes it the
other way, so a metric wobbling around the threshold does not flap
between firing and inactive. The breach clock (`since`) restarts after a
clear."""
if not isinstance(now, int) or isinstance(now, bool):
raise ValueError("now must be a whole number of seconds")
op = rule.op
threshold = rule.threshold
clear = threshold if rule.clear_threshold is None else rule.clear_threshold
_beyond(op, 0, 0)
upward = op in ("gt", "gte")
if (clear > threshold) if upward else (clear < threshold):
raise ValueError("clearThreshold must be on the non-breaching side of threshold: %s %d, clear %d" % (op, threshold, clear))
if rule.for_seconds < 0:
raise ValueError("forSeconds must not be negative, received %d" % (rule.for_seconds,))
stale = rule.stale_after_seconds
if stale is not None and stale < 0:
raise ValueError("staleAfterSeconds must not be negative, received %d" % (stale,))
# series_window checks the order of the whole series and keeps at <= now.
start = min(samples[0].at, now + 1) if len(samples) > 0 else now + 1
seen = series_window(samples, start, now + 1)
since: Optional[int] = None
for s in seen:
if since is None:
if _beyond(op, s.value, threshold):
since = s.at
elif not _beyond(op, s.value, clear):
since = None
if len(seen) == 0:
return RuleEvaluation(state="no-data", since=None, value=None, held_seconds=0)
latest = seen[-1]
if stale is not None and now - latest.at > stale:
return RuleEvaluation(state="no-data", since=None, value=latest.value, held_seconds=0)
if since is None:
return RuleEvaluation(state="inactive", since=None, value=latest.value, held_seconds=0)
held = now - since
return RuleEvaluation(state="firing" if held >= rule.for_seconds else "pending", since=since, value=latest.value, held_seconds=held)