package main import ( "context" "fmt" "os" "os/user" "strconv" "strings" "sync" "time" ) // The module's long-running code (novox/hq ADR 0198): the guard that warns before the machine kills // something for memory. It replaces the predecessor's `mem-guard`, a user unit that checked used RAM // every thirty seconds and called notify-send. // // What changed, and why: // // - **Three signals, not one.** Used RAM alone misses the two things that decide whether a kill is // coming: swap filling (systemd-oomd kills at SwapUsedLimit) and stall pressure (it kills a unit // whose pressure stays over its limit for twenty seconds). The guard warns on whichever comes // first, and says which. // - **Ten seconds, not thirty.** oomd's own window is twenty seconds; a check every thirty can // notice only after oomd has acted. Reading two files every ten seconds costs nothing measurable. // The kernel's PSI triggers would wake on pressure alone, but not on RAM or swap filling, so the // guard reads all three on one timer rather than run two mechanisms. // - **The notification names what to close.** It lists the largest units, not only a percentage. // - **An event as well as a notification.** `pressure.high` and `pressure.cleared` reach the mesh // whether or not anyone is at the desktop — a server has no desktop at all. // - **No session needed.** The runtime runs as the operator's account but outside the graphical // session, so it has no session bus address. The guard names the account's own bus, // /run/user//bus, and speaks to the notification service with busctl, the service manager's // client: nothing is installed for it, and a machine without a desktop simply has no such bus. // Thresholds, as constants until settings exist (novox/hq issue 168). Used RAM is the predecessor's, // raised by it from 90 to 95 after the lower line fired during ordinary work. const ( WarnUsedPercent = 95.0 ClearUsedPercent = 88.0 WarnSwapPercent = 80.0 // systemd-oomd's SwapUsedLimit is 90 % ClearSwapPercent = 70.0 WarnPressureAvg10 = 40.0 // "some" avg10; oomd acts at 60 % (80 % for the user manager) over 20 s ClearPressureAvg10 = 10.0 CheckEvery = 10 * time.Second NotificationID = 9010 // replaces the previous warning rather than stacking ) // Verdict is what one check of the machine concluded. type Verdict struct { High bool `json:"high"` Reasons []string `json:"reasons"` Clear bool `json:"clear"` } // Judge applies the thresholds to a snapshot. High is any warning line crossed; Clear is every value // below its clearing line, which is what re-arms the guard. func Judge(s Snapshot) Verdict { v := Verdict{Reasons: []string{}} clear := true if s.UsedPercent >= WarnUsedPercent { v.Reasons = append(v.Reasons, fmt.Sprintf("RAM %.0f%% used (warns at %.0f%%)", s.UsedPercent, WarnUsedPercent)) } if s.UsedPercent > ClearUsedPercent { clear = false } if s.SwapTotalGiB > 0 { if s.SwapPercent >= WarnSwapPercent { v.Reasons = append(v.Reasons, fmt.Sprintf("swap %.0f%% used (warns at %.0f%%; systemd-oomd kills at 90%%)", s.SwapPercent, WarnSwapPercent)) } if s.SwapPercent > ClearSwapPercent { clear = false } } if s.PressureSome != nil { if s.PressureSome.Avg10 >= WarnPressureAvg10 { v.Reasons = append(v.Reasons, fmt.Sprintf("tasks stalled on memory %.0f%% of the last 10 s (warns at %.0f%%)", s.PressureSome.Avg10, WarnPressureAvg10)) } if s.PressureSome.Avg10 > ClearPressureAvg10 { clear = false } } v.High = len(v.Reasons) > 0 v.Clear = clear return v } // Guard is the notifier's state, shared with the tools that report it. type Guard struct { m *Machine now func() time.Time emit func(string, any) error notify func(ctx context.Context, summary, body string) error mu sync.Mutex latched bool since time.Time last *Verdict lastAt time.Time notified string err string desktop string stopped string } func NewGuard(m *Machine, emit func(string, any) error) *Guard { g := &Guard{m: m, now: time.Now, emit: emit} g.notify = g.desktopNotify return g } // Check is one wake-up: read, judge, and on crossing a line warn once; on clearing every line, re-arm. func (g *Guard) Check(ctx context.Context) { s, err := g.m.Snapshot() g.mu.Lock() if err != nil { g.err = err.Error() g.mu.Unlock() return } v := Judge(s) now := g.now() g.last, g.lastAt, g.err = &v, now, "" warn := v.High && !g.latched cleared := g.latched && v.Clear if warn { g.latched, g.since = true, now } if cleared { g.latched = false } since := g.since g.mu.Unlock() if warn { largest := ByUnit(g.m.Processes(), 3) var names []string for _, u := range largest { names = append(names, fmt.Sprintf("%s %.1f GiB", firstNonEmpty(u.Largest, u.Unit), (u.RSSMiB+u.SwapMiB)/1024)) } body := strings.Join(v.Reasons, "; ") if len(names) > 0 { body += ". Largest: " + strings.Join(names, ", ") } body += ". systemd-oomd kills the worst unit if it climbs further." g.publish("pressure.high", map[string]any{"reasons": v.Reasons, "used_percent": s.UsedPercent, "swap_used_percent": s.SwapPercent, "pressure_some": s.PressureSome, "largest": largest}) err := g.notify(ctx, "Memory is running low", body) g.mu.Lock() if err != nil { g.desktop = "not reached: " + err.Error() } else { g.desktop, g.notified = "reached", now.Format(time.RFC3339) } g.mu.Unlock() } if cleared { g.publish("pressure.cleared", map[string]any{"used_percent": s.UsedPercent, "swap_used_percent": s.SwapPercent, "lasted_seconds": int(now.Sub(since).Seconds())}) } } func (g *Guard) publish(eventType string, body map[string]any) { if g.emit == nil { return } if err := g.emit(eventType, body); err != nil { fmt.Fprintf(os.Stderr, "%s not published: %v\n", eventType, err) } } // Bus is the operator's session bus socket: the account the runtime names, else the one this process // runs as. func Bus() (string, error) { uid := os.Getuid() if name := os.Getenv("MESH_OPERATOR_ACCOUNT"); name != "" { u, err := user.Lookup(name) if err != nil { return "", fmt.Errorf("the operator account %s: %w", name, err) } if n, err := strconv.Atoi(u.Uid); err == nil { if n != uid && uid != 0 { return "", fmt.Errorf("this process runs as uid %d and the operator's bus belongs to uid %d", uid, n) } uid = n } } return fmt.Sprintf("/run/user/%d/bus", uid), nil } // desktopNotify sends one critical notification to the operator's session, replacing the previous one. func (g *Guard) desktopNotify(ctx context.Context, summary, body string) error { bus, err := Bus() if err != nil { return err } if _, err := os.Stat(bus); err != nil { return fmt.Errorf("no session bus at %s (nobody is logged in to a desktop)", bus) } _, err = g.m.Run(ctx, "env", "DBUS_SESSION_BUS_ADDRESS=unix:path="+bus, "busctl", "--user", "call", "org.freedesktop.Notifications", "/org/freedesktop/Notifications", "org.freedesktop.Notifications", "Notify", "susssasa{sv}i", "memory-pressure", strconv.Itoa(NotificationID), "dialog-warning", summary, body, "0", "1", "urgency", "y", "2", "0") return err } // Run is the guard's life: a check every CheckEvery. func (g *Guard) Run(ctx context.Context) { defer func() { if r := recover(); r != nil { g.mu.Lock() g.stopped = fmt.Sprintf("the guard stopped on a fault: %v", r) g.mu.Unlock() fmt.Fprintln(os.Stderr, g.stopped) } }() t := time.NewTicker(CheckEvery) defer t.Stop() for { g.Check(ctx) select { case <-ctx.Done(): return case <-t.C: } } } // GuardReport is the guard as the guard tool shows it. type GuardReport struct { Running bool `json:"running"` Stopped string `json:"stopped,omitempty"` Warned bool `json:"warning_given"` WarnedSince *time.Time `json:"warning_since,omitempty"` LastCheck *time.Time `json:"last_check,omitempty"` LastVerdict *Verdict `json:"last_verdict,omitempty"` LastError string `json:"last_error,omitempty"` Desktop string `json:"desktop"` LastNotified string `json:"last_notified,omitempty"` Thresholds map[string]any `json:"thresholds"` } func (g *Guard) Report() GuardReport { g.mu.Lock() defer g.mu.Unlock() r := GuardReport{ Running: g.stopped == "" && !g.lastAt.IsZero(), Stopped: g.stopped, Warned: g.latched, LastCheck: when(g.lastAt), LastVerdict: g.last, LastError: g.err, Desktop: g.desktop, LastNotified: g.notified, Thresholds: map[string]any{ "warn_used_percent": WarnUsedPercent, "clear_used_percent": ClearUsedPercent, "warn_swap_percent": WarnSwapPercent, "clear_swap_percent": ClearSwapPercent, "warn_pressure_some_avg10": WarnPressureAvg10, "clear_pressure_some_avg10": ClearPressureAvg10, "check_every": CheckEvery.String(), "set_by": "constants until settings exist (novox/hq issue 168)", }, } if g.latched { r.WarnedSince = when(g.since) } if r.Desktop == "" { if bus, err := Bus(); err != nil { r.Desktop = "not reachable: " + err.Error() } else if _, err := os.Stat(bus); err != nil { r.Desktop = "no session bus at " + bus } else { r.Desktop = "reachable at " + bus + " (not yet used)" } } return r } func firstNonEmpty(ss ...string) string { for _, s := range ss { if s != "" { return s } } return "" } // when is a time for a report: absent rather than the zero time. func when(t time.Time) *time.Time { if t.IsZero() { return nil } return &t }