| new file mode 100644 |
| index 000000000..92a4bafea |
| --- /dev/null |
| +++ b/internal/console/metricsview.go |
| @@ -0,0 +1,319 @@ |
| +// Console-side read path for host/process monitoring: assembling what the |
| +// operator dashboard (Task 11) and owner app-metrics page (Task 12) render |
| +// out of what the box agent publishes (internal/metrics). Nothing here |
| +// writes anything — it is a bucket sweep plus an in-process cache, and the |
| +// wire shapes it reads (metrics.RawChunk, metrics.AlertState) are read |
| +// defensively: an unparsable object is skipped, never fatal to the sweep, |
| +// mirroring alertwatch.go's "never go silent" posture on the same bucket |
| +// prefixes. |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "sort" |
| + "strconv" |
| + "strings" |
| + "sync" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/metrics" |
| +) |
| + |
| +// UnitRow is one systemd unit's latest reading, ready to render. |
| +type UnitRow struct { |
| + Unit string // full unit name |
| + CPUFrac *float64 |
| + MemBytes *int64 |
| + ReadBytes, WriteBytes *int64 |
| +} |
| + |
| +// BoxMetricsView is one box's assembled dashboard row: latest host readings |
| +// plus a 48h sparkline per metric plus its units, sorted. |
| +type BoxMetricsView struct { |
| + Box string |
| + At time.Time // latest sample time |
| + CPU, Mem, Disk *float64 |
| + SparkCPU, SparkMem, SparkDisk string // SVG polyline points, "" when no data |
| + Units []UnitRow // sorted by CPUFrac desc, nil-CPU rows last |
| +} |
| + |
| +// AlertRow is one currently-firing rule on one box. LoadMetrics only ever |
| +// returns firing rules — a quiet rule has nothing to show. |
| +type AlertRow struct { |
| + Box, Rule string |
| + Value float64 |
| + FiredAt time.Time |
| +} |
| + |
| +// metricsCacheTTL bounds how long LoadMetrics reuses its last successful |
| +// sweep before touching the bucket again. All callers within the window |
| +// share the one sweep — the operator dashboard and every owner's app-metrics |
| +// page render off the same read. |
| +const metricsCacheTTL = 60 * time.Second |
| + |
| +// metricsSweepHours is how far back LoadMetrics reads raw chunks: the last |
| +// 48 UTC hours, days/ rollups untouched (a later task's job). |
| +const metricsSweepHours = 48 |
| + |
| +// metricsBoxesPrefix is where a box publishes its raw chunks and day |
| +// rollups (internal/metrics.RawKey / DayKey); box discovery lists this |
| +// prefix and reads the path segment right after "boxes/". |
| +const metricsBoxesPrefix = "metrics/boxes/" |
| + |
| +// metricsCache holds the result of the last successful bucket sweep. A |
| +// failed sweep never reaches here — LoadMetrics returns the error to its |
| +// caller and leaves the previous (possibly stale-but-valid, possibly zero) |
| +// cache exactly as it was. |
| +type metricsCache struct { |
| + mu sync.Mutex |
| + at time.Time |
| + boxes []BoxMetricsView |
| + alerts []AlertRow |
| +} |
| + |
| +// LoadMetrics reads every box under metrics/boxes/, the raw chunks of the |
| +// last 48 hours (missing hours skipped silently), and alerts/boxes/*.json, |
| +// caching the result in-process for 60s so concurrent callers — the |
| +// dashboard page and every owner's app-metrics page — share one sweep. |
| +func (s *Server) LoadMetrics(ctx context.Context) ([]BoxMetricsView, []AlertRow, error) { |
| + s.metricsCache.mu.Lock() |
| + defer s.metricsCache.mu.Unlock() |
| + |
| + if !s.metricsCache.at.IsZero() && time.Since(s.metricsCache.at) < metricsCacheTTL { |
| + return s.metricsCache.boxes, s.metricsCache.alerts, nil |
| + } |
| + |
| + boxNames, err := s.discoverBoxes(ctx) |
| + if err != nil { |
| + return nil, nil, err |
| + } |
| + |
| + views := make([]BoxMetricsView, 0, len(boxNames)) |
| + for _, box := range boxNames { |
| + view, err := s.sweepBox(ctx, box) |
| + if err != nil { |
| + return nil, nil, err |
| + } |
| + views = append(views, view) |
| + } |
| + |
| + alerts, err := s.sweepAlerts(ctx) |
| + if err != nil { |
| + return nil, nil, err |
| + } |
| + |
| + // Only a fully successful sweep is cached — a partial or errored one |
| + // falls through to the returns above without ever reaching this line. |
| + s.metricsCache.at = time.Now() |
| + s.metricsCache.boxes = views |
| + s.metricsCache.alerts = alerts |
| + return views, alerts, nil |
| +} |
| + |
| +// discoverBoxes lists metrics/boxes/ and returns the deduped, sorted set of |
| +// box names — the path segment right after "boxes/". |
| +func (s *Server) discoverBoxes(ctx context.Context) ([]string, error) { |
| + keys, err := s.cfg.Releases.List(ctx, metricsBoxesPrefix) |
| + if err != nil { |
| + return nil, fmt.Errorf("list %s: %w", metricsBoxesPrefix, err) |
| + } |
| + seen := make(map[string]bool) |
| + var boxes []string |
| + for _, key := range keys { |
| + rest := strings.TrimPrefix(key, metricsBoxesPrefix) |
| + idx := strings.Index(rest, "/") |
| + if idx <= 0 { |
| + continue // malformed/unexpected key shape directly under the prefix |
| + } |
| + box := rest[:idx] |
| + if seen[box] { |
| + continue |
| + } |
| + seen[box] = true |
| + boxes = append(boxes, box) |
| + } |
| + sort.Strings(boxes) |
| + return boxes, nil |
| +} |
| + |
| +// sweepBox reads box's raw chunks for the last 48 UTC hours and assembles |
| +// its BoxMetricsView. A missing hour (os.ErrNotExist) or an unparsable chunk |
| +// is skipped silently — a box that has been up for under 48h, or whose |
| +// agent hiccuped on one hour, still renders on what it does have. Any other |
| +// read error fails the whole sweep (and so is never cached). |
| +func (s *Server) sweepBox(ctx context.Context, box string) (BoxMetricsView, error) { |
| + now := time.Now().UTC() |
| + var samples []metrics.Sample |
| + for i := 0; i < metricsSweepHours; i++ { |
| + hour := metrics.HourOf(now.Add(-time.Duration(i) * time.Hour)) |
| + key := metrics.RawKey(box, hour) |
| + body, _, err := s.cfg.Releases.Get(ctx, key) |
| + if err != nil { |
| + if errors.Is(err, os.ErrNotExist) { |
| + continue |
| + } |
| + return BoxMetricsView{}, fmt.Errorf("read %s: %w", key, err) |
| + } |
| + var chunk metrics.RawChunk |
| + if err := json.Unmarshal(body, &chunk); err != nil { |
| + continue // unparsable chunk: skip, don't fail the sweep |
| + } |
| + samples = append(samples, chunk.Samples...) |
| + } |
| + sort.Slice(samples, func(i, j int) bool { return samples[i].At.Before(samples[j].At) }) |
| + return buildBoxView(box, samples), nil |
| +} |
| + |
| +// buildBoxView assembles one box's view from its swept samples, oldest |
| +// first. The latest sample's own values are used as-is for CPU/Mem/Disk/At |
| +// — nil stays nil; this is deliberately NOT "each field's newest non-nil |
| +// reading independently", which would quietly mix readings from different |
| +// points in time. |
| +func buildBoxView(box string, samples []metrics.Sample) BoxMetricsView { |
| + view := BoxMetricsView{Box: box} |
| + if len(samples) == 0 { |
| + return view |
| + } |
| + latest := samples[len(samples)-1] |
| + view.At = latest.At |
| + view.CPU = latest.Host.CPUFrac |
| + view.Mem = latest.Host.MemFrac |
| + view.Disk = latest.Host.DiskFrac |
| + |
| + cpu := make([]*float64, len(samples)) |
| + mem := make([]*float64, len(samples)) |
| + disk := make([]*float64, len(samples)) |
| + for i, sm := range samples { |
| + cpu[i] = sm.Host.CPUFrac |
| + mem[i] = sm.Host.MemFrac |
| + disk[i] = sm.Host.DiskFrac |
| + } |
| + view.SparkCPU = sparkPoints(cpu, 240, 40) |
| + view.SparkMem = sparkPoints(mem, 240, 40) |
| + view.SparkDisk = sparkPoints(disk, 240, 40) |
| + |
| + view.Units = buildUnitRows(latest.Units) |
| + return view |
| +} |
| + |
| +// buildUnitRows turns the latest sample's Units map into rows sorted by |
| +// CPUFrac descending, nil-CPU rows last, stable by unit name within ties. A |
| +// unit absent from the latest sample's map simply isn't in this slice — |
| +// never rendered, never aggregated as a zero (Task 4's "missing map key is |
| +// no reading" ruling, same map shape here). |
| +func buildUnitRows(units map[string]metrics.UnitSample) []UnitRow { |
| + rows := make([]UnitRow, 0, len(units)) |
| + for name, u := range units { |
| + rows = append(rows, UnitRow{ |
| + Unit: name, |
| + CPUFrac: u.CPUFrac, |
| + MemBytes: u.MemBytes, |
| + ReadBytes: u.ReadBytes, |
| + WriteBytes: u.WriteBytes, |
| + }) |
| + } |
| + // Name order first so the CPU-desc sort below, being stable, ties out |
| + // alphabetically rather than on map iteration order. |
| + sort.Slice(rows, func(i, j int) bool { return rows[i].Unit < rows[j].Unit }) |
| + sort.SliceStable(rows, func(i, j int) bool { |
| + a, b := rows[i].CPUFrac, rows[j].CPUFrac |
| + if a == nil { |
| + return false // nil never sorts before anything, including another nil |
| + } |
| + if b == nil { |
| + return true |
| + } |
| + return *a > *b |
| + }) |
| + return rows |
| +} |
| + |
| +// sweepAlerts reads every alerts/boxes/*.json and returns one AlertRow per |
| +// currently-firing rule. Unreadable-but-present keys fail the sweep (so it |
| +// is never cached half-done); an unparsable object is skipped, matching |
| +// alertPass's posture on the same prefix. |
| +func (s *Server) sweepAlerts(ctx context.Context) ([]AlertRow, error) { |
| + keys, err := s.cfg.Releases.List(ctx, alertPrefix) |
| + if err != nil { |
| + return nil, fmt.Errorf("list %s: %w", alertPrefix, err) |
| + } |
| + var rows []AlertRow |
| + for _, key := range keys { |
| + if !strings.HasSuffix(key, ".json") { |
| + continue |
| + } |
| + body, _, err := s.cfg.Releases.Get(ctx, key) |
| + if err != nil { |
| + if errors.Is(err, os.ErrNotExist) { |
| + continue |
| + } |
| + return nil, fmt.Errorf("read %s: %w", key, err) |
| + } |
| + var st metrics.AlertState |
| + if err := json.Unmarshal(body, &st); err != nil { |
| + continue // unparsable: skip, don't fail the sweep |
| + } |
| + box := st.Box |
| + if box == "" { |
| + // Trust the key over an empty body field, same as alertPass. |
| + box = strings.TrimSuffix(strings.TrimPrefix(key, alertPrefix), ".json") |
| + } |
| + names := make([]string, 0, len(st.Rules)) |
| + for name := range st.Rules { |
| + names = append(names, name) |
| + } |
| + sort.Strings(names) |
| + for _, name := range names { |
| + rs := st.Rules[name] |
| + if !rs.Firing { |
| + continue |
| + } |
| + rows = append(rows, AlertRow{Box: box, Rule: name, Value: rs.Value, FiredAt: rs.FiredAt}) |
| + } |
| + } |
| + return rows, nil |
| +} |
| + |
| +// sparkPoints maps vals (up to 48h of minute samples, one slot per sample |
| +// index) onto an SVG polyline "points" string on a w x h viewBox. Values are |
| +// fractions and clamp into [0,1] if out of range; y is inverted (SVG's y |
| +// axis points down, so a higher fraction plots at a SMALLER y — visually |
| +// "up"); x is spread evenly across the width by the sample's index in vals, |
| +// not by elapsed time. A nil sample contributes no pair at all — its x slot |
| +// is simply skipped, so the polyline draws a straight line across the gap; |
| +// breaking the line at gaps is not attempted in v1 (documented here per the |
| +// brief). Empty input or input that is nil throughout returns "". |
| +func sparkPoints(vals []*float64, w, h int) string { |
| + if len(vals) == 0 { |
| + return "" |
| + } |
| + var b strings.Builder |
| + n := len(vals) |
| + for i, v := range vals { |
| + if v == nil { |
| + continue |
| + } |
| + f := *v |
| + if f < 0 { |
| + f = 0 |
| + } else if f > 1 { |
| + f = 1 |
| + } |
| + x := 0.0 |
| + if n > 1 { |
| + x = float64(i) / float64(n-1) * float64(w) |
| + } |
| + y := float64(h) - f*float64(h) |
| + if b.Len() > 0 { |
| + b.WriteByte(' ') |
| + } |
| + b.WriteString(strconv.FormatFloat(x, 'f', 2, 64)) |
| + b.WriteByte(',') |
| + b.WriteString(strconv.FormatFloat(y, 'f', 2, 64)) |
| + } |
| + return b.String() |
| +} |
amadan