| index e136c9c4b..0cbf605d7 100644 |
| --- a/internal/host/metrics_loop.go |
| +++ b/internal/host/metrics_loop.go |
| @@ -51,15 +51,18 @@ const samplerMaxBuf = 120 |
| // reading (for host CPU), the previous per-unit cgroup usage_usec reading |
| // (for unit CPU), the wall-clock time of the previous tick (unit CPU needs |
| // elapsed wall time, not just the counter delta), the buffered samples |
| -// awaiting their next flush (hard-capped at samplerMaxBuf), and whether |
| -// this sampler has completed at least one flush of the current hour yet — |
| -// see metricsFlush's restart-merge doc comment for why that matters. |
| +// awaiting their next flush (hard-capped at samplerMaxBuf), and which hours |
| +// this process has already written/merged at least once — see |
| +// writeHourChunk's doc comment for why a set (not a single bool) is |
| +// required: a restart's first-ever flush can span an hour boundary, in |
| +// which case BOTH the old and the current hour are being written by this |
| +// process for the first time in the same call. |
| type sampler struct { |
| - prevCPU *metrics.CPUCounters |
| - prevUnit map[string]*int64 // unit -> cpu usec at previous tick |
| - prevAt time.Time |
| - buf []metrics.Sample |
| - firstFlushDone bool |
| + prevCPU *metrics.CPUCounters |
| + prevUnit map[string]*int64 // unit -> cpu usec at previous tick |
| + prevAt time.Time |
| + buf []metrics.Sample |
| + writtenHours map[string]bool // hour -> this process has written/merged it at least once |
| } |
| |
| // tick reads p once and appends one metrics.Sample to sm.buf. The very |
| @@ -144,6 +147,17 @@ func metricsRawPrefix(box string) string { |
| // buf must not grow unbounded by carrying it forever. Pruning of raw chunks |
| // older than 48h runs only when an hour actually rolled over (there is |
| // nothing new to prune otherwise). |
| +// |
| +// Every hour written here — whether it lands in the "old" bucket or is the |
| +// current one — goes through writeHourChunk, which merges in any |
| +// pre-existing chunk the FIRST time this process writes that hour. That |
| +// matters for old hours too, not just the current one: after a restart, a |
| +// sampler's very first flush can itself span an hour boundary (a crash |
| +// mid-hour, then a restart that ticks a couple of times before its first |
| +// flush lands in the next hour) — in that case the OLD hour is also being |
| +// written by this process for the first time, and skipping the merge there |
| +// would silently clobber that hour's pre-crash samples and permanently |
| +// corrupt its day rollup the moment RollupHour runs over the truncated set. |
| func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, now time.Time) error { |
| if len(sm.buf) == 0 { |
| return nil |
| @@ -166,10 +180,14 @@ func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, |
| } |
| |
| for _, hour := range oldHours { |
| - samples := oldByHour[hour] |
| - if err := metrics.WriteRaw(ctx, s, metrics.RawChunk{V: 1, Box: box, Hour: hour, Samples: samples}); err != nil { |
| + samples, err := sm.writeHourChunk(ctx, s, box, hour, oldByHour[hour]) |
| + if err != nil { |
| return err |
| } |
| + // RollupHour runs over the (possibly merged) full set, never just |
| + // what this process itself collected — otherwise a restart would |
| + // permanently truncate that hour's day rollup even though the raw |
| + // chunk itself was correctly merged above. |
| agg := metrics.RollupHour(samples) |
| if err := metrics.UpsertRollup(ctx, s, box, samples[0].At, agg); err != nil { |
| return err |
| @@ -183,35 +201,56 @@ func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, |
| } |
| |
| if len(curSamples) > 0 { |
| - if !sm.firstFlushDone { |
| - merged, err := mergeExistingChunk(ctx, s, box, curHour, curSamples) |
| - if err != nil { |
| - return err |
| - } |
| - curSamples = merged |
| - sm.firstFlushDone = true |
| - } |
| - if err := metrics.WriteRaw(ctx, s, metrics.RawChunk{V: 1, Box: box, Hour: curHour, Samples: curSamples}); err != nil { |
| + merged, err := sm.writeHourChunk(ctx, s, box, curHour, curSamples) |
| + if err != nil { |
| return err |
| } |
| + curSamples = merged |
| } |
| |
| sm.buf = curSamples |
| return nil |
| } |
| |
| -// mergeExistingChunk reads whatever raw chunk already exists for hour (the |
| -// box's still-open current-hour chunk, most likely written by this same box |
| -// before a restart) and merges it under inMemory: existing samples first, |
| -// then the in-memory ones, deduping on exact At-timestamp collisions with |
| -// the in-memory sample winning (it's the fresher read of that instant). A |
| -// fresh sampler's buf always starts empty after a restart, and WriteRaw is |
| -// a wholesale replace — without this, the first flush after a restart would |
| -// silently clobber everything the box measured earlier in the still-open |
| -// hour, and that loss would propagate permanently into the hour's day |
| -// rollup once it closes. Only called once per sampler lifetime (gated by |
| -// sm.firstFlushDone), since every flush after the first already carries |
| -// forward everything this process itself wrote to the current hour. A |
| +// writeHourChunk writes samples as hour's raw chunk, merging in whatever |
| +// chunk already exists at that key the first time THIS sampler writes that |
| +// particular hour (tracked in sm.writtenHours) — see metricsFlush's doc |
| +// comment for why both old and current hours need this, not just the |
| +// current one. Every write after the first for a given hour skips the |
| +// merge: by then sm.buf already carries forward everything this process |
| +// itself wrote there, so re-merging would be redundant (and for an old |
| +// hour, over by definition — it's dropped from buf right after this call). |
| +// Returns the actual sample set written (merged or not) so the caller uses |
| +// the right data for RollupHour / for keeping in buf. |
| +func (sm *sampler) writeHourChunk(ctx context.Context, s release.Store, box, hour string, samples []metrics.Sample) ([]metrics.Sample, error) { |
| + if sm.writtenHours == nil { |
| + sm.writtenHours = map[string]bool{} |
| + } |
| + if !sm.writtenHours[hour] { |
| + merged, err := mergeExistingChunk(ctx, s, box, hour, samples) |
| + if err != nil { |
| + return nil, err |
| + } |
| + samples = merged |
| + sm.writtenHours[hour] = true |
| + } |
| + if err := metrics.WriteRaw(ctx, s, metrics.RawChunk{V: 1, Box: box, Hour: hour, Samples: samples}); err != nil { |
| + return nil, err |
| + } |
| + return samples, nil |
| +} |
| + |
| +// mergeExistingChunk reads whatever raw chunk already exists for hour (most |
| +// likely written by this same box before a restart or crash) and merges it |
| +// under inMemory: existing samples first, then the in-memory ones, deduping |
| +// on exact At-timestamp collisions with the in-memory sample winning (it's |
| +// the fresher read of that instant). A fresh sampler's buf always starts |
| +// empty after a restart, and WriteRaw is a wholesale replace — without |
| +// this, the first flush that touches a given hour after a restart would |
| +// silently clobber everything the box measured earlier in that hour, and |
| +// that loss would propagate permanently into the hour's day rollup once it |
| +// closes (or immediately, if the hour in question is already "old" by the |
| +// time this process gets to it — see metricsFlush's doc comment). A |
| // missing chunk (no prior write this hour — the common case) is not an |
| // error; a corrupt one is logged and skipped rather than failing the flush, |
| // since a garbled pre-restart file must not block ongoing monitoring. |
| @@ -261,25 +300,43 @@ func pruneStaleRaw(ctx context.Context, s release.Store, box string, now time.Ti |
| return nil |
| } |
| |
| -// alertTick evaluates rules against latest, persists the resulting |
| -// AlertState to statePath (create parent dir; write via temp+rename — never |
| -// leaving a half-written state file a restart could load), publishes the |
| -// new state to metrics.AlertKey(box) only when a rule's firing state |
| -// actually changed (Evaluate emits Events exactly on those transitions — |
| -// see its own doc comment — so "any events" is exactly the "state changed" |
| -// test), and posts each event to webhookURL ("" skips the webhook |
| -// entirely). |
| +// alertTick evaluates rules against latest and, when that produces a |
| +// firing/clearing transition (Evaluate emits Events exactly on those — see |
| +// its own doc comment), publishes the new state to metrics.AlertKey(box) |
| +// and posts each event to webhookURL ("" skips the webhook). |
| +// |
| +// Retry semantics on a publish failure are the crux of this function: *st |
| +// is the caller's durable record of "what has been reliably told to the |
| +// world", so it is only ever advanced to newState when the publish that |
| +// matches it actually succeeded (or there was nothing to publish — no |
| +// events this tick). If putAlertState fails, *st is left exactly as it was: |
| +// the SAME transition is still "pending" as far as this sampler is |
| +// concerned, so the next tick's Evaluate call starts from the same |
| +// not-yet-committed state and — as long as the underlying condition is |
| +// still there — reproduces the identical event and retries both the |
| +// publish and the webhook post. Committing newState unconditionally (this |
| +// function's first-round shape) would silently swallow a one-tick store |
| +// blip forever: Evaluate only emits an event on an actual transition, so |
| +// once *st already says "firing", every later tick just confirms it's |
| +// still firing with no new event and nothing left to retry. |
| // |
| -// A local persist failure is logged and otherwise ignored — it must NEVER |
| -// suppress publish or the webhook post below. The local write is most |
| -// likely to fail exactly when the disk rule is the one firing (disk full), |
| -// which is precisely the alert that must not go silent because of it; the |
| -// store publish and the webhook are the channels that actually reach |
| -// anyone, and neither depends on the local file. A webhook post failure is |
| -// likewise logged and otherwise ignored: the console watcher polling |
| -// AlertKey(box) is the reliable channel, the webhook is best-effort. A |
| -// genuine publish failure (the store write itself) IS still returned, since |
| -// unlike the local file, that's the reliable channel actually failing. |
| +// The webhook post is still attempted this tick regardless of whether |
| +// publish succeeded — it's independently fire-and-forget, logged and |
| +// otherwise ignored on failure, same as before; it does not gate whether |
| +// *st advances. Local persistence to statePath is also unconditionally |
| +// best-effort (a failure is logged, never returned or allowed to block |
| +// publish/webhook — see round 1's fix, since the local write is most likely |
| +// to fail exactly when the disk rule is the one firing), but it always |
| +// persists whichever state ends up committed to *st, so a restart loads |
| +// exactly what this process believes right now — never a state the store |
| +// was never told about. |
| +// |
| +// Honest limit: for a sustain-window rule (mem, cpu — sustain > 0), not |
| +// committing *st on a publish failure also means that rule's Since |
| +// bookkeeping doesn't advance either, so a publish outage lasting the whole |
| +// sustain window would keep resetting the clock and the rule would never |
| +// actually fire until the store recovers. Immediate rules (disk, lag; |
| +// sustain 0) are unaffected — they fire the same tick regardless. |
| func alertTick(ctx context.Context, s release.Store, box string, rules metrics.Rules, |
| st *metrics.AlertState, latest metrics.Sample, statePath, webhookURL string, |
| post func(url string, body []byte) error) error { |
| @@ -293,10 +350,6 @@ func alertTick(ctx context.Context, s release.Store, box string, rules metrics.R |
| } |
| newState, events := metrics.Evaluate(rules, *st, obs) |
| |
| - if err := persistAlertState(statePath, newState); err != nil { |
| - log.Printf("host: alert state persist to %s failed, continuing to publish/post: %v", statePath, err) |
| - } |
| - |
| var publishErr error |
| if len(events) > 0 { |
| body, err := json.MarshalIndent(newState, "", " ") |
| @@ -307,6 +360,8 @@ func alertTick(ctx context.Context, s release.Store, box string, rules metrics.R |
| } |
| } |
| |
| + // Fire-and-forget, independent of publish's outcome: an event still |
| + // deserves a webhook attempt even if the store write for it failed. |
| for _, ev := range events { |
| if webhookURL == "" { |
| continue |
| @@ -321,7 +376,17 @@ func alertTick(ctx context.Context, s release.Store, box string, rules metrics.R |
| } |
| } |
| |
| - *st = newState |
| + committed := *st |
| + if publishErr == nil { |
| + committed = newState |
| + *st = newState |
| + } |
| + // else: *st stays exactly as it was — see doc comment above. |
| + |
| + if err := persistAlertState(statePath, committed); err != nil { |
| + log.Printf("host: alert state persist to %s failed, continuing: %v", statePath, err) |
| + } |
| + |
| return publishErr |
| } |
| |
amadan