package obs
import (
"log"
"os"
"strconv"
"strings"
"sync"
"time"
)
// What the process really holds, beyond the Go heap. SQLite connection
// memory is not in the heap, so the heap sample and GOMEMLIMIT cannot
// see it; resident anonymous memory can. The 2026-11-14 kill happened
// at 10.9 GB resident with a healthy-looking heap.
//
// Every reader here answers 0 when its source is absent (macOS, a test,
// a cgroup v1 host), and every warning rule treats 0 as "unknown", so a
// platform without /proc never warns.
// The thresholds, each with the reason it is where it is.
const (
// WarnAnon is resident anonymous memory a unit this size should not
// hold for long: well under the 10.9 GB the forge was killed at.
WarnAnon = 6 << 30
// WarnAnonFor is how long it must stay over, so a spike while
// importing a large repository does not warn.
WarnAnonFor = 10 * time.Minute
// WarnGrowthPerHour is steady growth no workload explains: a leak
// that has not yet reached WarnAnon is visible by its slope.
WarnGrowthPerHour = 300 << 20
// WarnGrowthFor is how long every hour must have grown that much.
WarnGrowthFor = 3 * time.Hour
// WarnHostAvailable is the host's MemAvailable under which the
// kernel is about to reclaim from or kill something.
WarnHostAvailable = 2 << 30
// WarnSwapUsed is the fraction of swap in use past which the host
// has nowhere left to put pages.
WarnSwapUsed = 0.9
// WarnCgroupFraction is the share of the cgroup's memory.max at
// which the unit is close enough to be killed to warrant a look.
WarnCgroupFraction = 0.8
)
// Warning kinds. A kind is the stable name of a condition.
const (
WarnKindAnon = "anon-high"
WarnKindGrowth = "anon-growth"
WarnKindHost = "host-low"
WarnKindSwap = "swap-full"
WarnKindCgroup = "cgroup-high"
)
// MemWarning is one memory condition as data: its kind, the number that
// tripped it and the line it was measured against. Words are made by
// whoever shows it. Units: bytes, except growth (bytes per hour) and
// swap (parts per thousand used).
type MemWarning struct {
Kind string `json:"kind"`
Value uint64 `json:"value"`
Threshold uint64 `json:"threshold"`
Since time.Time `json:"since,omitzero"`
// Forge is what the forge process itself holds when this was seen
// (anonymous plus file memory), 0 if unknown: a server short of
// memory may be short because of something else.
Forge uint64 `json:"forge,omitempty"`
}
// procRead reads a /proc file; tests replace it.
var procRead = os.ReadFile
// procMem is what /proc says about this process and host, in bytes.
type procMem struct {
Anon, File uint64
Available, SwapFree, SwapTotal uint64
}
// parseStatus reads RssAnon and RssFile from /proc/self/status text,
// falling back to VmRSS as all-anonymous when the kernel (before 4.5)
// does not split it.
func parseStatus(text string) (anon, file uint64) {
var rss uint64
var split bool
for _, line := range strings.Split(text, "\n") {
k, v, ok := strings.Cut(line, ":")
if !ok {
continue
}
switch k {
case "RssAnon":
anon, split = kb(v), true
case "RssFile":
file = kb(v)
case "VmRSS":
rss = kb(v)
}
}
if !split {
return rss, 0
}
return anon, file
}
// parseMeminfo reads the host's available memory and swap from
// /proc/meminfo text.
func parseMeminfo(text string) (avail, swapFree, swapTotal uint64) {
for _, line := range strings.Split(text, "\n") {
k, v, ok := strings.Cut(line, ":")
if !ok {
continue
}
switch k {
case "MemAvailable":
avail = kb(v)
case "SwapFree":
swapFree = kb(v)
case "SwapTotal":
swapTotal = kb(v)
}
}
return
}
// kb parses " 1234 kB" into bytes; 0 when it is not one.
func kb(v string) uint64 {
f := strings.Fields(v)
if len(f) == 0 {
return 0
}
n, err := strconv.ParseUint(f[0], 10, 64)
if err != nil {
return 0
}
return n << 10
}
func readProcMem() procMem {
var p procMem
if b, err := procRead("/proc/self/status"); err == nil {
p.Anon, p.File = parseStatus(string(b))
}
if b, err := procRead("/proc/meminfo"); err == nil {
p.Available, p.SwapFree, p.SwapTotal = parseMeminfo(string(b))
}
return p
}
// Warnings is every memory condition the samples show, newest sample
// last. cap is the cgroup's memory.max, 0 when none. Pure: the samples
// carry everything, so a test can build a day of them.
func Warnings(samples []MemSample, cap uint64) []MemWarning {
if len(samples) == 0 {
return nil
}
last := samples[len(samples)-1]
var out []MemWarning
// Resident anonymous memory over the line for the whole window.
if last.Anon >= WarnAnon {
first := last.At
for i := len(samples) - 1; i >= 0 && samples[i].Anon >= WarnAnon; i-- {
first = samples[i].At
}
if last.At.Sub(first) >= WarnAnonFor {
out = append(out, MemWarning{Kind: WarnKindAnon, Value: last.Anon, Threshold: WarnAnon, Since: first})
}
}
// Growth: each of the last three hours added at least the line.
if w, ok := growth(samples); ok {
out = append(out, w)
}
if last.Available > 0 && last.Available < WarnHostAvailable {
out = append(out, MemWarning{Kind: WarnKindHost, Value: last.Available, Threshold: WarnHostAvailable, Since: last.At})
}
if last.SwapTotal > 0 {
used := last.SwapTotal - min(last.SwapFree, last.SwapTotal)
if float64(used) >= WarnSwapUsed*float64(last.SwapTotal) {
out = append(out, MemWarning{Kind: WarnKindSwap, Value: used * 1000 / last.SwapTotal,
Threshold: uint64(WarnSwapUsed * 1000), Since: last.At})
}
}
if cap > 0 && last.Anon > 0 && float64(last.Anon) >= WarnCgroupFraction*float64(cap) {
out = append(out, MemWarning{Kind: WarnKindCgroup, Value: last.Anon, Threshold: uint64(WarnCgroupFraction * float64(cap)), Since: last.At})
}
if last.Anon > 0 {
for i := range out {
out[i].Forge = last.Anon + last.File
}
}
return out
}
// growth answers whether resident anonymous memory rose by the line in
// every one of the last WarnGrowthFor's hours.
func growth(samples []MemSample) (MemWarning, bool) {
last := samples[len(samples)-1]
if last.Anon == 0 {
return MemWarning{}, false
}
hours := int(WarnGrowthFor / time.Hour)
at := func(h int) (MemSample, bool) { // the reading at h hours back, or just before
cut := last.At.Add(-time.Duration(h) * time.Hour)
for i := len(samples) - 1; i >= 0; i-- {
if !samples[i].At.After(cut) {
return samples[i], true
}
}
return MemSample{}, false
}
prev := last
for h := 1; h <= hours; h++ {
s, ok := at(h)
if !ok || s.Anon == 0 || prev.Anon < s.Anon || prev.Anon-s.Anon < WarnGrowthPerHour {
return MemWarning{}, false
}
prev = s
}
return MemWarning{Kind: WarnKindGrowth, Value: (last.Anon - prev.Anon) / uint64(hours),
Threshold: WarnGrowthPerHour, Since: prev.At}, true
}
// Alerter is told when a memory warning starts and when it clears, with
// the reading it happened at: once each, never per sample. It is the seam a
// notifier plugs into so this package imports neither the API nor
// notify. A start is reported again after a restart, because the
// sampler cannot know what an earlier process already said; the
// alerter holds that memory.
type Alerter interface {
WarningStarted(w MemWarning, at time.Time)
WarningCleared(was MemWarning, now MemSample, after time.Duration)
}
// SetAlerter chooses who is told of warning edges; nil tells nobody.
func SetAlerter(a Alerter) {
memoryWatch.mu.Lock()
memoryWatch.alert = a
memoryWatch.mu.Unlock()
}
// alertQueue is how many edges may wait for the alerter before more are
// dropped (and logged): a hung notifier must not stall the sampler.
const alertQueue = 16
// memWatch writes one log line when a warning starts and one when it
// clears, never one per sample. The journal is what a person or an
// alerting rule reads when no page is open. The same edges go to the
// alerter, on a goroutine of its own that takes them in order.
type memWatch struct {
mu sync.Mutex
open map[string]MemWarning
since map[string]time.Time
logf func(format string, args ...any)
alert Alerter
queue chan func()
pending sync.WaitGroup
}
// send hands fn to the alerter's goroutine; a full queue drops it. The
// goroutine recovers a panic so a faulty alerter ends one edge, not the
// process. Called with mu held.
func (w *memWatch) send(logf func(string, ...any), fn func()) {
if w.queue == nil {
w.queue = make(chan func(), alertQueue)
go func() {
for f := range w.queue {
func() {
defer w.pending.Done()
defer func() {
if r := recover(); r != nil {
logf("memory: alerter failed: %v", r)
}
}()
f()
}()
}
}()
}
w.pending.Add(1)
select {
case w.queue <- fn:
default:
w.pending.Done()
logf("memory: alerter is behind; an alert was dropped")
}
}
// wait blocks until every queued edge has been handed over, for tests.
func (w *memWatch) wait() { w.pending.Wait() }
func (w *memWatch) observe(cur MemSample, ws []MemWarning) {
now := cur.At
w.mu.Lock()
defer w.mu.Unlock()
logf := w.logf
if logf == nil {
logf = log.Printf
}
if w.open == nil {
w.open, w.since = map[string]MemWarning{}, map[string]time.Time{}
}
present := map[string]bool{}
for _, x := range ws {
present[x.Kind] = true
if _, ok := w.open[x.Kind]; !ok {
w.open[x.Kind], w.since[x.Kind] = x, now
logf("memory: warning started kind=%s value=%d threshold=%d", x.Kind, x.Value, x.Threshold)
if a := w.alert; a != nil {
w.send(logf, func() { a.WarningStarted(x, now) })
}
}
}
for k, was := range w.open {
if !present[k] {
t := w.since[k]
delete(w.open, k)
delete(w.since, k)
after := now.Sub(t)
logf("memory: warning cleared kind=%s after=%s", k, after.Round(time.Second))
if a := w.alert; a != nil {
w.send(logf, func() { a.WarningCleared(was, cur, after) })
}
}
}
}
var memoryWatch memWatch