Skip to content

Commit 057a46a

Browse files
authored
Merge pull request #15 from ding-labs/add-run-lifetime-windowing
feat(evaluator): run-lifetime windowed aggregations (over run)
2 parents 9f2781f + 603e0a0 commit 057a46a

10 files changed

Lines changed: 487 additions & 38 deletions

File tree

ding.yaml.example

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -146,3 +146,15 @@ rules:
146146
# message: "job failed: exit code {{ .value }} after {{ .duration_seconds }}s"
147147
# alert:
148148
# - notifier: github_actions
149+
150+
# Whole-run aggregate alert — fires once at run exit if avg memory across
151+
# the entire run exceeded 80%. The "over run" window is bounded by the
152+
# ding run subprocess lifetime; no events are evicted by wall-clock time.
153+
# - name: high_avg_mem
154+
# match:
155+
# metric: mem_pct
156+
# condition: avg(value) over run > 80
157+
# mode: end-of-run
158+
# message: "avg memory was {{ .avg }}% across the run"
159+
# alert:
160+
# - notifier: slack

docs/configuration.md

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -266,6 +266,7 @@ A list of alerting rules. Rules are evaluated independently; each has its own co
266266
| `match.metric` | string | no | Metric name filter |
267267
| `condition` | string | yes | Evaluation expression (see below) |
268268
| `cooldown` | duration | no | Minimum time between consecutive alerts for the same label-set |
269+
| `mode` | string | no | Set to `end-of-run` to defer evaluation until `ding run` exits; omit for immediate (mid-run) evaluation |
269270
| `message` | string | no | Alert message template (Go `text/template` syntax) |
270271
| `alert` | list | yes | List of `{notifier: <name>}` targets |
271272

@@ -301,6 +302,51 @@ value < 5 OR count(value) over 1m > 100
301302
302303
Comparison operators: `>`, `>=`, `<`, `<=`, `==`, `!=`
303304
305+
#### Run-lifetime windows: `over run`
306+
307+
In addition to wall-clock durations like `over 5m`, the windowed-condition
308+
grammar accepts the literal `run`, which bounds the window to the lifetime
309+
of the `ding run` subprocess. Run-bounded windows do not evict entries by
310+
time — every event observed during the run is included in the aggregate,
311+
subject only to the configured `max_buffer_size` cap.
312+
313+
```yaml
314+
rules:
315+
# Whole-run aggregate, fires once at exit
316+
- name: high_avg_mem
317+
match: { metric: mem_pct }
318+
condition: avg(value) over run > 80
319+
mode: end-of-run
320+
message: "avg memory was {{ .avg }}% across the run"
321+
322+
# Run-bounded sliding window, fires mid-run on threshold cross
323+
- name: errors_pile_up
324+
match: { metric: errors }
325+
condition: count(value) over run > 10
326+
cooldown: 30s
327+
message: "errors in this run: {{ .count }}"
328+
```
329+
330+
The behavior matrix:
331+
332+
| condition window | `mode: end-of-run`? | result |
333+
|---|---|---|
334+
| `over 5m` | no | wall-clock sliding (default) |
335+
| `over 5m` | yes | aggregate of last 5m of run, fires at exit |
336+
| `over run` | no | run-bounded sliding, fires mid-run when threshold crosses |
337+
| `over run` | yes | whole-run aggregate, fires once at exit |
338+
339+
**Cooldown caveat.** Aggregates like `count` are monotonically non-decreasing
340+
under `over run` — once `count > 10`, it stays `> 10`. Without `mode:
341+
end-of-run` or a `cooldown:`, such a rule fires on every subsequent matching
342+
event. Pair `over run` mid-run rules with a meaningful `cooldown:` (or use
343+
`mode: end-of-run` for fire-once-at-exit semantics).
344+
345+
**`ding serve` mode.** `over run` is supported syntactically in the daemon
346+
mode, where it means "since daemon start" (the buffer accumulates indefinitely,
347+
capped by `max_buffer_size`). The wedge use case is `ding run`; prefer
348+
wall-clock windows in long-running serve deployments.
349+
304350
### Message template variables
305351

306352
| Variable | Available | Description |

internal/cli/test_rule_test.go

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,104 @@ rules:
9090
}
9191
}
9292

93+
func TestRunTestRule_OverRunWindow_AggregatesAcrossRun(t *testing.T) {
94+
dir := t.TempDir()
95+
cfg := filepath.Join(dir, "ding.yaml")
96+
if err := os.WriteFile(cfg, []byte(`
97+
notifiers:
98+
slack:
99+
type: webhook
100+
url: https://example.invalid/webhook
101+
rules:
102+
- name: high_avg_mem
103+
match: { metric: mem }
104+
condition: avg(value) over run > 50
105+
mode: end-of-run
106+
message: "avg mem: {{ .avg }}"
107+
alert:
108+
- notifier: slack
109+
`), 0644); err != nil {
110+
t.Fatalf("write config: %v", err)
111+
}
112+
113+
// Three events spanning 2 hours — far wider than any wall-clock window
114+
// the test would plausibly use. avg(40, 60, 80) = 60 > 50, so the
115+
// end-of-run rule should fire with avg=60.
116+
events := `{"metric":"mem","value":40,"timestamp":"2026-05-08T10:00:00Z"}
117+
{"metric":"mem","value":60,"timestamp":"2026-05-08T11:00:00Z"}
118+
{"metric":"mem","value":80,"timestamp":"2026-05-08T12:00:00Z"}
119+
`
120+
in := strings.NewReader(events)
121+
var out, errBuf bytes.Buffer
122+
123+
err := runTestRule(cfg, "text", false, in, &out, &errBuf)
124+
if err != nil {
125+
t.Fatalf("runTestRule: %v\nstderr: %s", err, errBuf.String())
126+
}
127+
128+
got := out.String()
129+
if !strings.Contains(got, "high_avg_mem") {
130+
t.Errorf("expected high_avg_mem rule to fire at end-of-run:\n%s", got)
131+
}
132+
if !strings.Contains(got, "avg mem: 60") {
133+
t.Errorf("expected rendered message to show avg=60 across whole run:\n%s", got)
134+
}
135+
if got := strings.Count(out.String(), "would fire"); got != 1 {
136+
t.Errorf("expected exactly 1 fire (mode: end-of-run), got %d:\n%s", got, out.String())
137+
}
138+
}
139+
140+
func TestRunTestRule_WindowedRule_EvictsOldestBeforeWindow(t *testing.T) {
141+
// Sibling to TestRunTestRule_WindowedRule_FiresAfterEnoughEvents.
142+
// That test fits all events inside the 5m window so it doesn't
143+
// exercise eviction. This one spans the boundary: events at t0 and
144+
// t0+10m with a 5m window — the older event must be evicted by the
145+
// time the newer one is processed, leaving the rule's avg computed
146+
// over the second event alone.
147+
dir := t.TempDir()
148+
cfg := filepath.Join(dir, "ding.yaml")
149+
if err := os.WriteFile(cfg, []byte(`
150+
notifiers:
151+
slack:
152+
type: webhook
153+
url: https://example.invalid/webhook
154+
rules:
155+
- name: hot_avg
156+
match: { metric: temp }
157+
condition: avg(value) over 5m > 50
158+
message: "avg high: {{ .avg }}"
159+
alert:
160+
- notifier: slack
161+
`), 0644); err != nil {
162+
t.Fatalf("write config: %v", err)
163+
}
164+
165+
// First event at t0 is well below threshold (10).
166+
// Second event 10 minutes later is above threshold (90).
167+
// With a 5m window, the first event is evicted before the second
168+
// is evaluated. The avg of {90} alone is 90, condition true.
169+
// If eviction were broken (no eviction), avg would be (10+90)/2=50,
170+
// which is NOT > 50, and the rule would NOT fire.
171+
events := `{"metric":"temp","value":10,"timestamp":"2026-05-08T10:00:00Z"}
172+
{"metric":"temp","value":90,"timestamp":"2026-05-08T10:10:00Z"}
173+
`
174+
in := strings.NewReader(events)
175+
var out, errBuf bytes.Buffer
176+
177+
err := runTestRule(cfg, "text", false, in, &out, &errBuf)
178+
if err != nil {
179+
t.Fatalf("runTestRule: %v\nstderr: %s", err, errBuf.String())
180+
}
181+
182+
got := out.String()
183+
if !strings.Contains(got, "hot_avg") {
184+
t.Errorf("expected hot_avg to fire (eviction should drop the first event so avg=90):\n%s", got)
185+
}
186+
if !strings.Contains(got, "avg high: 90") {
187+
t.Errorf("expected avg=90 (only second event remains after 5m eviction):\n%s", got)
188+
}
189+
}
190+
93191
func TestRunTestRule_NoMatch_SilentStdout(t *testing.T) {
94192
dir := t.TempDir()
95193
cfg := writeFixtureConfig(t, dir)

internal/evaluator/condition.go

Lines changed: 20 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -10,16 +10,17 @@ import (
1010

1111
var (
1212
reEventCond = regexp.MustCompile(`^value\s*(>|>=|<|<=|==|!=)\s*(-?\d+(?:\.\d+)?)$`)
13-
reWindowedCond = regexp.MustCompile(`^(avg|max|min|count|sum)\(value\)\s+over\s+(\d+[smh])\s*(>|>=|<|<=|==|!=)\s*(-?\d+(?:\.\d+)?)$`)
13+
reWindowedCond = regexp.MustCompile(`^(avg|max|min|count|sum)\(value\)\s+over\s+(\d+[smh]|run)\s*(>|>=|<|<=|==|!=)\s*(-?\d+(?:\.\d+)?)$`)
1414
)
1515

1616
// Condition is a parsed alert condition.
1717
type Condition struct {
18-
Windowed bool
19-
Op string
20-
Literal float64
21-
Func string // windowed only: avg|max|min|count|sum
22-
Window time.Duration // windowed only
18+
Windowed bool
19+
RunBounded bool // true when "over run"; Window is zero in this case
20+
Op string
21+
Literal float64
22+
Func string // windowed only: avg|max|min|count|sum
23+
Window time.Duration // windowed only; zero when RunBounded
2324
}
2425

2526
// ParseCondition parses a condition string from ding.yaml.
@@ -30,12 +31,17 @@ func ParseCondition(s string) (Condition, error) {
3031
}
3132
if m := reWindowedCond.FindStringSubmatch(s); m != nil {
3233
fn := m[1]
33-
dur, err := time.ParseDuration(m[2])
34+
windowToken := m[2]
35+
op := m[3]
36+
lit, _ := strconv.ParseFloat(m[4], 64)
37+
if windowToken == "run" {
38+
return Condition{Windowed: true, RunBounded: true, Func: fn, Window: 0, Op: op, Literal: lit}, nil
39+
}
40+
dur, err := time.ParseDuration(windowToken)
3441
if err != nil {
3542
return Condition{}, fmt.Errorf("invalid duration in condition %q: %w", s, err)
3643
}
37-
lit, _ := strconv.ParseFloat(m[4], 64)
38-
return Condition{Windowed: true, Func: fn, Window: dur, Op: m[3], Literal: lit}, nil
44+
return Condition{Windowed: true, Func: fn, Window: dur, Op: op, Literal: lit}, nil
3945
}
4046
return Condition{}, fmt.Errorf("unrecognized condition syntax: %q", s)
4147
}
@@ -82,9 +88,10 @@ type evalContext struct {
8288
// ID is the sequential integer assigned at parse time (zero-based per rule) and
8389
// is the same integer used as the middle segment of the buffer key.
8490
type windowedLeaf struct {
85-
ID int
86-
Func string
87-
Window time.Duration
91+
ID int
92+
Func string
93+
Window time.Duration
94+
RunBounded bool
8895
}
8996

9097
// leafExpr is a leaf node wrapping a single parsed Condition.
@@ -106,7 +113,7 @@ func (l *leafExpr) eval(ctx evalContext) bool {
106113

107114
func (l *leafExpr) collectWindowedLeaves() []windowedLeaf {
108115
if l.Windowed {
109-
return []windowedLeaf{{ID: l.id, Func: l.Func, Window: l.Window}}
116+
return []windowedLeaf{{ID: l.id, Func: l.Func, Window: l.Window, RunBounded: l.RunBounded}}
110117
}
111118
return nil
112119
}

internal/evaluator/end_of_run_test.go

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -214,6 +214,98 @@ func TestProcess_DuringRunStillFiresAlongsideEndOfRun(t *testing.T) {
214214
}
215215
}
216216

217+
// TestProcessEndOfRun_OverRun_AggregatesAcrossWholeRun verifies that a rule
218+
// with `condition: avg(value) over run > X` and `mode: end-of-run` aggregates
219+
// across the entire run, regardless of how long the run lasted, and fires
220+
// correctly at run exit. Distinguishes "over run" from "over Nm" by spanning
221+
// timestamps wider than any plausible wall-clock window.
222+
func TestProcessEndOfRun_OverRun_AggregatesAcrossWholeRun(t *testing.T) {
223+
rules := []EngineRule{
224+
{
225+
Name: "whole_run_avg",
226+
Match: map[string]string{"metric": "mem"},
227+
Condition: "avg(value) over run > 50",
228+
Message: "avg mem was {{ .avg }}",
229+
Alerts: []string{"stdout"},
230+
Mode: "end-of-run",
231+
},
232+
}
233+
eng, err := NewEngine(rules, 1000)
234+
if err != nil {
235+
t.Fatalf("NewEngine: %v", err)
236+
}
237+
238+
// Feed events spanning 2 hours — far beyond any wall-clock window.
239+
t0 := time.Now()
240+
eng.Process(ingester.Event{Metric: "mem", Value: 40, At: t0}, t0)
241+
eng.Process(ingester.Event{Metric: "mem", Value: 60, At: t0.Add(1 * time.Hour)}, t0.Add(1*time.Hour))
242+
eng.Process(ingester.Event{Metric: "mem", Value: 80, At: t0.Add(2 * time.Hour)}, t0.Add(2*time.Hour))
243+
244+
// At end-of-run the buffer should still hold all three entries:
245+
// avg(40, 60, 80) = 60, which is > 50.
246+
alerts := eng.ProcessEndOfRun(t0.Add(2 * time.Hour))
247+
if len(alerts) != 1 {
248+
t.Fatalf("expected 1 end-of-run alert, got %d", len(alerts))
249+
}
250+
if alerts[0].Avg != 60 {
251+
t.Errorf("alert.Avg = %v, want 60 (avg of full run)", alerts[0].Avg)
252+
}
253+
if alerts[0].Count != 3 {
254+
t.Errorf("alert.Count = %v, want 3 (all events in run)", alerts[0].Count)
255+
}
256+
}
257+
258+
// TestProcess_OverRun_FiresMidRunWithCooldown verifies the orthogonal
259+
// composition: `over run` without `mode: end-of-run` fires mid-run when
260+
// the threshold crosses, and cooldown prevents repeated firing.
261+
func TestProcess_OverRun_FiresMidRunWithCooldown(t *testing.T) {
262+
rules := []EngineRule{
263+
{
264+
Name: "errors_pile_up",
265+
Match: map[string]string{"metric": "errors"},
266+
Condition: "count(value) over run > 2",
267+
Cooldown: 10 * time.Minute,
268+
Message: "errors: {{ .count }}",
269+
Alerts: []string{"stdout"},
270+
},
271+
}
272+
eng, err := NewEngine(rules, 1000)
273+
if err != nil {
274+
t.Fatalf("NewEngine: %v", err)
275+
}
276+
277+
t0 := time.Now()
278+
// Events 1, 2 — count is at most 2, condition (count > 2) false.
279+
for i := 0; i < 2; i++ {
280+
alerts := eng.Process(ingester.Event{
281+
Metric: "errors", Value: 1,
282+
At: t0.Add(time.Duration(i) * time.Second),
283+
}, t0.Add(time.Duration(i)*time.Second))
284+
if len(alerts) > 0 {
285+
t.Fatalf("event %d: rule fired prematurely (count=%v)", i, alerts[0].Count)
286+
}
287+
}
288+
// Event 3 — count crosses to 3, condition true, alert fires.
289+
alerts := eng.Process(ingester.Event{
290+
Metric: "errors", Value: 1,
291+
At: t0.Add(2 * time.Second),
292+
}, t0.Add(2*time.Second))
293+
if len(alerts) != 1 {
294+
t.Fatalf("expected 1 alert at event 3, got %d", len(alerts))
295+
}
296+
if alerts[0].Count != 3 {
297+
t.Errorf("alert.Count = %v, want 3", alerts[0].Count)
298+
}
299+
// Event 4 — count is 4, condition true, but cooldown blocks the alert.
300+
alerts = eng.Process(ingester.Event{
301+
Metric: "errors", Value: 1,
302+
At: t0.Add(3 * time.Second),
303+
}, t0.Add(3*time.Second))
304+
if len(alerts) != 0 {
305+
t.Errorf("expected cooldown to block alert at event 4, got %d", len(alerts))
306+
}
307+
}
308+
217309
// TestParseLabelKey verifies that the labelKey reverser handles the formats
218310
// produced by LabelSetKey.
219311
func TestParseLabelKey(t *testing.T) {

internal/evaluator/engine.go

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ func (e *Engine) Process(event ingester.Event, now time.Time) []Alert {
135135
}
136136
for _, leaf := range leaves {
137137
leafBufKey := rule.Name + ":" + strconv.Itoa(leaf.ID) + ":" + labelKey
138-
buf := e.getOrCreateBuffer(leafBufKey, leaf.Window)
138+
buf := e.getOrCreateBuffer(leafBufKey, leaf.Window, leaf.RunBounded)
139139
buf.Add(event.Value, event.At)
140140
if buf.HasEntries(now) {
141141
ctx.Available[leaf.ID] = true
@@ -189,7 +189,7 @@ func (e *Engine) Process(event ingester.Event, now time.Time) []Alert {
189189
if len(leaves) == 1 {
190190
leaf := leaves[0]
191191
leafBufKey := rule.Name + ":" + strconv.Itoa(leaf.ID) + ":" + labelKey
192-
buf := e.getOrCreateBuffer(leafBufKey, leaf.Window)
192+
buf := e.getOrCreateBuffer(leafBufKey, leaf.Window, leaf.RunBounded)
193193
alert.Avg = buf.Avg(now)
194194
alert.Max = buf.Max(now)
195195
alert.Min = buf.Min(now)
@@ -364,13 +364,14 @@ func (e *Engine) trackLabelKey(ruleName, labelKey string) {
364364

365365
// getOrCreateBuffer returns the ring buffer for a buffer key, creating it if needed.
366366
// Uses bufMu independently of the RWMutex so it is safe to call from Process() under RLock.
367-
func (e *Engine) getOrCreateBuffer(key string, window time.Duration) *RingBuffer {
367+
// runBounded is honored only on first creation; subsequent calls return the existing buffer.
368+
func (e *Engine) getOrCreateBuffer(key string, window time.Duration, runBounded bool) *RingBuffer {
368369
e.bufMu.Lock()
369370
defer e.bufMu.Unlock()
370371
if buf, ok := e.buffers[key]; ok {
371372
return buf
372373
}
373-
buf := NewRingBuffer(window, e.maxBuf)
374+
buf := NewRingBuffer(window, e.maxBuf, runBounded)
374375
e.buffers[key] = buf
375376
return buf
376377
}

0 commit comments

Comments
 (0)