| 1 | // The metrics sampler loop — the box-side half of host-process monitoring. |
| 2 | // Every 60s it samples the whole host plus every monitored systemd unit |
| 3 | // (sampler.tick), evaluates on-box alert rules against the freshest sample |
| 4 | // (alertTick), and every fifth tick flushes the buffered samples out to the |
| 5 | // deployment bucket as raw chunks, rolling up and pruning across hour |
| 6 | // boundaries (metricsFlush). The sampler/probes split mirrors statusLoop's |
| 7 | // own status.Probes seam: sampleProbes is the injectable fixture bundle so |
| 8 | // host_test.go never touches /proc, a cgroup file, or a real clock. |
| 9 | package host |
| 10 | |
| 11 | import ( |
| 12 | "context" |
| 13 | "encoding/json" |
| 14 | "errors" |
| 15 | "log" |
| 16 | "os" |
| 17 | "path/filepath" |
| 18 | "time" |
| 19 | |
| 20 | "github.com/carlosframework/platform/internal/alerting" |
| 21 | "github.com/carlosframework/platform/internal/metrics" |
| 22 | "github.com/carlosframework/platform/internal/release" |
| 23 | "github.com/carlosframework/platform/internal/status" |
| 24 | ) |
| 25 | |
| 26 | // sampleProbes is the injectable fixture bundle sampler.tick reads from each |
| 27 | // tick, like status.Probes for statusTick. now is its own seam (rather than |
| 28 | // time.Now called directly) so tests control the elapsed time between ticks |
| 29 | // that unit CPU deltas depend on. |
| 30 | type sampleProbes struct { |
| 31 | procStat func() *metrics.CPUCounters |
| 32 | loadavg func() (l1, l5, l15 *float64) |
| 33 | memFrac func() *float64 |
| 34 | diskFrac func() *float64 |
| 35 | units func() map[string]unitRaw |
| 36 | now func() time.Time |
| 37 | } |
| 38 | |
| 39 | // samplerMaxBuf hard-caps sm.buf at ~2h of 60s ticks. metricsFlush only |
| 40 | // trims buf on a fully successful flush, so a store that's up for reads but |
| 41 | // persistently failing writes (an IAM policy that allows PutObject but |
| 42 | // denies DeleteObject, a full disk, whatever) would otherwise grow buf by |
| 43 | // one sample every tick forever, re-marshaling the whole (ever-growing) |
| 44 | // buffer on every retry — this bounds that regardless of how flush is |
| 45 | // doing, since the cap is enforced in tick, not in flush. |
| 46 | const samplerMaxBuf = 120 |
| 47 | |
| 48 | // sampler carries the delta state between ticks: the previous /proc/stat |
| 49 | // reading (for host CPU), the previous per-unit cgroup usage_usec reading |
| 50 | // (for unit CPU), the wall-clock time of the previous tick (unit CPU needs |
| 51 | // elapsed wall time, not just the counter delta), the buffered samples |
| 52 | // awaiting their next flush (hard-capped at samplerMaxBuf), and which hours |
| 53 | // this process has already written/merged at least once — see |
| 54 | // writeHourChunk's doc comment for why a set (not a single bool) is |
| 55 | // required: a restart's first-ever flush can span an hour boundary, in |
| 56 | // which case BOTH the old and the current hour are being written by this |
| 57 | // process for the first time in the same call. |
| 58 | type sampler struct { |
| 59 | prevCPU *metrics.CPUCounters |
| 60 | prevUnit map[string]*int64 // unit -> cpu usec at previous tick |
| 61 | prevAt time.Time |
| 62 | buf []metrics.Sample |
| 63 | writtenHours map[string]bool // hour -> this process has written/merged it at least once |
| 64 | backfilled bool // the one-shot previous-hour rollup repair has run (see backfillPrevHourRollup) |
| 65 | } |
| 66 | |
| 67 | // tick reads p once and appends one metrics.Sample to sm.buf. The very |
| 68 | // first tick on a fresh sampler has no previous reading to diff against, so |
| 69 | // its host and unit CPU fractions are nil — not a guessed zero — exactly |
| 70 | // like metrics.CPUFrac/UnitCPUFrac's own nil-on-no-prior-reading contract. |
| 71 | func (sm *sampler) tick(p sampleProbes) { |
| 72 | now := p.now() |
| 73 | curCPU := p.procStat() |
| 74 | l1, l5, l15 := p.loadavg() |
| 75 | mem := p.memFrac() |
| 76 | disk := p.diskFrac() |
| 77 | units := p.units() |
| 78 | |
| 79 | var elapsed time.Duration |
| 80 | if !sm.prevAt.IsZero() { |
| 81 | elapsed = now.Sub(sm.prevAt) |
| 82 | } |
| 83 | |
| 84 | hostCPU := metrics.CPUFrac(sm.prevCPU, curCPU) |
| 85 | |
| 86 | var unitSamples map[string]metrics.UnitSample |
| 87 | var newPrevUnit map[string]*int64 |
| 88 | if units != nil { |
| 89 | unitSamples = make(map[string]metrics.UnitSample, len(units)) |
| 90 | newPrevUnit = make(map[string]*int64, len(units)) |
| 91 | for name, u := range units { |
| 92 | var prevUsec *int64 |
| 93 | if sm.prevUnit != nil { |
| 94 | prevUsec = sm.prevUnit[name] |
| 95 | } |
| 96 | unitSamples[name] = metrics.UnitSample{ |
| 97 | CPUFrac: metrics.UnitCPUFrac(prevUsec, u.cpuUsec, elapsed), |
| 98 | MemBytes: u.memBytes, |
| 99 | MemPeakBytes: u.memPeakBytes, |
| 100 | ReadBytes: u.readBytes, |
| 101 | WriteBytes: u.writeBytes, |
| 102 | } |
| 103 | newPrevUnit[name] = u.cpuUsec |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | sm.buf = append(sm.buf, metrics.Sample{ |
| 108 | At: now, |
| 109 | Host: metrics.HostSample{ |
| 110 | CPUFrac: hostCPU, |
| 111 | Load1: l1, |
| 112 | Load5: l5, |
| 113 | Load15: l15, |
| 114 | MemFrac: mem, |
| 115 | DiskFrac: disk, |
| 116 | }, |
| 117 | Units: unitSamples, |
| 118 | }) |
| 119 | |
| 120 | if len(sm.buf) > samplerMaxBuf { |
| 121 | drop := len(sm.buf) - samplerMaxBuf |
| 122 | log.Printf("host: sampler buffer exceeded %d samples (flush stuck?) — dropping %d oldest unflushed sample(s)", samplerMaxBuf, drop) |
| 123 | sm.buf = sm.buf[drop:] |
| 124 | } |
| 125 | |
| 126 | sm.prevCPU = curCPU |
| 127 | sm.prevUnit = newPrevUnit |
| 128 | sm.prevAt = now |
| 129 | } |
| 130 | |
| 131 | // metricsRawPrefix is the bucket prefix under which a box's raw chunks live |
| 132 | // — metrics.RawKey's own prefix, duplicated here only for List's sake (List |
| 133 | // takes a bare prefix, not a templated key). |
| 134 | func metricsRawPrefix(box string) string { |
| 135 | return "metrics/boxes/" + box + "/raw/" |
| 136 | } |
| 137 | |
| 138 | // metricsFlush writes sm.buf out as raw chunks grouped by each sample's own |
| 139 | // hour — never mixing two hours into one chunk, even when a single flush |
| 140 | // call spans a boundary. Samples in the current hour (relative to now) are |
| 141 | // written as that hour's chunk and kept in buf (the current hour's chunk |
| 142 | // gets rewritten wholesale on every flush as more samples accumulate, |
| 143 | // mirroring WriteRaw's own "replace whatever is there" contract). Samples |
| 144 | // from any older hour are written to their own chunk, rolled up via |
| 145 | // RollupHour+UpsertRollup, and then dropped from buf — that hour is done, |
| 146 | // buf must not grow unbounded by carrying it forever. Pruning of raw chunks |
| 147 | // older than 48h runs only when an hour actually rolled over (there is |
| 148 | // nothing new to prune otherwise). |
| 149 | // |
| 150 | // Every hour written here — whether it lands in the "old" bucket or is the |
| 151 | // current one — goes through writeHourChunk, which merges in any |
| 152 | // pre-existing chunk the FIRST time this process writes that hour. That |
| 153 | // matters for old hours too, not just the current one: after a restart, a |
| 154 | // sampler's very first flush can itself span an hour boundary (a crash |
| 155 | // mid-hour, then a restart that ticks a couple of times before its first |
| 156 | // flush lands in the next hour) — in that case the OLD hour is also being |
| 157 | // written by this process for the first time, and skipping the merge there |
| 158 | // would silently clobber that hour's pre-crash samples and permanently |
| 159 | // corrupt its day rollup the moment RollupHour runs over the truncated set. |
| 160 | func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, now time.Time) error { |
| 161 | // Once per process, before anything else: repair the hour a restart may |
| 162 | // have left un-rolled-up (see backfillPrevHourRollup). It is deliberately |
| 163 | // not allowed to fail the flush — it is repair work for an hour that is |
| 164 | // already over, and this flush's own samples matter more. A failure just |
| 165 | // leaves the flag unset so the next flush tries again. |
| 166 | if !sm.backfilled { |
| 167 | if err := backfillPrevHourRollup(ctx, s, box, now); err != nil { |
| 168 | log.Printf("host: previous-hour rollup backfill for %s failed, retrying next flush: %v", box, err) |
| 169 | } else { |
| 170 | sm.backfilled = true |
| 171 | } |
| 172 | } |
| 173 | |
| 174 | if len(sm.buf) == 0 { |
| 175 | return nil |
| 176 | } |
| 177 | curHour := metrics.HourOf(now) |
| 178 | |
| 179 | curSamples := make([]metrics.Sample, 0, len(sm.buf)) |
| 180 | oldByHour := map[string][]metrics.Sample{} |
| 181 | var oldHours []string |
| 182 | for _, samp := range sm.buf { |
| 183 | h := metrics.HourOf(samp.At) |
| 184 | if h == curHour { |
| 185 | curSamples = append(curSamples, samp) |
| 186 | continue |
| 187 | } |
| 188 | if _, ok := oldByHour[h]; !ok { |
| 189 | oldHours = append(oldHours, h) |
| 190 | } |
| 191 | oldByHour[h] = append(oldByHour[h], samp) |
| 192 | } |
| 193 | |
| 194 | for _, hour := range oldHours { |
| 195 | samples, merged, err := sm.writeHourChunk(ctx, s, box, hour, oldByHour[hour]) |
| 196 | if err != nil { |
| 197 | return err |
| 198 | } |
| 199 | // RollupHour runs over the (possibly merged) full set, never just |
| 200 | // what this process itself collected — otherwise a restart would |
| 201 | // permanently truncate that hour's day rollup even though the raw |
| 202 | // chunk itself was correctly merged above. |
| 203 | agg := metrics.RollupHour(samples) |
| 204 | if err := metrics.UpsertRollup(ctx, s, box, samples[0].At, agg); err != nil { |
| 205 | return err |
| 206 | } |
| 207 | // Both durable writes for this hour succeeded — NOW it's safe to |
| 208 | // mark it written (only meaningful when this call actually merged; |
| 209 | // see writeHourChunk/markHourWritten) and retire its samples from |
| 210 | // buf immediately, rather than waiting for the whole flush call to |
| 211 | // finish. If a LATER hour's write fails below, buf must not still |
| 212 | // be carrying THIS hour's already-durably-written samples — a |
| 213 | // retry that reprocessed them would see writtenHours already true, |
| 214 | // skip the merge, and overwrite this hour's correctly-merged chunk |
| 215 | // with just the bare in-memory set, silently regressing it. |
| 216 | if merged { |
| 217 | sm.markHourWritten(hour) |
| 218 | } |
| 219 | sm.dropHourFromBuf(hour) |
| 220 | } |
| 221 | |
| 222 | if len(oldHours) > 0 { |
| 223 | if err := pruneStaleRaw(ctx, s, box, now); err != nil { |
| 224 | return err |
| 225 | } |
| 226 | } |
| 227 | |
| 228 | if len(curSamples) > 0 { |
| 229 | merged, wasMerged, err := sm.writeHourChunk(ctx, s, box, curHour, curSamples) |
| 230 | if err != nil { |
| 231 | return err |
| 232 | } |
| 233 | curSamples = merged |
| 234 | if wasMerged { |
| 235 | sm.markHourWritten(curHour) |
| 236 | } |
| 237 | } |
| 238 | |
| 239 | sm.buf = curSamples |
| 240 | return nil |
| 241 | } |
| 242 | |
| 243 | // backfillPrevHourRollup rolls up the PREVIOUS hour's stored raw chunk if the |
| 244 | // day object has no entry for that hour yet. It runs once per process, on the |
| 245 | // first flush. |
| 246 | // |
| 247 | // The gap it closes: rollups are otherwise derived only from what this |
| 248 | // process itself buffered. An agent that stops at 09:58 and starts again at |
| 249 | // 10:07 never has an hour-09 sample in buf, so nothing ever calls RollupHour |
| 250 | // for hour 09 — the raw chunk written before the stop sits there with no day |
| 251 | // entry, and 48h later pruneStaleRaw deletes it. The hour is then gone from |
| 252 | // the trend history permanently, silently, and only for the hours that |
| 253 | // contain a restart (i.e. exactly the interesting ones). |
| 254 | // |
| 255 | // Deliberately narrow, in three ways. Only the immediately previous hour is |
| 256 | // considered: a longer outage has no raw chunk to recover from anyway, since |
| 257 | // the agent that would have written it was down. An hour whose rollup already |
| 258 | // exists is left completely alone — this repairs a gap, it never rewrites a |
| 259 | // closed hour. And a missing chunk, an empty one, or an unreadable one is a |
| 260 | // no-op rather than an error: there is simply nothing to recover. |
| 261 | func backfillPrevHourRollup(ctx context.Context, s release.Store, box string, now time.Time) error { |
| 262 | prev := now.Add(-time.Hour) |
| 263 | hour := metrics.HourOf(prev) |
| 264 | key := metrics.RawKey(box, hour) |
| 265 | |
| 266 | body, _, err := s.Get(ctx, key) |
| 267 | if err != nil { |
| 268 | if errors.Is(err, os.ErrNotExist) { |
| 269 | return nil // no chunk for that hour: nothing to roll up |
| 270 | } |
| 271 | return err |
| 272 | } |
| 273 | var chunk metrics.RawChunk |
| 274 | if err := json.Unmarshal(body, &chunk); err != nil { |
| 275 | log.Printf("host: raw chunk %s unreadable, skipping rollup backfill: %v", key, err) |
| 276 | return nil |
| 277 | } |
| 278 | if len(chunk.Samples) == 0 { |
| 279 | return nil |
| 280 | } |
| 281 | |
| 282 | has, err := rollupHasHour(ctx, s, box, prev) |
| 283 | if err != nil { |
| 284 | return err |
| 285 | } |
| 286 | if has { |
| 287 | return nil |
| 288 | } |
| 289 | |
| 290 | log.Printf("host: hour %s has a raw chunk but no day rollup (restart across the boundary) — rolling it up now", hour) |
| 291 | return metrics.UpsertRollup(ctx, s, box, prev, metrics.RollupHour(chunk.Samples)) |
| 292 | } |
| 293 | |
| 294 | // rollupHasHour reports whether box's day object already carries hour's UTC |
| 295 | // hour bucket. A missing day object is a plain "no". An unreadable one |
| 296 | // reports "yes": UpsertRollup would fail on the same unmarshal anyway, and a |
| 297 | // corrupt day object is a separate problem the normal rollover path will |
| 298 | // surface — the backfill must not spin on it every flush. |
| 299 | func rollupHasHour(ctx context.Context, s release.Store, box string, hour time.Time) (bool, error) { |
| 300 | key := metrics.DayKey(box, hour.UTC().Format("2006-01-02")) |
| 301 | body, _, err := s.Get(ctx, key) |
| 302 | if err != nil { |
| 303 | if errors.Is(err, os.ErrNotExist) { |
| 304 | return false, nil |
| 305 | } |
| 306 | return false, err |
| 307 | } |
| 308 | var day metrics.DayRollup |
| 309 | if err := json.Unmarshal(body, &day); err != nil { |
| 310 | log.Printf("host: day rollup %s unreadable, skipping rollup backfill: %v", key, err) |
| 311 | return true, nil |
| 312 | } |
| 313 | _, ok := day.Hours[hour.UTC().Format("15")] |
| 314 | return ok, nil |
| 315 | } |
| 316 | |
| 317 | // writeHourChunk writes samples as hour's raw chunk, merging in whatever |
| 318 | // chunk already exists at that key the first time THIS sampler writes that |
| 319 | // particular hour (tracked in sm.writtenHours) — see metricsFlush's doc |
| 320 | // comment for why both old and current hours need this, not just the |
| 321 | // current one. Every write after the first for a given hour skips the |
| 322 | // merge: by then sm.buf already carries forward everything this process |
| 323 | // itself wrote there, so re-merging would be redundant (and for an old |
| 324 | // hour, over by definition — it's dropped from buf right after this call). |
| 325 | // |
| 326 | // Deliberately does NOT mark sm.writtenHours itself — only that the merge |
| 327 | // happened this call (the returned merged bool). Marking is the caller's |
| 328 | // job, done only once every durable write this hour's data depends on has |
| 329 | // actually succeeded (for an old hour that's this WriteRaw AND its |
| 330 | // UpsertRollup; for the current hour, just this WriteRaw). Marking here, |
| 331 | // before the caller even knows whether those later steps succeed, would |
| 332 | // let a downstream failure retry with the merge silently skipped — exactly |
| 333 | // the bug this shape avoids. Returns the actual sample set written (merged |
| 334 | // or not) so the caller uses the right data for RollupHour / for keeping |
| 335 | // in buf. |
| 336 | func (sm *sampler) writeHourChunk(ctx context.Context, s release.Store, box, hour string, samples []metrics.Sample) (out []metrics.Sample, merged bool, err error) { |
| 337 | if !sm.writtenHours[hour] { |
| 338 | m, err := mergeExistingChunk(ctx, s, box, hour, samples) |
| 339 | if err != nil { |
| 340 | return nil, false, err |
| 341 | } |
| 342 | samples = m |
| 343 | merged = true |
| 344 | } |
| 345 | if err := metrics.WriteRaw(ctx, s, metrics.RawChunk{V: 1, Box: box, Hour: hour, Samples: samples}); err != nil { |
| 346 | return nil, false, err |
| 347 | } |
| 348 | return samples, merged, nil |
| 349 | } |
| 350 | |
| 351 | // markHourWritten records that hour's chunk has been durably written (and, |
| 352 | // where applicable, rolled up) at least once by this sampler — see |
| 353 | // writeHourChunk's doc comment for why the caller, not writeHourChunk |
| 354 | // itself, decides when that's true. |
| 355 | func (sm *sampler) markHourWritten(hour string) { |
| 356 | if sm.writtenHours == nil { |
| 357 | sm.writtenHours = map[string]bool{} |
| 358 | } |
| 359 | sm.writtenHours[hour] = true |
| 360 | } |
| 361 | |
| 362 | // dropHourFromBuf removes every sample belonging to hour from sm.buf. Old |
| 363 | // hours are retired from buf the moment their write is fully durable |
| 364 | // (metricsFlush calls this right after, per hour, inside its oldHours |
| 365 | // loop) rather than in one pass at the very end — so that if a LATER old |
| 366 | // hour's write fails within the same flush call, THIS hour's samples are |
| 367 | // already gone from buf and cannot be reprocessed (and, worse, rewritten |
| 368 | // unmerged — see metricsFlush's doc comment at the call site) on the next |
| 369 | // flush attempt. |
| 370 | func (sm *sampler) dropHourFromBuf(hour string) { |
| 371 | kept := make([]metrics.Sample, 0, len(sm.buf)) |
| 372 | for _, samp := range sm.buf { |
| 373 | if metrics.HourOf(samp.At) != hour { |
| 374 | kept = append(kept, samp) |
| 375 | } |
| 376 | } |
| 377 | sm.buf = kept |
| 378 | } |
| 379 | |
| 380 | // mergeExistingChunk reads whatever raw chunk already exists for hour (most |
| 381 | // likely written by this same box before a restart or crash) and merges it |
| 382 | // under inMemory: existing samples first, then the in-memory ones, deduping |
| 383 | // on exact At-timestamp collisions with the in-memory sample winning (it's |
| 384 | // the fresher read of that instant). A fresh sampler's buf always starts |
| 385 | // empty after a restart, and WriteRaw is a wholesale replace — without |
| 386 | // this, the first flush that touches a given hour after a restart would |
| 387 | // silently clobber everything the box measured earlier in that hour, and |
| 388 | // that loss would propagate permanently into the hour's day rollup once it |
| 389 | // closes (or immediately, if the hour in question is already "old" by the |
| 390 | // time this process gets to it — see metricsFlush's doc comment). A |
| 391 | // missing chunk (no prior write this hour — the common case) is not an |
| 392 | // error; a corrupt one is logged and skipped rather than failing the flush, |
| 393 | // since a garbled pre-restart file must not block ongoing monitoring. |
| 394 | func mergeExistingChunk(ctx context.Context, s release.Store, box, hour string, inMemory []metrics.Sample) ([]metrics.Sample, error) { |
| 395 | key := metrics.RawKey(box, hour) |
| 396 | body, _, err := s.Get(ctx, key) |
| 397 | if err != nil { |
| 398 | if errors.Is(err, os.ErrNotExist) { |
| 399 | return inMemory, nil |
| 400 | } |
| 401 | return nil, err |
| 402 | } |
| 403 | var existing metrics.RawChunk |
| 404 | if err := json.Unmarshal(body, &existing); err != nil { |
| 405 | log.Printf("host: existing raw chunk %s unreadable, flushing without it: %v", key, err) |
| 406 | return inMemory, nil |
| 407 | } |
| 408 | |
| 409 | inMemoryAt := make(map[time.Time]bool, len(inMemory)) |
| 410 | for _, samp := range inMemory { |
| 411 | inMemoryAt[samp.At] = true |
| 412 | } |
| 413 | |
| 414 | merged := make([]metrics.Sample, 0, len(existing.Samples)+len(inMemory)) |
| 415 | for _, samp := range existing.Samples { |
| 416 | if !inMemoryAt[samp.At] { |
| 417 | merged = append(merged, samp) |
| 418 | } |
| 419 | } |
| 420 | merged = append(merged, inMemory...) |
| 421 | return merged, nil |
| 422 | } |
| 423 | |
| 424 | // pruneStaleRaw deletes this box's raw chunks older than 48h — List + |
| 425 | // metrics.StaleRawKeys (which never returns a key it cannot parse — see its |
| 426 | // own doc comment) + Delete. |
| 427 | func pruneStaleRaw(ctx context.Context, s release.Store, box string, now time.Time) error { |
| 428 | keys, err := s.List(ctx, metricsRawPrefix(box)) |
| 429 | if err != nil { |
| 430 | return err |
| 431 | } |
| 432 | for _, key := range metrics.StaleRawKeys(keys, now, 48*time.Hour) { |
| 433 | if err := s.Delete(ctx, key); err != nil { |
| 434 | return err |
| 435 | } |
| 436 | } |
| 437 | return nil |
| 438 | } |
| 439 | |
| 440 | // alertTick evaluates rules against latest and, when that produces a |
| 441 | // firing/clearing transition (Evaluate emits Events exactly on those — see |
| 442 | // its own doc comment), publishes the new state to metrics.AlertKey(box) |
| 443 | // and delivers each event to sinks (alertsink.go: the webhook, the SNS topic, |
| 444 | // either, both or neither; nil is none). |
| 445 | // |
| 446 | // Retry semantics on a publish failure are the crux of this function: *st |
| 447 | // is the caller's durable record of "what has been reliably told to the |
| 448 | // world", so it is only ever advanced to newState when the publish that |
| 449 | // matches it actually succeeded (or there was nothing to publish — no |
| 450 | // events this tick). If putAlertState fails, *st is left exactly as it was: |
| 451 | // the SAME transition is still "pending" as far as this sampler is |
| 452 | // concerned, so the next tick's Evaluate call starts from the same |
| 453 | // not-yet-committed state and — as long as the underlying condition is |
| 454 | // still there — reproduces the identical event and retries both the |
| 455 | // publish and the webhook post. Committing newState unconditionally (this |
| 456 | // function's first-round shape) would silently swallow a one-tick store |
| 457 | // blip forever: Evaluate only emits an event on an actual transition, so |
| 458 | // once *st already says "firing", every later tick just confirms it's |
| 459 | // still firing with no new event and nothing left to retry. |
| 460 | // |
| 461 | // Delivery to the sinks is still attempted this tick regardless of whether |
| 462 | // publish succeeded — it's independently fire-and-forget, logged and counted |
| 463 | // on failure, never returned; it does not gate whether *st advances. Local persistence to statePath is also unconditionally |
| 464 | // best-effort (a failure is logged, never returned or allowed to block |
| 465 | // publish/webhook — see round 1's fix, since the local write is most likely |
| 466 | // to fail exactly when the disk rule is the one firing), but it always |
| 467 | // persists whichever state ends up committed to *st, so a restart loads |
| 468 | // exactly what this process believes right now — never a state the store |
| 469 | // was never told about. |
| 470 | // |
| 471 | // Honest limit: for a sustain-window rule (mem, cpu — sustain > 0), the |
| 472 | // Since bookkeeping itself is NOT at risk from a publish outage — every |
| 473 | // tick before the rule is actually ready to fire carries zero events, so |
| 474 | // publish is never even attempted on those ticks and *st (Since included) |
| 475 | // commits normally regardless of the store's health. The outage only bites |
| 476 | // on the tick(s) where the rule is ready to fire: if publish fails exactly |
| 477 | // then, *st doesn't advance, so the next tick re-evaluates the same |
| 478 | // already-overdue transition and retries. The alert ends up firing (and |
| 479 | // published) as soon as the store recovers, delayed by at most the |
| 480 | // outage's own duration — it does not compound beyond that. Immediate |
| 481 | // rules (disk, lag; sustain 0) retry the same way, just starting from |
| 482 | // their very first over-threshold tick instead of a sustain deadline. |
| 483 | func alertTick(ctx context.Context, s release.Store, box string, rules metrics.Rules, |
| 484 | st *metrics.AlertState, latest metrics.Sample, statePath string, |
| 485 | sinks *alertSinks) error { |
| 486 | |
| 487 | obs := metrics.Observation{ |
| 488 | At: latest.At, |
| 489 | CPUFrac: latest.Host.CPUFrac, |
| 490 | MemFrac: latest.Host.MemFrac, |
| 491 | DiskFrac: latest.Host.DiskFrac, |
| 492 | LagSecs: nil, // v1: no lag reading feeds the agent loop yet |
| 493 | GeoIP: geoIPAlertReading(latest.At), |
| 494 | } |
| 495 | newState, events := metrics.Evaluate(rules, *st, obs) |
| 496 | |
| 497 | var publishErr error |
| 498 | if len(events) > 0 { |
| 499 | body, err := json.MarshalIndent(newState, "", " ") |
| 500 | if err != nil { |
| 501 | publishErr = err |
| 502 | } else if err := putAlertState(ctx, s, metrics.AlertKey(box), body); err != nil { |
| 503 | publishErr = err |
| 504 | } |
| 505 | } |
| 506 | |
| 507 | // Fire-and-forget, independent of publish's outcome: an event still |
| 508 | // deserves a delivery attempt even if the store write for it failed. |
| 509 | for _, ev := range events { |
| 510 | sinks.deliver(ctx, rules, latest.At, ev) |
| 511 | } |
| 512 | |
| 513 | committed := *st |
| 514 | if publishErr == nil { |
| 515 | committed = newState |
| 516 | *st = newState |
| 517 | } |
| 518 | // else: *st stays exactly as it was — see doc comment above. |
| 519 | |
| 520 | if err := persistAlertState(statePath, committed); err != nil { |
| 521 | log.Printf("host: alert state persist to %s failed, continuing: %v", statePath, err) |
| 522 | } |
| 523 | |
| 524 | return publishErr |
| 525 | } |
| 526 | |
| 527 | // persistAlertState writes st to path as indented JSON via the repo's |
| 528 | // standard temp+rename idiom (internal/appconfig/sync.go's renderEnv, |
| 529 | // internal/release/store.go's DirStore.Put): a crashed write must never |
| 530 | // leave a half-written state file behind for the next boot to load and |
| 531 | // choke on. The parent directory (<dataDir>/monitor/) is created if |
| 532 | // missing — this is the first thing that ever writes into it. |
| 533 | func persistAlertState(path string, st metrics.AlertState) error { |
| 534 | if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { |
| 535 | return err |
| 536 | } |
| 537 | body, err := json.MarshalIndent(st, "", " ") |
| 538 | if err != nil { |
| 539 | return err |
| 540 | } |
| 541 | tmp := path + ".tmp" |
| 542 | if err := os.WriteFile(tmp, body, 0o644); err != nil { |
| 543 | return err |
| 544 | } |
| 545 | if err := os.Rename(tmp, path); err != nil { |
| 546 | os.Remove(tmp) |
| 547 | return err |
| 548 | } |
| 549 | return nil |
| 550 | } |
| 551 | |
| 552 | // putAlertState upserts key with a replace-or-create Put, retrying once on |
| 553 | // an ErrCASConflict — the same read-etag/Put/retry-once dance WriteRaw and |
| 554 | // UpsertRollup use, since this box is the only writer that should ever be |
| 555 | // targeting its own alert key. |
| 556 | func putAlertState(ctx context.Context, s release.Store, key string, body []byte) error { |
| 557 | etag, err := currentAlertEtag(ctx, s, key) |
| 558 | if err != nil { |
| 559 | return err |
| 560 | } |
| 561 | if _, err := s.Put(ctx, key, body, etag); err != nil { |
| 562 | if !errors.Is(err, release.ErrCASConflict) { |
| 563 | return err |
| 564 | } |
| 565 | etag, err = currentAlertEtag(ctx, s, key) |
| 566 | if err != nil { |
| 567 | return err |
| 568 | } |
| 569 | if _, err := s.Put(ctx, key, body, etag); err != nil { |
| 570 | return err |
| 571 | } |
| 572 | } |
| 573 | return nil |
| 574 | } |
| 575 | |
| 576 | // currentAlertEtag reads key's current etag for a replace-only Put, folding |
| 577 | // "doesn't exist yet" into "" (create-only) — the same convention |
| 578 | // metrics.WriteRaw's own currentRawEtag uses. |
| 579 | func currentAlertEtag(ctx context.Context, s release.Store, key string) (string, error) { |
| 580 | _, etag, err := s.Get(ctx, key) |
| 581 | if err != nil { |
| 582 | if errors.Is(err, os.ErrNotExist) { |
| 583 | return "", nil |
| 584 | } |
| 585 | return "", err |
| 586 | } |
| 587 | return etag, nil |
| 588 | } |
| 589 | |
| 590 | // httpPostWebhook is the production webhook post func: a 5s-timeout POST |
| 591 | // with a JSON content type. Any non-2xx status is an error — the caller |
| 592 | // (alertTick) only logs it, but it still needs to know delivery failed. |
| 593 | func httpPostWebhook(url string, body []byte) error { return alerting.PostJSON(url, body) } |
| 594 | |
| 595 | // realSampleProbes wires sampleProbes to the actual OS/cgroup readers in |
| 596 | // metrics_probes.go and probes.go — the production seam metricsLoop uses, |
| 597 | // never called from a test (host_test.go's convention: tests inject |
| 598 | // fixtures, never touch /proc or a syscall). |
| 599 | // |
| 600 | // stampedUnits is re-read inside the units probe rather than captured as a |
| 601 | // slice, because sampleProbes is built ONCE and read every tick: a snapshot |
| 602 | // taken here would pin the stamped set to whatever the registry held at agent |
| 603 | // start, and a route stamped this afternoon would go unsampled until the next |
| 604 | // restart. A nil func means "no stamped units" — the darwin/dev shape, and |
| 605 | // what every test double passes. |
| 606 | func realSampleProbes(execRoot string, stampedUnits func() []string) sampleProbes { |
| 607 | return sampleProbes{ |
| 608 | procStat: readProcStat, |
| 609 | loadavg: readLoadAvg, |
| 610 | memFrac: readMemFrac, |
| 611 | diskFrac: readDiskFrac, |
| 612 | units: func() map[string]unitRaw { |
| 613 | var stamped []string |
| 614 | if stampedUnits != nil { |
| 615 | stamped = stampedUnits() |
| 616 | } |
| 617 | return realUnitReadings(execRoot, stamped) |
| 618 | }, |
| 619 | now: time.Now, |
| 620 | } |
| 621 | } |
| 622 | |
| 623 | // realUnitReadings lists the box's currently active CARLOS units and reads |
| 624 | // each one's cgroup usage — the production units func sampleProbes.units |
| 625 | // wraps. Keyed by UNIT NAME (what the console matches against), read from the |
| 626 | // unit's own cgroup DIRECTORY (which for a tenant instance is a level down; |
| 627 | // see listCarlosUnits) — plus the operator-stamped bespoke units named in |
| 628 | // stamped (registry Route.Unit; see listStampedUnits) and, when delegation is |
| 629 | // armed, the per-host exec-child cgroups under execRoot (keyed |
| 630 | // carlos-exec@<host>; see listExecCgroups). |
| 631 | // |
| 632 | // The three scans may overlap — a stamped unit that also matches the carlos- |
| 633 | // convention is discovered twice — and the map write below is what dedupes |
| 634 | // them: both entries name the same unit and the same cgroup dir, so the |
| 635 | // second read is redundant, never contradictory. |
| 636 | func realUnitReadings(execRoot string, stamped []string) map[string]unitRaw { |
| 637 | units := listCarlosUnits(cgroupSliceDir) |
| 638 | // carlos-edge.service moved to carlos.slice (isolated-exec design §4.3; |
| 639 | // see cgroupCarlosSliceDir's doc comment), so it is found only by also |
| 640 | // scanning that second root. system.slice is still scanned above for |
| 641 | // every other platform unit — litestream, adopt, relay and the rest — |
| 642 | // which have not moved. |
| 643 | units = append(units, listCarlosUnits(cgroupCarlosSliceDir)...) |
| 644 | units = append(units, listStampedUnits(cgroupSliceDir, stamped)...) |
| 645 | units = append(units, listExecCgroups(execRoot)...) |
| 646 | if len(units) == 0 { |
| 647 | return nil |
| 648 | } |
| 649 | out := make(map[string]unitRaw, len(units)) |
| 650 | for _, u := range units { |
| 651 | out[u.name] = readUnitRaw(u.dir) |
| 652 | } |
| 653 | return out |
| 654 | } |
| 655 | |
| 656 | // alertPctEnvVars are the percent-integer threshold overrides |
| 657 | // metrics.RulesFromEnv reads. |
| 658 | var alertPctEnvVars = []string{"CARLOS_ALERT_DISK_PCT", "CARLOS_ALERT_MEM_PCT", "CARLOS_ALERT_CPU_PCT"} |
| 659 | |
| 660 | // logRejectedAlertOverrides says so when one of those is SET but not |
| 661 | // accepted. RulesFromEnv is pure and silently keeps the default, which is |
| 662 | // the right behaviour — an operator's typo must never stop the agent — but |
| 663 | // silence is the wrong report: someone who writes 0.95 (a fraction, the |
| 664 | // units this took before the README was corrected) would otherwise believe |
| 665 | // they had lowered the CPU threshold while the box quietly ran on 0.95's |
| 666 | // rejection and the built-in default. |
| 667 | func logRejectedAlertOverrides(getenv func(string) string) { |
| 668 | for _, name := range alertPctEnvVars { |
| 669 | v := getenv(name) |
| 670 | if v == "" { |
| 671 | continue |
| 672 | } |
| 673 | if _, ok := metrics.PctFrac(v); !ok { |
| 674 | log.Printf("agent: %s=%q ignored: want a percent integer 0-100 (e.g. 95) — keeping the built-in default", name, v) |
| 675 | } |
| 676 | } |
| 677 | } |
| 678 | |
| 679 | // loadAlertState reads a persisted AlertState from path, falling back to a |
| 680 | // fresh state for box on anything that isn't a clean read — a missing file |
| 681 | // (first boot) or a corrupt one (a previous crash mid-write, or a hand |
| 682 | // edit) must never crash the agent loop; monitoring starting from a blank |
| 683 | // slate is always a safe, honest state to resume from. |
| 684 | func loadAlertState(path, box string) metrics.AlertState { |
| 685 | fresh := metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}} |
| 686 | body, err := os.ReadFile(path) |
| 687 | if err != nil { |
| 688 | return fresh |
| 689 | } |
| 690 | var st metrics.AlertState |
| 691 | if err := json.Unmarshal(body, &st); err != nil { |
| 692 | return fresh |
| 693 | } |
| 694 | if st.Rules == nil { |
| 695 | st.Rules = map[string]metrics.RuleState{} |
| 696 | } |
| 697 | return st |
| 698 | } |
| 699 | |
| 700 | // metricsBoxLabel resolves the label metrics/alerts publish under — pulled |
| 701 | // out of metricsLoop as its own pure seam (I5, final customer-fleets review) |
| 702 | // so it is testable without a bucket, mirroring how statusTick takes an |
| 703 | // explicit labelOverride rather than re-deriving it inline. |
| 704 | // |
| 705 | // It must be the SAME resolved identity statusLoop uses |
| 706 | // (fleetidentity.go's FleetIdentity.Label, the operator-chosen |
| 707 | // fleet_boxes.label the edge actually dials), NEVER status.Label(machineID)'s |
| 708 | // hashed pseudonym, when this is a fleet box. Diverging from statusLoop here |
| 709 | // would put this box's heartbeat under one label |
| 710 | // (fleets/<fleet>/status/boxes/<label>.json) and its metrics under a |
| 711 | // DIFFERENT one (status.Label's hash), which breaks the fleet-channel |
| 712 | // identity a customer's monitoring page keys everything by — see |
| 713 | // fleetidentity.go's doc comment for why the hashed pseudonym is wrong here |
| 714 | // at all. An invalid (not merely absent) identity degrades to the hashed |
| 715 | // label rather than disabling metrics outright, mirroring statusLoop's own |
| 716 | // degrade-don't-disable choice. |
| 717 | func metricsBoxLabel(getenv func(string) string, machineID string) string { |
| 718 | if groupReferenced(getenv) { |
| 719 | // A box its region's group launched (boxreference.go). |
| 720 | return groupBoxReference() |
| 721 | } |
| 722 | if identity, err := resolveFleetIdentity(getenv); err == nil { |
| 723 | return identity.Label |
| 724 | } else if !errors.Is(err, errNotFleetBox) { |
| 725 | log.Printf("agent: %v — metrics will publish under the hashed pseudonym instead", err) |
| 726 | } |
| 727 | return status.Label(machineID) |
| 728 | } |
| 729 | |
| 730 | // BoxLabel is metricsBoxLabel for callers outside this package — the relay |
| 731 | // (`carlos relay serve`, relay spec §3) meters and heartbeats under the same |
| 732 | // box label the metrics loop publishes under, so the two agree about which |
| 733 | // box a row belongs to. A thin wrapper deliberately: the selection rule |
| 734 | // stays in one place, tested there. |
| 735 | func BoxLabel() string { return metricsBoxLabel(os.Getenv, readMachineID()) } |
| 736 | |
| 737 | // metricsLoop runs the metrics sampler's 60s tick forever, once armed — thin |
| 738 | // and untested by the same convention as statusLoop/adoptLoop: everything |
| 739 | // with behavior lives in sampler.tick, metricsFlush, alertTick, and |
| 740 | // metricsBoxLabel, all exercised above with fixtures. Missing |
| 741 | // status/deployment bucket config disables the loop with one log line, the |
| 742 | // same "a box with no store is still a valid deployment" shape every other |
| 743 | // optional loop in this package uses. Flush runs every 5th tick (5 minutes |
| 744 | // at the 60s tick cadence); alertTick runs every tick, since a threshold |
| 745 | // breach deserves faster detection than a metrics flush does. |
| 746 | // |
| 747 | // stampedUnits is the sweep's one registry seam (agent.go's |
| 748 | // stampedUnitNames): the bespoke units to sample alongside the carlos- ones, |
| 749 | // re-read once per tick so a freshly stamped route is picked up within a |
| 750 | // minute. It is a func, not the registry handle, so this loop keeps its "no |
| 751 | // state but a bucket and /proc" shape; nil = none. |
| 752 | // |
| 753 | // sinks are the agent's alert sinks, built and armed for paging by runAgent |
| 754 | // at startup (armAgentAlerts); the loop delivers its alerts on them and |
| 755 | // never builds or arms its own. nil delivers nowhere. |
| 756 | func metricsLoop(execRoot string, stampedUnits func() []string, sinks *alertSinks) { |
| 757 | // This loop is always on and shares a process with the agent that serves |
| 758 | // this box's traffic, so a panic in it — an unreadable cgroup shape, a |
| 759 | // nil no one expected — must not take the box down with it. Monitoring |
| 760 | // stopping is a bad day; the box stopping is an outage. |
| 761 | defer func() { |
| 762 | if r := recover(); r != nil { |
| 763 | log.Printf("agent: metrics loop panic, monitoring stopped for this process: %v", r) |
| 764 | } |
| 765 | }() |
| 766 | |
| 767 | ctx := context.Background() |
| 768 | // Same selection rule statusLoop uses, in one place for both (agent.go's |
| 769 | // openStatusStore): a status/deployment bucket wins and keeps direct S3; |
| 770 | // a fleet box with neither set publishes through the edge's store door. |
| 771 | raw, target, err := openStatusStore(ctx, os.Getenv) |
| 772 | if err != nil { |
| 773 | if errors.Is(err, errNoStatusStore) { |
| 774 | log.Printf("agent: metrics loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set") |
| 775 | } else { |
| 776 | log.Printf("agent: metrics loop disabled: %v", err) |
| 777 | } |
| 778 | return |
| 779 | } |
| 780 | // s is what metricsFlush/alertTick actually read/write through: the same |
| 781 | // CARLOS_STATUS_PREFIX rebase statusLoop applies (release.NewPrefixStore), |
| 782 | // so a fleet box's metrics/alerts land under its own fleets/<name>/ |
| 783 | // namespace instead of the shared metrics/boxes/ + alerts/boxes/ paths a |
| 784 | // platform box uses — without this, a customer's raw metrics would |
| 785 | // surface on the platform operator's monitoring page (console |
| 786 | // metricsview lists metrics/boxes/) and an alert would page the |
| 787 | // platform's own webhook for customer hardware. Empty prefix (every |
| 788 | // platform box today) is a pure passthrough — see the package doc |
| 789 | // comment's CARLOS_STATUS_PREFIX entry. resolveStatusPrefix, not a bare |
| 790 | // env read: on a door-backed fleet box an empty prefix would put every |
| 791 | // flush outside the door's write ACL, so that branch defaults to |
| 792 | // fleets/<fleet>/. |
| 793 | s := release.NewPrefixStore(raw, resolveStatusPrefix(os.Getenv)) |
| 794 | |
| 795 | box := metricsBoxLabel(os.Getenv, readMachineID()) |
| 796 | statePath := filepath.Join(dataDir(), "monitor", "alerts.json") |
| 797 | st := loadAlertState(statePath, box) |
| 798 | rules := metrics.RulesFromEnv(os.Getenv) |
| 799 | logRejectedAlertOverrides(os.Getenv) |
| 800 | |
| 801 | log.Printf("agent: metrics loop armed (60s ticks, 5min flush), publishing to %s", target) |
| 802 | |
| 803 | sm := &sampler{} |
| 804 | probes := realSampleProbes(execRoot, stampedUnits) |
| 805 | tick := 0 |
| 806 | // One watch for the life of the loop: its memory of what it has already |
| 807 | // alerted is what keeps a standing refusal from paging every minute. |
| 808 | // dataDir() is the bake dir the agent hands release.BakeDir. |
| 809 | var refusals refusalWatch |
| 810 | for { |
| 811 | time.Sleep(60 * time.Second) |
| 812 | tick++ |
| 813 | |
| 814 | rows := refusals.tick(ctx, dataDir(), sinks, time.Now()) |
| 815 | releaseRefusals.Store(&rows) |
| 816 | |
| 817 | sm.tick(probes) |
| 818 | latest := sm.buf[len(sm.buf)-1] |
| 819 | if err := alertTick(ctx, s, box, rules, &st, latest, statePath, sinks); err != nil { |
| 820 | log.Printf("agent: alert tick: %v", err) |
| 821 | } |
| 822 | |
| 823 | if tick%5 == 0 { |
| 824 | if err := metricsFlush(ctx, s, box, sm, time.Now()); err != nil { |
| 825 | log.Printf("agent: metrics flush: %v", err) |
| 826 | } |
| 827 | } |
| 828 | } |
| 829 | } |
| 830 | |
amadan