| 1 | // Console-side read path for host/process monitoring: assembling what the |
| 2 | // operator dashboard (Task 11) and owner app-metrics page (Task 12) render |
| 3 | // out of what the box agent publishes (internal/metrics). Nothing here |
| 4 | // writes anything — it is a bucket sweep plus an in-process cache, and the |
| 5 | // wire shapes it reads (metrics.RawChunk, metrics.AlertState) are read |
| 6 | // defensively: an unparsable object is skipped, never fatal to the sweep, |
| 7 | // mirroring alertwatch.go's "never go silent" posture on the same bucket |
| 8 | // prefixes. |
| 9 | package console |
| 10 | |
| 11 | import ( |
| 12 | "context" |
| 13 | "encoding/json" |
| 14 | "errors" |
| 15 | "fmt" |
| 16 | "log" |
| 17 | "os" |
| 18 | "sort" |
| 19 | "strconv" |
| 20 | "strings" |
| 21 | "sync" |
| 22 | "time" |
| 23 | |
| 24 | "github.com/carlosframework/platform/internal/activator" |
| 25 | "github.com/carlosframework/platform/internal/metrics" |
| 26 | ) |
| 27 | |
| 28 | // UnitRow is one per-instance reading, ready to render, plus its own 48h |
| 29 | // sparklines (the owner app-metrics view renders a unit on its own, with no |
| 30 | // host row to borrow a shape from). Keyed by sample name, which is the |
| 31 | // reading's IDENTITY, not always a systemd unit: an exec-backed instance has |
| 32 | // no unit at all — it is a child of `carlos agent`, accounted in a delegated |
| 33 | // cgroup — and reports under carlos-exec@<host> (no .service; exec-child |
| 34 | // accounting spec). Boxes running the superseded pid sampler (issue #118, |
| 35 | // retired by that spec) published exec children under the |
| 36 | // carlos-app@<host>.service name they WOULD have had, and those keys survive |
| 37 | // in raw chunks for the 48h retention window. Either way the route host |
| 38 | // inside the name is the thing to grep the agent's journal for. |
| 39 | type UnitRow struct { |
| 40 | Unit string // full unit name |
| 41 | CPUFrac *float64 |
| 42 | MemBytes *int64 |
| 43 | ReadBytes, WriteBytes *int64 |
| 44 | SparkCPU, SparkMem string // SVG polyline points, "" when no data |
| 45 | } |
| 46 | |
| 47 | // BoxMetricsView is one box's assembled dashboard row: latest host readings |
| 48 | // plus a 48h sparkline per metric plus its units, sorted. |
| 49 | type BoxMetricsView struct { |
| 50 | Box string |
| 51 | At time.Time // latest sample time |
| 52 | CPU, Mem, Disk *float64 |
| 53 | SparkCPU, SparkMem, SparkDisk string // SVG polyline points, "" when no data |
| 54 | Units []UnitRow // sorted by CPUFrac desc, nil-CPU rows last |
| 55 | } |
| 56 | |
| 57 | // AlertRow is one currently-firing rule on one box. LoadMetrics only ever |
| 58 | // returns firing rules — a quiet rule has nothing to show. |
| 59 | type AlertRow struct { |
| 60 | Box, Rule string |
| 61 | Value float64 |
| 62 | FiredAt time.Time |
| 63 | } |
| 64 | |
| 65 | // metricsCacheTTL is how soon after a successful sweep LoadMetrics considers |
| 66 | // the cache FRESH and skips even trying to refresh it. It does NOT bound how |
| 67 | // long a result may be served overall — see metricsStaleServeCeiling for |
| 68 | // that outer bound. Past metricsCacheTTL the cache becomes STALE (still |
| 69 | // served immediately, but a background refresh is triggered) rather than |
| 70 | // unusable. All callers within the FRESH window share the one sweep — the |
| 71 | // operator dashboard and every owner's app-metrics page render off the same |
| 72 | // read. |
| 73 | const metricsCacheTTL = 60 * time.Second |
| 74 | |
| 75 | // metricsStaleServeCeiling bounds how old the cache may be and still be |
| 76 | // handed out immediately while a background refresh runs (LoadMetrics's |
| 77 | // STALE-WITHIN-CEILING state). It exists because nothing keeps the cache |
| 78 | // warm between requests — its age tracks the gap since the last REQUEST, |
| 79 | // not a timer — so without a ceiling, the first request after a quiet |
| 80 | // stretch would serve data that is arbitrarily old (a healthy box's actual |
| 81 | // reporting gap, PLUS however long nobody happened to ask), and a |
| 82 | // persistent bucket outage would let LoadMetrics return stale data with a |
| 83 | // nil error forever, since a background sweep's failure is only logged (see |
| 84 | // startBackgroundSweep) and nothing else ever revisits it. Past this |
| 85 | // ceiling, LoadMetrics falls through to the same blocking single-flight |
| 86 | // path the COLD case uses — costing on the order of a second now that |
| 87 | // sweepBox parallelises its GETs, rather than the several seconds a |
| 88 | // sequential sweep used to cost — in exchange for two things a stale answer |
| 89 | // can't give: data no older than one fresh sweep, and a sweep error |
| 90 | // actually reaching a caller instead of vanishing into a log line. No |
| 91 | // separate failure backoff is needed on top of this: a sweep that keeps |
| 92 | // failing never advances c.at, so the cache's age crosses the ceiling on its |
| 93 | // own within a couple of minutes regardless of how long the outage lasts. |
| 94 | const metricsStaleServeCeiling = 2 * time.Minute |
| 95 | |
| 96 | // metricsSweepHours is how far back LoadMetrics reads raw chunks: the last |
| 97 | // 48 UTC hours, days/ rollups untouched (a later task's job). |
| 98 | const metricsSweepHours = 48 |
| 99 | |
| 100 | // metricsFreshWindow is how old a box's newest sample may be before its |
| 101 | // readings stop being anyone's CURRENT figure. The value is derived, not |
| 102 | // chosen freehand, from three terms: |
| 103 | // |
| 104 | // - Flush cadence, 5 minutes — a FLOOR, not an exact figure: metricsLoop |
| 105 | // sleeps 60s per tick and does per-tick work on top of that (alertTick |
| 106 | // every tick, metricsFlush's own bucket writes every 5th), so the actual |
| 107 | // gap between flushes only ever drifts upward from 5:00, never down. |
| 108 | // - metricsStaleServeCeiling, 2 minutes — the oldest a cache |
| 109 | // LoadMetrics hands back under stale-while-revalidate may be, i.e. how |
| 110 | // far behind the true "latest flush" a box.At a caller sees can sit even |
| 111 | // with a store that is never slow. |
| 112 | // - One more minute of slack, covering both the flush-cadence drift above |
| 113 | // and the ≤metricsSweepTimeout (15s) a background refresh can still be |
| 114 | // mid-flight for when its result gets read. |
| 115 | // |
| 116 | // 5 + 2 + 1 = 8 minutes. This window must exceed that sum or a perfectly |
| 117 | // healthy, perfectly reachable box flaps between reporting and |
| 118 | // no-recent-readings for the tail of every cycle. |
| 119 | // |
| 120 | // It still has to do the job it always did on top of that: the 48h sweep |
| 121 | // keeps serving a dead box's final sample for two days, so without SOME |
| 122 | // bound an instance that vanished on Monday still reads "11% / 1.0 KB, |
| 123 | // current" on Wednesday. Owners get the no-recent-readings state instead |
| 124 | // (they have no timestamp on the page to judge staleness for themselves; the |
| 125 | // operator dashboard prints "Latest <time>" and so is deliberately left |
| 126 | // unbounded). |
| 127 | // |
| 128 | // What this window does NOT promise: it does not survive even ONE missed |
| 129 | // flush — miss a single 5-minute flush and the gap to the NEXT successful |
| 130 | // one is already ~10 minutes, past this window, and the box drops out |
| 131 | // exactly as it would have before this fix. That was true of the old 5-minute window too; widening |
| 132 | // far enough to also cover a missed flush would mean a truly dead box's |
| 133 | // stale reading survives even longer before it stops being anyone's |
| 134 | // "current" figure — not a trade worth making for one skipped beat. |
| 135 | const metricsFreshWindow = 8 * time.Minute |
| 136 | |
| 137 | // metricsBoxesPrefix is where a box publishes its raw chunks and day |
| 138 | // rollups (internal/metrics.RawKey / DayKey); box discovery lists this |
| 139 | // prefix and reads the path segment right after "boxes/". |
| 140 | const metricsBoxesPrefix = "metrics/boxes/" |
| 141 | |
| 142 | // metricsSweepBoxes and metricsSweepReads bound how much of a sweep is in |
| 143 | // flight at once: boxes swept in parallel, and hourly chunks read in parallel |
| 144 | // within a box. A cold sweep is boxes x 48 round trips to the bucket and is |
| 145 | // latency-bound, not CPU-bound — serially that is the whole "Instances page |
| 146 | // is slow" complaint — but it is still someone else's rate limit at the far |
| 147 | // end, so the fan-out is bounded rather than "one goroutine per key". The |
| 148 | // worst case is metricsSweepBoxes x metricsSweepReads requests outstanding. |
| 149 | const ( |
| 150 | metricsSweepBoxes = 4 |
| 151 | metricsSweepReads = 12 |
| 152 | ) |
| 153 | |
| 154 | // metricsSweepTimeout bounds one whole bucket sweep: a List plus 48 Gets per |
| 155 | // box plus the alerts pass. It exists because the sweep no longer runs on the |
| 156 | // requesting page's context (see LoadMetrics) and so has no deadline of its |
| 157 | // own to inherit — a store that accepts connections and then hangs would |
| 158 | // otherwise pin the in-flight flag forever and every later caller with it. |
| 159 | // Fifteen seconds is far beyond a healthy sweep of a fleet this size and far |
| 160 | // short of any browser's patience. |
| 161 | const metricsSweepTimeout = 15 * time.Second |
| 162 | |
| 163 | // metricsCache holds the result of the last successful bucket sweep. A |
| 164 | // failed sweep never reaches here — a failed BLOCKING sweep (c.at still |
| 165 | // zero, OR the cache is stale past metricsStaleServeCeiling) returns the |
| 166 | // error to its caller and leaves the cache exactly as it was; a failed |
| 167 | // BACKGROUND sweep (c.at set and still within the ceiling) is only logged, |
| 168 | // since nobody is waiting on it and the stale cache it leaves behind is |
| 169 | // still valid data, just not fresher. |
| 170 | // |
| 171 | // No separate failure backoff sits on top of the ceiling: a sweep that keeps |
| 172 | // failing never advances c.at, so the cache's age crosses |
| 173 | // metricsStaleServeCeiling on its own within a couple of minutes regardless |
| 174 | // of how long an outage lasts, and every caller after that lands back on the |
| 175 | // blocking path where the failure is visible again — see |
| 176 | // metricsStaleServeCeiling's own comment. |
| 177 | // |
| 178 | // mu guards the fields and is deliberately NOT held across the sweep's bucket |
| 179 | // IO; sweeping/done are what serialise the IO instead, so that a cold sweep |
| 180 | // delays only the callers that actually need its result rather than every |
| 181 | // caller that touches the cache. sweeping is true while one goroutine (either |
| 182 | // a blocking caller's own, or the detached refresh goroutine LoadMetrics |
| 183 | // launches for stale-while-revalidate) is out at the bucket, and done is the |
| 184 | // channel it closes when it comes back — the SAME channel and flag serve |
| 185 | // both shapes of sweep, so a caller arriving mid-refresh, whichever kind |
| 186 | // started it, coalesces onto it rather than starting a second one. |
| 187 | type metricsCache struct { |
| 188 | mu sync.Mutex |
| 189 | at time.Time |
| 190 | boxes []BoxMetricsView |
| 191 | alerts []AlertRow |
| 192 | sweeping bool |
| 193 | done chan struct{} |
| 194 | } |
| 195 | |
| 196 | // LoadMetrics reads every box under metrics/boxes/, the raw chunks of the |
| 197 | // last 48 hours (missing hours skipped silently), and alerts/boxes/*.json, |
| 198 | // caching the result in-process so concurrent callers — the dashboard page |
| 199 | // and every owner's app-metrics page — usually share one sweep. |
| 200 | // |
| 201 | // Three cache states: |
| 202 | // |
| 203 | // - FRESH (c.at within metricsCacheTTL): the cached answer, immediately, |
| 204 | // no bucket touched at all. |
| 205 | // - STALE-WITHIN-CEILING (c.at is set, older than metricsCacheTTL, but no |
| 206 | // older than metricsStaleServeCeiling): the cached answer is STILL |
| 207 | // returned immediately — stale-while-revalidate — and, if no sweep is |
| 208 | // already in flight, a fresh one is kicked off on a detached goroutine |
| 209 | // (startBackgroundSweep) for the NEXT caller to benefit from. This is |
| 210 | // the fix for the sweep's own cost: sweepBox's 48 hourly GETs for a |
| 211 | // single box were measured at ~2.5s run sequentially, and that cost |
| 212 | // used to land inline on whichever request's bad luck it was to arrive |
| 213 | // right after a cache expiry — the dashboard page and every tenant's |
| 214 | // app page alike. |
| 215 | // - COLD (c.at is zero — nothing has ever swept successfully) or STALE |
| 216 | // PAST THE CEILING (c.at is set but older than |
| 217 | // metricsStaleServeCeiling — nothing has kept the cache warm for a |
| 218 | // while, e.g. a quiet console between requests, or every sweep since |
| 219 | // has been failing): the ORIGINAL blocking behaviour. There is either no |
| 220 | // servable stale answer (COLD) or the one that exists is too old to |
| 221 | // hand out one more time (STALE-PAST-CEILING) — either way this caller |
| 222 | // (or whichever caller's sweep is already in flight) waits for a real |
| 223 | // sweep and gets ITS error if the sweep fails. The ceiling is what keeps |
| 224 | // a persistent bucket outage visible: without it, stale-while-revalidate |
| 225 | // alone would let LoadMetrics return stale data with a nil error |
| 226 | // forever regardless of how long the store stays down, because a |
| 227 | // background sweep's failure is only logged (see startBackgroundSweep) |
| 228 | // — crossing the ceiling forces the next caller back onto the path that |
| 229 | // actually surfaces the error. Blocking here now costs on the order of |
| 230 | // a second for a fleet this size rather than the several seconds it |
| 231 | // cost before sweepBox parallelised its GETs, which is what makes |
| 232 | // paying that cost here — instead of serving indefinitely-stale data — |
| 233 | // an acceptable trade. |
| 234 | // |
| 235 | // Two properties matter beyond the cache states themselves, both learned |
| 236 | // from the shape this replaced (mutex held across the whole sweep, on the |
| 237 | // request's own context): |
| 238 | // |
| 239 | // - The cache mutex is never held across bucket IO. Only the goroutine that |
| 240 | // actually sweeps is slow; everyone else either takes the cached answer |
| 241 | // immediately or waits on the in-flight sweep's channel (only the |
| 242 | // blocking case waits at all), and neither blocks a caller whose cache |
| 243 | // is fresh or stale-but-servable. handleApp calls this on every tenant's |
| 244 | // app page, so a cold sweep must not become a queue every page on the |
| 245 | // console stands in. |
| 246 | // - Every sweep — blocking or background — runs on a context detached from |
| 247 | // any particular caller's: context.WithoutCancel of whatever was handed |
| 248 | // in (context.Background() for the background case, since there is no |
| 249 | // caller to derive from), plus metricsSweepTimeout. A request cancelled |
| 250 | // mid-sweep (a reload, a navigate-away) would otherwise abort a sweep |
| 251 | // other callers are waiting on AND, since a failed blocking sweep is |
| 252 | // never cached, leave the next request to start from scratch: reload |
| 253 | // impatiently enough and the cache never populates at all. The caller's |
| 254 | // own context still governs WAITING in the blocking case (a cancelled |
| 255 | // waiter gives up rather than hanging), it just never governs the sweep |
| 256 | // itself. |
| 257 | func (s *Server) LoadMetrics(ctx context.Context) ([]BoxMetricsView, []AlertRow, error) { |
| 258 | c := &s.metricsCache |
| 259 | for { |
| 260 | c.mu.Lock() |
| 261 | if !c.at.IsZero() && time.Since(c.at) < metricsCacheTTL { |
| 262 | // FRESH. |
| 263 | boxes, alerts := c.boxes, c.alerts |
| 264 | c.mu.Unlock() |
| 265 | return boxes, alerts, nil |
| 266 | } |
| 267 | if !c.at.IsZero() && time.Since(c.at) <= metricsStaleServeCeiling { |
| 268 | // STALE-WITHIN-CEILING: hand back what we have right now — it |
| 269 | // is still a real sweep's result, just older than |
| 270 | // metricsCacheTTL — and let a refresh happen off to the side |
| 271 | // rather than making this caller pay for it. If one is already |
| 272 | // running (background OR, in principle, another blocking |
| 273 | // caller's), don't start a second; this caller just doesn't |
| 274 | // wait for it either way. |
| 275 | boxes, alerts := c.boxes, c.alerts |
| 276 | if !c.sweeping { |
| 277 | s.startBackgroundSweep(c) |
| 278 | } |
| 279 | c.mu.Unlock() |
| 280 | return boxes, alerts, nil |
| 281 | } |
| 282 | // COLD, or STALE-PAST-THE-CEILING: no servable stale answer exists |
| 283 | // (COLD) or the one that does is too old to hand out again |
| 284 | // (past metricsStaleServeCeiling) — this caller must wait for real |
| 285 | // data and see its error if the sweep fails, exactly as the |
| 286 | // original blocking behaviour did. |
| 287 | if c.sweeping { |
| 288 | // Someone is already at the bucket: wait for their result |
| 289 | // rather than duplicating the sweep, then re-check (their |
| 290 | // sweep may have failed, in which case this iteration takes |
| 291 | // the work over). |
| 292 | wait := c.done |
| 293 | c.mu.Unlock() |
| 294 | select { |
| 295 | case <-wait: |
| 296 | continue |
| 297 | case <-ctx.Done(): |
| 298 | return nil, nil, ctx.Err() |
| 299 | } |
| 300 | } |
| 301 | c.sweeping = true |
| 302 | c.done = make(chan struct{}) |
| 303 | done := c.done |
| 304 | c.mu.Unlock() |
| 305 | |
| 306 | // err starts non-nil so that if sweepAll panics (unwinding past the |
| 307 | // assignment below), the deferred bookkeeping still sees a failed |
| 308 | // sweep: nothing gets cached and the in-flight flag is cleared |
| 309 | // instead of wedging every future caller. net/http's per-connection |
| 310 | // recover keeps the process up; this keeps the cache usable. |
| 311 | var ( |
| 312 | boxes []BoxMetricsView |
| 313 | alerts []AlertRow |
| 314 | err = errors.New("console: metrics sweep did not complete") |
| 315 | ) |
| 316 | func() { |
| 317 | defer func() { |
| 318 | c.mu.Lock() |
| 319 | if err == nil { |
| 320 | // Only a fully successful sweep is cached; a failed |
| 321 | // one leaves the previous cache exactly as it was. |
| 322 | c.at = time.Now() |
| 323 | c.boxes = boxes |
| 324 | c.alerts = alerts |
| 325 | } |
| 326 | c.sweeping = false |
| 327 | close(done) |
| 328 | c.mu.Unlock() |
| 329 | }() |
| 330 | boxes, alerts, err = s.sweepAll(ctx) |
| 331 | }() |
| 332 | |
| 333 | if err != nil { |
| 334 | return nil, nil, err |
| 335 | } |
| 336 | return boxes, alerts, nil |
| 337 | } |
| 338 | } |
| 339 | |
| 340 | // startBackgroundSweep launches sweepAll on its own goroutine to refresh a |
| 341 | // cache that has swept successfully at least once and is still within |
| 342 | // metricsStaleServeCeiling, without making any caller wait on it — see |
| 343 | // LoadMetrics's STALE-WITHIN-CEILING case. "Swept successfully at least |
| 344 | // once" is exactly c.at being set: an empty-bucket console (c.at set, |
| 345 | // c.boxes nil) takes this path just like a populated one, since nil boxes |
| 346 | // is itself a complete, valid swept result, not an absent one. Must be |
| 347 | // called with c.mu held: it sets c.sweeping and installs c.done itself, the |
| 348 | // very same fields the blocking path uses, so a caller arriving while this |
| 349 | // is in flight (whether via the STALE-WITHIN-CEILING branch or the blocking |
| 350 | // one) coalesces onto it instead of starting a second sweep. sweepAll is |
| 351 | // handed context.Background() rather than any request's |
| 352 | // context because there IS no request behind this goroutine — it already |
| 353 | // detaches internally via context.WithoutCancel and bounds itself with |
| 354 | // metricsSweepTimeout, so Background() costs nothing extra. |
| 355 | // |
| 356 | // Unlike the blocking path, this goroutine has no net/http per-connection |
| 357 | // recover sitting above it — an unrecovered panic here would take the whole |
| 358 | // process down, not just one request — so it recovers explicitly. A failed |
| 359 | // or panicked background sweep is only logged: no caller is waiting on its |
| 360 | // error, and the stale cache it leaves untouched is still valid data. |
| 361 | func (s *Server) startBackgroundSweep(c *metricsCache) { |
| 362 | c.sweeping = true |
| 363 | c.done = make(chan struct{}) |
| 364 | done := c.done |
| 365 | go func() { |
| 366 | var ( |
| 367 | boxes []BoxMetricsView |
| 368 | alerts []AlertRow |
| 369 | err = errors.New("console: background metrics sweep did not complete") |
| 370 | ) |
| 371 | defer func() { |
| 372 | rec := recover() |
| 373 | c.mu.Lock() |
| 374 | if err == nil { |
| 375 | c.at = time.Now() |
| 376 | c.boxes = boxes |
| 377 | c.alerts = alerts |
| 378 | } |
| 379 | c.sweeping = false |
| 380 | close(done) |
| 381 | c.mu.Unlock() |
| 382 | switch { |
| 383 | case rec != nil: |
| 384 | log.Printf("console: background metrics sweep panic, cache left at its last good sweep: %v", rec) |
| 385 | case err != nil: |
| 386 | log.Printf("console: background metrics sweep: %v", err) |
| 387 | } |
| 388 | }() |
| 389 | boxes, alerts, err = s.sweepAll(context.Background()) |
| 390 | }() |
| 391 | } |
| 392 | |
| 393 | // sweepAll is the bucket half of LoadMetrics: discovery, every box's 48h raw |
| 394 | // sweep (metricsSweepBoxes of them at a time), then the alerts pass, on a |
| 395 | // context detached from the caller's (WithoutCancel) and bounded by |
| 396 | // metricsSweepTimeout. It holds no lock — see LoadMetrics's doc comment for |
| 397 | // why that is the point. |
| 398 | // |
| 399 | // Boxes are swept concurrently but land in boxNames order, never completion |
| 400 | // order: the dashboard's row order must not depend on which box's bucket |
| 401 | // answered first. Any box failing fails the whole sweep, with the |
| 402 | // lowest-ordered box's error — the same error the serial sweep returned, so a |
| 403 | // flaky box reads the same in the logs however the reads interleaved. |
| 404 | func (s *Server) sweepAll(ctx context.Context) ([]BoxMetricsView, []AlertRow, error) { |
| 405 | sweepCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), metricsSweepTimeout) |
| 406 | defer cancel() |
| 407 | |
| 408 | boxNames, err := s.discoverBoxes(sweepCtx) |
| 409 | if err != nil { |
| 410 | return nil, nil, err |
| 411 | } |
| 412 | |
| 413 | views := make([]BoxMetricsView, len(boxNames)) |
| 414 | errs := make([]error, len(boxNames)) |
| 415 | var ( |
| 416 | wg sync.WaitGroup |
| 417 | sem = make(chan struct{}, metricsSweepBoxes) |
| 418 | once sync.Once |
| 419 | failed = make(chan struct{}) |
| 420 | ) |
| 421 | for i, box := range boxNames { |
| 422 | wg.Add(1) |
| 423 | go func() { |
| 424 | defer wg.Done() |
| 425 | select { |
| 426 | case sem <- struct{}{}: |
| 427 | case <-failed: |
| 428 | // Another box has already failed the sweep; nothing this one |
| 429 | // could read would be returned. Boxes already at the bucket |
| 430 | // still finish — they run on sweepCtx, so the error a failed |
| 431 | // sweep reports is always a box's own, never one this |
| 432 | // function inflicted. |
| 433 | return |
| 434 | } |
| 435 | defer func() { <-sem }() |
| 436 | view, err := s.sweepBox(sweepCtx, box) |
| 437 | if err != nil { |
| 438 | errs[i] = err |
| 439 | once.Do(func() { close(failed) }) |
| 440 | return |
| 441 | } |
| 442 | views[i] = view |
| 443 | }() |
| 444 | } |
| 445 | wg.Wait() |
| 446 | for _, err := range errs { |
| 447 | if err != nil { |
| 448 | return nil, nil, err |
| 449 | } |
| 450 | } |
| 451 | |
| 452 | alerts, err := s.sweepAlerts(sweepCtx) |
| 453 | if err != nil { |
| 454 | return nil, nil, err |
| 455 | } |
| 456 | return views, alerts, nil |
| 457 | } |
| 458 | |
| 459 | // discoverBoxes lists metrics/boxes/ and returns the deduped, sorted set of |
| 460 | // box names — the path segment right after "boxes/". |
| 461 | func (s *Server) discoverBoxes(ctx context.Context) ([]string, error) { |
| 462 | keys, err := s.cfg.Releases.List(ctx, metricsBoxesPrefix) |
| 463 | if err != nil { |
| 464 | return nil, fmt.Errorf("list %s: %w", metricsBoxesPrefix, err) |
| 465 | } |
| 466 | seen := make(map[string]bool) |
| 467 | var boxes []string |
| 468 | for _, key := range keys { |
| 469 | rest := strings.TrimPrefix(key, metricsBoxesPrefix) |
| 470 | idx := strings.Index(rest, "/") |
| 471 | if idx <= 0 { |
| 472 | continue // malformed/unexpected key shape directly under the prefix |
| 473 | } |
| 474 | box := rest[:idx] |
| 475 | if seen[box] { |
| 476 | continue |
| 477 | } |
| 478 | seen[box] = true |
| 479 | boxes = append(boxes, box) |
| 480 | } |
| 481 | sort.Strings(boxes) |
| 482 | return boxes, nil |
| 483 | } |
| 484 | |
| 485 | // sweepBox reads box's raw chunks for the last 48 UTC hours, metricsSweepReads |
| 486 | // at a time, and assembles its BoxMetricsView. A missing hour (os.ErrNotExist) |
| 487 | // or an unparsable chunk is skipped silently — a box that has been up for |
| 488 | // under 48h, or whose agent hiccuped on one hour, still renders on what it |
| 489 | // does have. Any other read error fails the whole sweep (and so is never |
| 490 | // cached), reporting the newest failing hour's error as the serial read did. |
| 491 | // |
| 492 | // Each hour lands in its own slot and the slots are concatenated in hour |
| 493 | // order before sorting, because the sort below is NOT stable: samples sharing |
| 494 | // a timestamp keep whatever relative order they were appended in, so appending |
| 495 | // as the reads complete would let a box's current reading flap between |
| 496 | // renders. |
| 497 | func (s *Server) sweepBox(ctx context.Context, box string) (BoxMetricsView, error) { |
| 498 | now := time.Now().UTC() |
| 499 | hourly := make([][]metrics.Sample, metricsSweepHours) |
| 500 | errs := make([]error, metricsSweepHours) |
| 501 | var ( |
| 502 | wg sync.WaitGroup |
| 503 | sem = make(chan struct{}, metricsSweepReads) |
| 504 | once sync.Once |
| 505 | failed = make(chan struct{}) |
| 506 | ) |
| 507 | for i := range metricsSweepHours { |
| 508 | wg.Add(1) |
| 509 | go func() { |
| 510 | defer wg.Done() |
| 511 | select { |
| 512 | case sem <- struct{}{}: |
| 513 | case <-failed: |
| 514 | return // an earlier hour already failed the sweep |
| 515 | } |
| 516 | defer func() { <-sem }() |
| 517 | hour := metrics.HourOf(now.Add(-time.Duration(i) * time.Hour)) |
| 518 | key := metrics.RawKey(box, hour) |
| 519 | body, _, err := s.cfg.Releases.Get(ctx, key) |
| 520 | if err != nil { |
| 521 | if errors.Is(err, os.ErrNotExist) { |
| 522 | return |
| 523 | } |
| 524 | errs[i] = fmt.Errorf("read %s: %w", key, err) |
| 525 | once.Do(func() { close(failed) }) |
| 526 | return |
| 527 | } |
| 528 | var chunk metrics.RawChunk |
| 529 | if err := json.Unmarshal(body, &chunk); err != nil { |
| 530 | return // unparsable chunk: skip, don't fail the sweep |
| 531 | } |
| 532 | hourly[i] = chunk.Samples |
| 533 | }() |
| 534 | } |
| 535 | wg.Wait() |
| 536 | for _, err := range errs { |
| 537 | if err != nil { |
| 538 | return BoxMetricsView{}, err |
| 539 | } |
| 540 | } |
| 541 | |
| 542 | var samples []metrics.Sample |
| 543 | for _, h := range hourly { |
| 544 | samples = append(samples, h...) |
| 545 | } |
| 546 | sort.Slice(samples, func(i, j int) bool { return samples[i].At.Before(samples[j].At) }) |
| 547 | return buildBoxView(box, samples), nil |
| 548 | } |
| 549 | |
| 550 | // buildBoxView assembles one box's view from its swept samples, oldest |
| 551 | // first. The latest sample's own values are used as-is for CPU/Mem/Disk/At |
| 552 | // — nil stays nil; this is deliberately NOT "each field's newest non-nil |
| 553 | // reading independently", which would quietly mix readings from different |
| 554 | // points in time. |
| 555 | func buildBoxView(box string, samples []metrics.Sample) BoxMetricsView { |
| 556 | view := BoxMetricsView{Box: box} |
| 557 | if len(samples) == 0 { |
| 558 | return view |
| 559 | } |
| 560 | latest := samples[len(samples)-1] |
| 561 | view.At = latest.At |
| 562 | view.CPU = latest.Host.CPUFrac |
| 563 | view.Mem = latest.Host.MemFrac |
| 564 | view.Disk = latest.Host.DiskFrac |
| 565 | |
| 566 | cpu := make([]*float64, len(samples)) |
| 567 | mem := make([]*float64, len(samples)) |
| 568 | disk := make([]*float64, len(samples)) |
| 569 | for i, sm := range samples { |
| 570 | cpu[i] = sm.Host.CPUFrac |
| 571 | mem[i] = sm.Host.MemFrac |
| 572 | disk[i] = sm.Host.DiskFrac |
| 573 | } |
| 574 | view.SparkCPU = sparkPoints(cpu, 240, 40) |
| 575 | view.SparkMem = sparkPoints(mem, 240, 40) |
| 576 | view.SparkDisk = sparkPoints(disk, 240, 40) |
| 577 | |
| 578 | view.Units = buildUnitRows(samples) |
| 579 | return view |
| 580 | } |
| 581 | |
| 582 | // buildUnitRows turns the latest sample's Units map into rows sorted by |
| 583 | // CPUFrac descending, nil-CPU rows last, stable by unit name within ties, |
| 584 | // each carrying its own sparklines drawn across the whole swept window. A |
| 585 | // unit absent from the latest sample's map simply isn't in this slice — |
| 586 | // never rendered, never aggregated as a zero (Task 4's "missing map key is |
| 587 | // no reading" ruling, same map shape here) — even if it appears in older |
| 588 | // samples: the row set is "what is running now", not "what ever ran". |
| 589 | func buildUnitRows(samples []metrics.Sample) []UnitRow { |
| 590 | if len(samples) == 0 { |
| 591 | return nil |
| 592 | } |
| 593 | units := samples[len(samples)-1].Units |
| 594 | rows := make([]UnitRow, 0, len(units)) |
| 595 | for name, u := range units { |
| 596 | rows = append(rows, UnitRow{ |
| 597 | Unit: name, |
| 598 | CPUFrac: u.CPUFrac, |
| 599 | MemBytes: u.MemBytes, |
| 600 | ReadBytes: u.ReadBytes, |
| 601 | WriteBytes: u.WriteBytes, |
| 602 | SparkCPU: sparkPoints(unitCPUSeries(samples, name), 240, 40), |
| 603 | SparkMem: sparkPoints(unitMemSeries(samples, name), 240, 40), |
| 604 | }) |
| 605 | } |
| 606 | // Name order first so the CPU-desc sort below, being stable, ties out |
| 607 | // alphabetically rather than on map iteration order. |
| 608 | sort.Slice(rows, func(i, j int) bool { return rows[i].Unit < rows[j].Unit }) |
| 609 | sort.SliceStable(rows, func(i, j int) bool { |
| 610 | a, b := rows[i].CPUFrac, rows[j].CPUFrac |
| 611 | if a == nil { |
| 612 | return false // nil never sorts before anything, including another nil |
| 613 | } |
| 614 | if b == nil { |
| 615 | return true |
| 616 | } |
| 617 | return *a > *b |
| 618 | }) |
| 619 | return rows |
| 620 | } |
| 621 | |
| 622 | // unitCPUSeries is one unit's CPU fraction at each swept sample index. A |
| 623 | // sample where the unit is absent, or present without a CPU reading, stays |
| 624 | // nil — sparkPoints draws a gap, never a zero. |
| 625 | func unitCPUSeries(samples []metrics.Sample, unit string) []*float64 { |
| 626 | out := make([]*float64, len(samples)) |
| 627 | for i, sm := range samples { |
| 628 | if u, ok := sm.Units[unit]; ok { |
| 629 | out[i] = u.CPUFrac |
| 630 | } |
| 631 | } |
| 632 | return out |
| 633 | } |
| 634 | |
| 635 | // unitMemSeries is one unit's memory at each swept sample index, scaled to |
| 636 | // its OWN 48h peak. Unlike CPU, a unit's memory has no 0..1 ceiling the |
| 637 | // console can know — the cgroup limit isn't published, and the box's total |
| 638 | // RAM is deliberately not in an owner's view — so the sparkline shows the |
| 639 | // shape of this instance's memory against its own high-water mark, and the |
| 640 | // number beside it stays the absolute byte count. Absent/nil readings stay |
| 641 | // nil; a unit that never reported memory gets no line at all. |
| 642 | func unitMemSeries(samples []metrics.Sample, unit string) []*float64 { |
| 643 | var peak int64 |
| 644 | for _, sm := range samples { |
| 645 | if u, ok := sm.Units[unit]; ok && u.MemBytes != nil && *u.MemBytes > peak { |
| 646 | peak = *u.MemBytes |
| 647 | } |
| 648 | } |
| 649 | if peak <= 0 { |
| 650 | return nil |
| 651 | } |
| 652 | out := make([]*float64, len(samples)) |
| 653 | for i, sm := range samples { |
| 654 | u, ok := sm.Units[unit] |
| 655 | if !ok || u.MemBytes == nil { |
| 656 | continue |
| 657 | } |
| 658 | f := float64(*u.MemBytes) / float64(peak) |
| 659 | out[i] = &f |
| 660 | } |
| 661 | return out |
| 662 | } |
| 663 | |
| 664 | // InstanceMetrics is one of an APP OWNER's instances, named the way owners |
| 665 | // name their instances: by route host. It carries that instance's unit |
| 666 | // readings and nothing else — no host totals, no sibling units, no alerts. |
| 667 | // Box is here for internal use (picking the freshest reporting box when a |
| 668 | // host has moved); the app page must not print it, because which box a |
| 669 | // tenant sits on is not an owner's business. |
| 670 | type InstanceMetrics struct { |
| 671 | ObservedAt time.Time |
| 672 | |
| 673 | Host string // the route host — how owners name their instances |
| 674 | Box string |
| 675 | CPUFrac *float64 |
| 676 | MemBytes *int64 |
| 677 | |
| 678 | SparkCPU, SparkMem string |
| 679 | |
| 680 | // Hibernate is the route's hibernation setting, copied in by the caller |
| 681 | // (handleApp) — the sweep cannot see it. It only matters when Reporting |
| 682 | // is false, where it decides what the absence of readings may claim: a |
| 683 | // hibernating instance is presumed asleep, anything else is NOT — it may |
| 684 | // be serving right now from outside per-instance accounting (a bespoke |
| 685 | // unit like keymail-relay.service, or an exec child on a box whose agent |
| 686 | // has no delegated subtree yet), and the template must not assert a |
| 687 | // sleep state nobody observed. |
| 688 | Hibernate bool |
| 689 | } |
| 690 | |
| 691 | // Reporting is the template's one question: did any box report this |
| 692 | // instance's unit in the last sweep? False is the no-readings case, whose |
| 693 | // rendering splits on Hibernate: hibernating instances read "asleep — no |
| 694 | // recent readings", everything else the state-neutral "no per-instance |
| 695 | // readings" — never zeros either way. (Reporting with |
| 696 | // nil readings is a different, honest state: the unit is up but that metric |
| 697 | // was unreadable, which renders as the usual em-dash.) |
| 698 | func (m InstanceMetrics) Reporting() bool { return m.Box != "" } |
| 699 | |
| 700 | // OwnerAppMetrics filters the cached 48h sweep down to EXACTLY the given |
| 701 | // hosts' own per-instance readings — carlos-app@<host>.service for a |
| 702 | // unit-backed tenant, carlos-exec@<host> for an exec-backed one, or the |
| 703 | // operator-stamped bespoke unit named in unitByHost[host] (registry |
| 704 | // Route.Unit, #110) when the caller supplies one. It returns unit-level |
| 705 | // rows only — no host totals, no other units, no alert data. The allowlist |
| 706 | // is the hosts slice the caller derived from the app's OWN routes, plus |
| 707 | // unitByHost entries drawn from those same routes; nothing else can enter |
| 708 | // the view. |
| 709 | // |
| 710 | // Every allowlisted host gets exactly one row, in host order, whether or not |
| 711 | // the sweep saw it: a host with no unit data is an UNKNOWN row (nil |
| 712 | // readings, Reporting false), never zeros and never silently dropped — an |
| 713 | // owner whose instance is asleep must still see it listed. |
| 714 | func (s *Server) OwnerAppMetrics(ctx context.Context, hosts []string, unitByHost map[string]string) ([]InstanceMetrics, error) { |
| 715 | // The allowlist, keyed by the unit names each host can legally match. |
| 716 | // Matching is exact-string on this map: a unit the caller did not ask |
| 717 | // for cannot be reached by any path below. |
| 718 | want := make(map[string]string, 2*len(hosts)) |
| 719 | for _, h := range hosts { |
| 720 | if h == "" { |
| 721 | continue |
| 722 | } |
| 723 | // Two legal keys per host: the systemd unit a unit-backed tenant |
| 724 | // reports under, and the delegated exec cgroup an exec-backed one |
| 725 | // does. One route, one backing — a host can't legally report under |
| 726 | // both at once; if both ever appear in a sweep, the freshest-box |
| 727 | // rule below picks deterministically, no special case. |
| 728 | want[activator.UnitName(h)] = h |
| 729 | want[activator.ExecName(h)] = h |
| 730 | // The third legal key: an operator-stamped bespoke unit (registry |
| 731 | // Route.Unit, #110). Opt-in per host — the caller maps host → |
| 732 | // stamped unit from the route rows it already holds; an empty map |
| 733 | // costs nothing and matches nothing new. (A nil map indexes fine |
| 734 | // in Go — no guard needed.) |
| 735 | if u := unitByHost[h]; u != "" { |
| 736 | want[u] = h |
| 737 | } |
| 738 | } |
| 739 | if len(want) == 0 { |
| 740 | return nil, nil |
| 741 | } |
| 742 | |
| 743 | boxes, _, err := s.LoadMetrics(ctx) |
| 744 | if err != nil { |
| 745 | return nil, err |
| 746 | } |
| 747 | |
| 748 | found := make(map[string]InstanceMetrics, len(want)) |
| 749 | at := make(map[string]time.Time, len(want)) |
| 750 | now := time.Now() |
| 751 | for _, box := range boxes { |
| 752 | // A box that stopped publishing keeps its final sample inside the |
| 753 | // 48h sweep for two days; those numbers are history, not this |
| 754 | // instance's current figure, so the whole box drops out of the |
| 755 | // owner view and its instances read as no-recent-readings. |
| 756 | if box.At.IsZero() || now.Sub(box.At) > metricsFreshWindow { |
| 757 | continue |
| 758 | } |
| 759 | for _, u := range box.Units { |
| 760 | host, ok := want[u.Unit] |
| 761 | if !ok { |
| 762 | continue // not this app's — the whole boundary is this line |
| 763 | } |
| 764 | // A host that reports on two boxes at once (a move in flight) |
| 765 | // shows the freshest box's reading, not whichever sorted first. |
| 766 | if prev, taken := at[host]; taken && !box.At.After(prev) { |
| 767 | continue |
| 768 | } |
| 769 | at[host] = box.At |
| 770 | found[host] = InstanceMetrics{ |
| 771 | Host: host, |
| 772 | ObservedAt: box.At, |
| 773 | Box: box.Box, |
| 774 | CPUFrac: u.CPUFrac, |
| 775 | MemBytes: u.MemBytes, |
| 776 | SparkCPU: u.SparkCPU, |
| 777 | SparkMem: u.SparkMem, |
| 778 | } |
| 779 | } |
| 780 | } |
| 781 | |
| 782 | // Iterate the callers' hosts, not want's values: want now carries TWO |
| 783 | // keys per host, and "every allowlisted host gets exactly one row" is a |
| 784 | // promise about hosts, not keys. seen collapses duplicate input hosts, |
| 785 | // which the old unique-keyed map did implicitly. |
| 786 | rows := make([]InstanceMetrics, 0, len(hosts)) |
| 787 | seen := make(map[string]bool, len(hosts)) |
| 788 | for _, h := range hosts { |
| 789 | if h == "" || seen[h] { |
| 790 | continue |
| 791 | } |
| 792 | seen[h] = true |
| 793 | if row, ok := found[h]; ok { |
| 794 | rows = append(rows, row) |
| 795 | } else { |
| 796 | rows = append(rows, InstanceMetrics{Host: h}) |
| 797 | } |
| 798 | } |
| 799 | sort.Slice(rows, func(i, j int) bool { return rows[i].Host < rows[j].Host }) |
| 800 | return rows, nil |
| 801 | } |
| 802 | |
| 803 | // sweepAlerts reads every alerts/boxes/*.json and returns one AlertRow per |
| 804 | // currently-firing rule. Unreadable-but-present keys fail the sweep (so it |
| 805 | // is never cached half-done); an unparsable object is skipped, matching |
| 806 | // alertPass's posture on the same prefix. |
| 807 | func (s *Server) sweepAlerts(ctx context.Context) ([]AlertRow, error) { |
| 808 | keys, err := s.cfg.Releases.List(ctx, alertPrefix) |
| 809 | if err != nil { |
| 810 | return nil, fmt.Errorf("list %s: %w", alertPrefix, err) |
| 811 | } |
| 812 | var rows []AlertRow |
| 813 | for _, key := range keys { |
| 814 | if !strings.HasSuffix(key, ".json") { |
| 815 | continue |
| 816 | } |
| 817 | body, _, err := s.cfg.Releases.Get(ctx, key) |
| 818 | if err != nil { |
| 819 | if errors.Is(err, os.ErrNotExist) { |
| 820 | continue |
| 821 | } |
| 822 | return nil, fmt.Errorf("read %s: %w", key, err) |
| 823 | } |
| 824 | var st metrics.AlertState |
| 825 | if err := json.Unmarshal(body, &st); err != nil { |
| 826 | continue // unparsable: skip, don't fail the sweep |
| 827 | } |
| 828 | box := st.Box |
| 829 | if box == "" { |
| 830 | // Trust the key over an empty body field, same as alertPass. |
| 831 | box = strings.TrimSuffix(strings.TrimPrefix(key, alertPrefix), ".json") |
| 832 | } |
| 833 | names := make([]string, 0, len(st.Rules)) |
| 834 | for name := range st.Rules { |
| 835 | names = append(names, name) |
| 836 | } |
| 837 | sort.Strings(names) |
| 838 | for _, name := range names { |
| 839 | rs := st.Rules[name] |
| 840 | if !rs.Firing { |
| 841 | continue |
| 842 | } |
| 843 | rows = append(rows, AlertRow{Box: box, Rule: name, Value: rs.Value, FiredAt: rs.FiredAt}) |
| 844 | } |
| 845 | } |
| 846 | return rows, nil |
| 847 | } |
| 848 | |
| 849 | // sparkPoints maps vals (up to 48h of minute samples, one slot per sample |
| 850 | // index) onto an SVG polyline "points" string on a w x h viewBox. Values are |
| 851 | // fractions and clamp into [0,1] if out of range; y is inverted (SVG's y |
| 852 | // axis points down, so a higher fraction plots at a SMALLER y — visually |
| 853 | // "up"); x is spread evenly across the width by the sample's index, not by |
| 854 | // elapsed time. A nil sample contributes no pair at all — its x slot is |
| 855 | // simply skipped, so the polyline draws a straight line across the gap; |
| 856 | // breaking the line at gaps is not attempted in v1 (documented here per the |
| 857 | // brief). Empty input or input that is nil throughout returns "". |
| 858 | // |
| 859 | // Input longer than w is bucketed down to w points first (see downsample): |
| 860 | // 48h of minute samples is ~2880 readings into a 240px box — twelve points |
| 861 | // per pixel of detail nobody can see, paid for in a coordinate string held |
| 862 | // in the shared cache for every metric of every unit on every box. |
| 863 | func sparkPoints(vals []*float64, w, h int) string { |
| 864 | if len(vals) == 0 { |
| 865 | return "" |
| 866 | } |
| 867 | vals = downsample(vals, w) |
| 868 | var b strings.Builder |
| 869 | n := len(vals) |
| 870 | for i, v := range vals { |
| 871 | if v == nil { |
| 872 | continue |
| 873 | } |
| 874 | f := *v |
| 875 | if f < 0 { |
| 876 | f = 0 |
| 877 | } else if f > 1 { |
| 878 | f = 1 |
| 879 | } |
| 880 | x := 0.0 |
| 881 | if n > 1 { |
| 882 | x = float64(i) / float64(n-1) * float64(w) |
| 883 | } |
| 884 | y := float64(h) - f*float64(h) |
| 885 | if b.Len() > 0 { |
| 886 | b.WriteByte(' ') |
| 887 | } |
| 888 | b.WriteString(strconv.FormatFloat(x, 'f', 2, 64)) |
| 889 | b.WriteByte(',') |
| 890 | b.WriteString(strconv.FormatFloat(y, 'f', 2, 64)) |
| 891 | } |
| 892 | return b.String() |
| 893 | } |
| 894 | |
| 895 | // downsample buckets vals into at most max evenly-sized buckets, each |
| 896 | // bucket's value being the MEAN of its non-nil readings — a spike lifts its |
| 897 | // bucket rather than disappearing, and a bucket with nothing readable in it |
| 898 | // stays nil, so "unknown" survives bucketing as a gap and never becomes a |
| 899 | // zero. Input at or under max is returned untouched (the common case: a box |
| 900 | // that has been up for minutes, and every unit test's handful of samples). |
| 901 | func downsample(vals []*float64, max int) []*float64 { |
| 902 | if max <= 0 || len(vals) <= max { |
| 903 | return vals |
| 904 | } |
| 905 | n := len(vals) |
| 906 | out := make([]*float64, max) |
| 907 | for b := 0; b < max; b++ { |
| 908 | lo := b * n / max |
| 909 | hi := (b + 1) * n / max |
| 910 | if hi > n { |
| 911 | hi = n |
| 912 | } |
| 913 | var sum float64 |
| 914 | var cnt int |
| 915 | for _, v := range vals[lo:hi] { |
| 916 | if v == nil { |
| 917 | continue |
| 918 | } |
| 919 | sum += *v |
| 920 | cnt++ |
| 921 | } |
| 922 | if cnt == 0 { |
| 923 | continue // nothing readable in this bucket: a gap, not a zero |
| 924 | } |
| 925 | mean := sum / float64(cnt) |
| 926 | out[b] = &mean |
| 927 | } |
| 928 | return out |
| 929 | } |
| 930 | |
amadan