package api
import (
"context"
"fmt"
"net/url"
"os"
"sort"
"sync"
"time"
"craftdesign.group/quarry/internal/health"
"craftdesign.group/quarry/internal/obs"
"craftdesign.group/quarry/internal/timing"
)
// The operations page's headline is a list of findings, and so is the
// answer an agent reads from ops.status and the alert notify sends:
// one rule set, computed here, so none of the three can disagree. A
// finding has a stable key, so this process can say when it first saw
// it and notice when it closes.
// opsFinding is one unusual thing: what it is about, how bad, the
// sentence that says so, and where to go.
type opsFinding struct {
// Key names the condition across snapshots: "memory",
// "lane:archives", "slow:GET /r/{repo}/timeline", "store:tcl".
Key string `json:"key"`
About string `json:"about"` // memory | lane | slow | errors | store | history | dependency | disk
Severity string `json:"severity"` // "bad" or "warn"
Text string `json:"text"`
// Since is when this process first saw the condition, RFC3339.
// Empty only on a finding built outside ops.status.
Since string `json:"since,omitempty"`
// Link is the page on this forge that opens the condition up.
Link string `json:"link,omitempty"`
}
// opsClosed is a finding that stopped being one, with when it ended.
type opsClosed struct {
Key string `json:"key"`
About string `json:"about"`
Text string `json:"text"`
Since string `json:"since"`
Until string `json:"until"`
Link string `json:"link,omitempty"`
}
// opsSignals is obs.Signals with the fourth signal joined in from the
// error journal, which keeps its own per-occurrence record.
type opsSignals struct {
Minutes int `json:"minutes"`
Renders []int `json:"renders"`
Slow []int `json:"slow"`
Busy []int `json:"busy"`
Errors []int `json:"errors,omitempty"`
}
// opsDep is health.State as the ops page reads it.
type opsDep struct {
Name string `json:"name"`
Why string `json:"why,omitempty"`
Up bool `json:"up"`
// Since is when it entered this state; DownFor words the outage.
Since string `json:"since,omitempty"`
DownFor string `json:"downFor,omitempty"`
Err string `json:"err,omitempty"`
Checked string `json:"checked,omitempty"`
}
// opsDisk is the filesystem under the derived data: its size, what is
// free, and how much of the used part the forge can account for.
type opsDisk struct {
Path string `json:"path"`
Total uint64 `json:"total"`
Free uint64 `json:"free"`
Used uint64 `json:"used"`
// Stores, Indexes and Analytics are the forge's own files; Other
// is the rest of what is used, which is everything else on the
// filesystem.
Stores uint64 `json:"stores"`
Indexes uint64 `json:"indexes"`
Analytics uint64 `json:"analytics"`
Other uint64 `json:"other"`
}
// opsStore is one served store and the state of its index.
type opsStore struct {
Name string `json:"name"`
// State is current | behind | rebuilding | starting | failing |
// stalled | gaps, worst first in the list.
State string `json:"state"`
// Why is the fact behind a state that is not "current".
Why string `json:"why,omitempty"`
StoreBytes uint64 `json:"storeBytes"`
IndexBytes uint64 `json:"indexBytes"`
// LastSync is when a round last succeeded, RFC3339; empty if never.
LastSync string `json:"lastSync,omitempty"`
RoundSeconds float64 `json:"roundSeconds"`
Checkins int `json:"checkins"`
}
// behindAfter is how stale a store's last successful round may be
// before its index counts as behind. Rounds run every few seconds and
// complete even when nothing arrived, so a last success this old means
// the poller is not completing them.
const behindAfter = 2 * time.Minute
// diskWorry and diskBad are the used fractions a disk finding starts at.
const (
diskWorry = 0.8
diskBad = 0.9
)
// processStart stands in for Deps.Started when the root did not set it.
var processStart = time.Now()
// findingClock remembers when each finding was first seen by this
// process and the last one that closed. A ring of 400 renders cannot
// say when a route went slow; this can, for as long as the process
// lives, which is also how long the alert dedup lives.
type findingClock struct {
mu sync.Mutex
since map[string]time.Time
seen map[string]opsFinding
closed *opsClosed
}
var findings findingClock
// stamp fills Since on each finding, records the ones that have gone,
// and answers the last closed one.
func (c *findingClock) stamp(now time.Time, fs []opsFinding) *opsClosed {
c.mu.Lock()
defer c.mu.Unlock()
if c.since == nil {
c.since, c.seen = map[string]time.Time{}, map[string]opsFinding{}
}
present := map[string]bool{}
for i := range fs {
k := fs[i].Key
present[k] = true
t, ok := c.since[k]
if !ok {
t = now
c.since[k] = t
}
c.seen[k] = fs[i]
fs[i].Since = t.UTC().Format(time.RFC3339)
}
for k, t := range c.since {
if present[k] {
continue
}
was := c.seen[k]
c.closed = &opsClosed{Key: k, About: was.About, Text: was.Text, Link: was.Link,
Since: t.UTC().Format(time.RFC3339), Until: now.UTC().Format(time.RFC3339)}
delete(c.since, k)
delete(c.seen, k)
}
if c.closed == nil {
return nil
}
cp := *c.closed
return &cp
}
// resetFindings forgets everything, for tests.
func resetFindings() {
findings.mu.Lock()
defer findings.mu.Unlock()
findings.since, findings.seen, findings.closed = nil, nil, nil
}
// opsFindings is every rule in one place. Order: bad before warn, then
// by key, so the list is stable between ticks.
func (d Deps) opsFindings(ctx context.Context, out *opsStatusOut, snap obs.Snapshot) []opsFinding {
var fs []opsFinding
for _, f := range memoryFindings(out.Memory) {
f.Key = "memory"
fs = append(fs, f)
}
// A lane that is full and has turned callers away: the reader's
// busy page, with its cause. Closed when a slot frees.
for _, l := range out.Lanes {
if l.Refused == 0 || l.Bound == 0 || l.Running < l.Bound {
continue
}
text := fmt.Sprintf("%s turned %s away · %d of %d running", l.Name, counted(l.Refused, "caller"), l.Running, l.Bound)
if l.Waiting > 0 {
text += fmt.Sprintf(", %d waiting", l.Waiting)
}
if l.WaitedMs > 0 {
text += " up to " + timing.Format(time.Duration(l.WaitedMs)*time.Millisecond)
}
fs = append(fs, opsFinding{Key: "lane:" + l.Name, About: "lane", Severity: "bad", Text: text, Link: "/fleet/ops#lanes"})
}
// Routes over budget, each with the worst page it served, which is
// what anybody can open. The same threshold notifies.
worst := map[string]obs.Render{}
for _, r := range snap.Renders {
if r.Elapsed > worst[r.Path].Elapsed {
worst[r.Path] = r
}
}
for _, p := range snap.Paths {
if p.P95 <= obs.Budget {
continue
}
text := fmt.Sprintf("%s p95 %s over the %s budget", p.Path, timing.Format(p.P95), timing.Format(obs.Budget))
if w := worst[p.Path]; w.Elapsed > 0 && d.siteReaches(ctx, w.Repo) {
if u := d.renderURL(ctx, w.Repo, w.URL, w.Path); u != "" && u != p.Path {
text += fmt.Sprintf(" · worst %s %s", u, timing.Format(w.Elapsed))
}
}
fs = append(fs, opsFinding{Key: "slow:" + p.Path, About: "slow", Severity: "warn", Text: text,
Link: "/fleet/ops/route?path=" + url.QueryEscape(p.Path)})
}
// Server errors in the last hour, from the journal.
if n := sum(out.Signals.Errors); n > 0 {
text := counted(n, "server error") + " in the last hour"
if d.Errors != nil {
if _, _, last := d.Errors.Stats(); !last.IsZero() {
text += " · newest " + last.UTC().Format(time.RFC3339)
}
}
fs = append(fs, opsFinding{Key: "errors", About: "errors", Severity: "bad", Text: text, Link: "/fleet/exceptions"})
}
// Stores whose index is not keeping up with them.
for _, s := range out.Stores {
sev := ""
switch s.State {
case "stalled", "failing":
sev = "bad"
case "behind", "gaps":
sev = "warn"
}
if sev == "" {
continue
}
fs = append(fs, opsFinding{Key: "store:" + s.Name, About: "store", Severity: sev,
Text: s.Name + " index " + s.State + ": " + s.Why, Link: "/fleet#" + url.PathEscape(s.Name)})
}
// The history store, when it is configured and not working. Off is
// a choice and not a finding.
if h := out.Store; h.Configured {
if h.Err != "" {
fs = append(fs, opsFinding{Key: "history", About: "history", Severity: "bad",
Text: "history is not being recorded: " + h.Err, Link: "/fleet/ops/history"})
} else if h.Dropped > 0 {
fs = append(fs, opsFinding{Key: "history", About: "history", Severity: "warn",
Text: fmt.Sprintf("%d observations were not recorded: the store fell behind or a write failed", h.Dropped),
Link: "/fleet/ops/history"})
}
}
for _, dep := range out.DependsOn {
if dep.Up {
continue
}
text := dep.Name + " is unreachable"
if dep.DownFor != "" {
text += " for " + dep.DownFor
}
if dep.Why != "" {
text += ": " + dep.Why
}
fs = append(fs, opsFinding{Key: "dep:" + dep.Name, About: "dependency", Severity: "bad", Text: text, Link: "/fleet"})
}
if dk := out.Disk; dk != nil && dk.Total > 0 {
used := float64(dk.Used) / float64(dk.Total)
if used >= diskWorry {
sev := "warn"
if used >= diskBad {
sev = "bad"
}
fs = append(fs, opsFinding{Key: "disk", About: "disk", Severity: sev,
Text: fmt.Sprintf("disk under %s is %.0f%% full · %s free of %s", dk.Path, used*100, gb(dk.Free), gb(dk.Total)),
Link: "/fleet/ops#disk"})
}
}
sort.SliceStable(fs, func(i, j int) bool {
if fs[i].Severity != fs[j].Severity {
return fs[i].Severity == "bad"
}
return fs[i].Key < fs[j].Key
})
return fs
}
// opsDeps is the health watcher's beliefs, worded.
func opsDeps(states []health.State) []opsDep {
out := make([]opsDep, 0, len(states))
for _, s := range states {
dep := opsDep{Name: s.Name, Why: s.Why, Up: s.Up, Err: s.Err, DownFor: s.DownFor()}
if !s.Since.IsZero() {
dep.Since = s.Since.UTC().Format(time.RFC3339)
}
if !s.Checked.IsZero() {
dep.Checked = s.Checked.UTC().Format(time.RFC3339)
}
out = append(out, dep)
}
return out
}
// opsStores is every served store with its index's state, the ones
// needing attention first.
func (d Deps) opsStores(now time.Time) []opsStore {
if d.Repos == nil {
return nil
}
var out []opsStore
for _, name := range d.Repos.Names() {
repo, ok := d.Repos.Get(name)
if !ok || repo == nil {
continue
}
s := opsStore{Name: name, State: "current"}
if repo.Zit != nil {
s.StoreBytes = fileBytes(repo.Zit.Path())
}
if repo.Index != nil {
s.IndexBytes = fileBytes(repo.Index.Path())
s.Checkins, _ = repo.Index.CheckinCount()
}
var h RepoHealth
if repo.Health != nil {
h = repo.Health()
}
s.LastSync, s.RoundSeconds = h.LastOK, h.RoundSeconds
last, _ := time.Parse(time.RFC3339, h.LastOK)
switch {
case repo.Indexing != nil && repo.Indexing():
s.State, s.Why = "rebuilding", "its first fold is still running"
case h.Stalled:
s.State, s.Why = "stalled", "ingest stopped retrying after "+counted(h.Failures, "failure")
if h.LastErr != "" {
s.Why += ": " + h.LastErr
}
case h.Failures >= 3:
s.State, s.Why = "failing", counted(h.Failures, "round")+" in a row failed"
if h.LastErr != "" {
s.Why += ": " + h.LastErr
}
case h.LastOK == "":
s.State, s.Why = "starting", "no round has completed yet"
case !last.IsZero() && now.Sub(last) > behindAfter:
s.State, s.Why = "behind", "last successful round "+agoWords(now.Sub(last))+" ago"
case h.Gaps > 0:
s.State, s.Why = "gaps", counted(h.Gaps, "artifact")+" missing from the index"
}
out = append(out, s)
}
rank := map[string]int{"stalled": 0, "failing": 1, "behind": 2, "gaps": 3, "rebuilding": 4, "starting": 5, "current": 6}
sort.SliceStable(out, func(i, j int) bool {
if rank[out[i].State] != rank[out[j].State] {
return rank[out[i].State] < rank[out[j].State]
}
return out[i].Name < out[j].Name
})
return out
}
// opsDisk measures the filesystem under the data directory and what
// the forge holds on it. Sizes are stat calls, a few dozen of them,
// cached briefly so the live stream does not repeat them every tick.
func (d Deps) opsDisk(stores []opsStore, analytics string) *opsDisk {
if d.DataDir == "" {
return nil
}
total, free, ok := statfs(d.DataDir)
if !ok {
return nil
}
dk := &opsDisk{Path: d.DataDir, Total: total, Free: free}
if total > free {
dk.Used = total - free
}
for _, s := range stores {
dk.Stores += s.StoreBytes
dk.Indexes += s.IndexBytes
}
if analytics != "" {
dk.Analytics = fileBytes(analytics)
}
if own := dk.Stores + dk.Indexes + dk.Analytics; dk.Used > own {
dk.Other = dk.Used - own
}
return dk
}
// sizeCache keeps each file's measured size for a short while: the ops
// stream re-reads the snapshot every few seconds and a store's size
// does not move that fast.
var sizeCache = struct {
sync.Mutex
m map[string]sizeAt
}{m: map[string]sizeAt{}}
type sizeAt struct {
at time.Time
bytes uint64
}
const sizeCacheFor = 30 * time.Second
// fileBytes is a SQLite database's footprint: the file and the
// journal, WAL and shared-memory files beside it when present.
func fileBytes(path string) uint64 {
if path == "" {
return 0
}
now := time.Now()
sizeCache.Lock()
if c, ok := sizeCache.m[path]; ok && now.Sub(c.at) < sizeCacheFor {
sizeCache.Unlock()
return c.bytes
}
sizeCache.Unlock()
var n uint64
for _, suffix := range []string{"", "-wal", "-shm", "-journal"} {
if st, err := os.Stat(path + suffix); err == nil && !st.IsDir() {
n += uint64(st.Size())
}
}
sizeCache.Lock()
sizeCache.m[path] = sizeAt{at: now, bytes: n}
sizeCache.Unlock()
return n
}
// agoWords is a staleness in a maintainer's units: minutes under an
// hour, hours and minutes under two days, days after.
func agoWords(d time.Duration) string {
switch {
case d < time.Minute:
return fmt.Sprintf("%ds", int(d.Seconds()))
case d < time.Hour:
return fmt.Sprintf("%dm", int(d.Minutes()))
case d < 48*time.Hour:
return fmt.Sprintf("%dh%02dm", int(d.Hours()), int(d.Minutes())%60)
}
return fmt.Sprintf("%dd", int(d.Hours()/24))
}
func sum(xs []int) int {
n := 0
for _, x := range xs {
n += x
}
return n
}
func counted(n int, noun string) string {
if n == 1 {
return "1 " + noun
}
return fmt.Sprintf("%d %ss", n, noun)
}
// errorsByMinute is the hour of server errors as one series, summed
// over signatures; nil when the forge has no durable journal.
func (d Deps) errorsByMinute() []int {
if d.ErrorRates == nil {
return nil
}
rates, err := d.ErrorRates(obs.SignalMinutes, obs.SignalMinutes*time.Minute)
if err != nil {
return nil
}
out := make([]int, obs.SignalMinutes)
for _, r := range rates {
for i, n := range r {
if i < len(out) {
out[i] += n
}
}
}
return out
}
// routesOver counts the routes whose p95 is over budget, of all seen.
func routesOver(paths []obs.PathStat) (over, all int) {
for _, p := range paths {
all++
if p.P95 > obs.Budget {
over++
}
}
return over, all
}