package obs
import (
"bufio"
"context"
"math"
"os"
"path/filepath"
"runtime"
"runtime/debug"
"strconv"
"strings"
"sync"
"time"
)
// MemSample is one reading of this process's memory.
type MemSample struct {
At time.Time `json:"at"`
Heap uint64 `json:"heap"` // live and not-yet-swept heap objects
Swap uint64 `json:"swap"` // the unit's swapped-out bytes, 0 if unknown
GCs uint32 `json:"gcs"` // collections since start
// What the process holds beyond the heap, in bytes; each 0 when the
// platform has no /proc or cgroup to ask (macOS, tests). Zero is
// "unknown", and no warning depends on an unknown.
Anon uint64 `json:"anon,omitempty"` // resident anonymous memory (RssAnon, else VmRSS)
File uint64 `json:"file,omitempty"` // resident file-backed memory (RssFile)
Current uint64 `json:"current,omitempty"` // the cgroup's memory.current
Available uint64 `json:"available,omitempty"` // the host's MemAvailable
SwapFree uint64 `json:"swapFree,omitempty"` // the host's SwapFree
SwapTotal uint64 `json:"swapTotal,omitempty"` // the host's SwapTotal
}
// Memory is what the operations page knows about memory now: the heap
// against the two lines that matter for it, the process's own memory
// target (GOMEMLIMIT) and the unit's hard cap (its cgroup's memory.max),
// and the last day of readings.
//
// The target is where the collector starts working harder; the cap is
// where the kernel kills the process. A heap above the target is
// already a fault, because the collector then runs nearly without
// pause and the heap still does not shrink: that is how the 10 GB
// lexer-memo leak looked for days, with a 7 GiB target.
type Memory struct {
Heap uint64 `json:"heap"`
Limit uint64 `json:"limit"` // GOMEMLIMIT; 0 when unset
Cap uint64 `json:"cap"` // the cgroup's memory.max; 0 when unbounded or unknown
Swap uint64 `json:"swap"`
GCs uint32 `json:"gcs"`
// GCPerMinute is collections a minute over the last sampled ten
// minutes: a collector running many times a minute is a heap that
// cannot shrink.
GCPerMinute float64 `json:"gcPerMinute"`
// OverSince is when the heap last went above the target and stayed
// there, zero when it is under it now.
OverSince time.Time `json:"overSince,omitzero"`
// What the process holds beyond the heap now (see MemSample), and
// the conditions the last day of readings shows. The Go limit bounds
// the heap only; SQLite's memory is outside it, so these are the
// numbers that see it.
Anon uint64 `json:"anon,omitempty"`
File uint64 `json:"file,omitempty"`
Current uint64 `json:"current,omitempty"`
Available uint64 `json:"available,omitempty"`
SwapFree uint64 `json:"swapFree,omitempty"`
SwapTotal uint64 `json:"swapTotal,omitempty"`
Warnings []MemWarning `json:"warnings"`
Samples []MemSample `json:"samples"`
}
// MemorySampleEvery and MemoryKept are the sampler's cadence and memory:
// one reading a minute, a day of them. The growth rule needs
// WarnGrowthFor (3 h) of them and the day covers it.
const (
MemorySampleEvery = time.Minute
MemoryKept = 24 * time.Hour
)
type memRing struct {
mu sync.Mutex
samples []MemSample
}
var memory memRing
// StartMemorySampler reads memory once a minute until ctx ends. One
// reading is taken at once, so the page has a point the moment it opens.
//
// Each reading is also judged, and the journal gets one line when a
// warning starts and one when it clears, so a problem is on record
// without anybody having the operations page open.
func StartMemorySampler(ctx context.Context) {
sample := func() {
s := readMemory()
memory.add(s)
memory.mu.Lock()
all := append([]MemSample(nil), memory.samples...)
memory.mu.Unlock()
memoryWatch.observe(s, Warnings(all, cgroupLimit().max))
}
sample()
go func() {
t := time.NewTicker(MemorySampleEvery)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
sample()
}
}
}()
}
func (m *memRing) add(s MemSample) {
m.mu.Lock()
defer m.mu.Unlock()
m.samples = append(m.samples, s)
cut := s.At.Add(-MemoryKept)
i := 0
for i < len(m.samples) && m.samples[i].At.Before(cut) {
i++
}
m.samples = m.samples[i:]
}
// MemoryNow is a fresh reading with the day's samples behind it.
func MemoryNow() Memory {
now := readMemory()
memory.mu.Lock()
samples := append([]MemSample(nil), memory.samples...)
memory.mu.Unlock()
m := Memory{Heap: now.Heap, Swap: now.Swap, GCs: now.GCs, Samples: samples,
Limit: memoryLimit(), Cap: cgroupLimit().max,
Anon: now.Anon, File: now.File, Current: now.Current,
Available: now.Available, SwapFree: now.SwapFree, SwapTotal: now.SwapTotal}
m.Warnings = Warnings(append(samples, now), m.Cap)
if m.Warnings == nil {
m.Warnings = []MemWarning{}
}
m.GCPerMinute = gcRate(append(samples, now), 10*time.Minute)
if m.Limit > 0 && now.Heap > m.Limit {
m.OverSince = now.At
for i := len(samples) - 1; i >= 0 && samples[i].Heap > m.Limit; i-- {
m.OverSince = samples[i].At
}
}
return m
}
func readMemory() MemSample {
var ms runtime.MemStats
runtime.ReadMemStats(&ms)
p := readProcMem()
return MemSample{At: time.Now(), Heap: ms.HeapAlloc, GCs: ms.NumGC, Swap: cgroupUint("memory.swap.current"),
Anon: p.Anon, File: p.File, Current: cgroupLimit().current(),
Available: p.Available, SwapFree: p.SwapFree, SwapTotal: p.SwapTotal}
}
// memoryLimit is GOMEMLIMIT as the runtime holds it; unset reads as the
// largest int64, which means no target.
func memoryLimit() uint64 {
l := debug.SetMemoryLimit(-1)
if l <= 0 || l == math.MaxInt64 {
return 0
}
return uint64(l)
}
func gcRate(s []MemSample, over time.Duration) float64 {
if len(s) < 2 {
return 0
}
last := s[len(s)-1]
first := s[0]
for _, x := range s {
if last.At.Sub(x.At) <= over {
first = x
break
}
}
mins := last.At.Sub(first.At).Minutes()
if mins <= 0 || last.GCs < first.GCs {
return 0
}
return float64(last.GCs-first.GCs) / mins
}
// cgroupDir is this process's cgroup v2 directory, "" when there is none
// (not Linux, or cgroup v1).
var cgroupDir = sync.OnceValue(func() string {
f, err := os.Open("/proc/self/cgroup")
if err != nil {
return ""
}
defer f.Close()
sc := bufio.NewScanner(f)
for sc.Scan() {
if rest, ok := strings.CutPrefix(sc.Text(), "0::"); ok {
return filepath.Join("/sys/fs/cgroup", rest)
}
}
return ""
})
// cgroupBound is the memory limit that binds this process and the cgroup
// that carries it.
type cgroupBound struct {
dir string
max uint64
}
// current is how much the bounded cgroup holds, which is what the kernel
// weighs against the limit: this process and everything beside it there.
func (b cgroupBound) current() uint64 {
if b.dir == "" {
return cgroupUint("memory.current")
}
return readUint(filepath.Join(b.dir, "memory.current"))
}
// cgroupLimit finds the tightest memory.max between this process's
// cgroup and the root. The process's own cgroup is often not the one
// with the limit: under systemd the forge runs in a child of its unit
// (quarry.service/main), where memory.max reads "max", and the unit's
// MemoryMax sits on the parent. Reading only the leaf gave no limit, so
// the near-the-limit warning could never fire.
var cgroupLimit = sync.OnceValue(func() cgroupBound {
return cgroupLimitFrom(cgroupDir(), "/sys/fs/cgroup")
})
func cgroupLimitFrom(dir, root string) cgroupBound {
var b cgroupBound
if dir == "" {
return b
}
for d := filepath.Clean(dir); strings.HasPrefix(d, root) && d != root; d = filepath.Dir(d) {
if n := readUint(filepath.Join(d, "memory.max")); n > 0 && (b.max == 0 || n < b.max) {
b = cgroupBound{dir: d, max: n}
}
}
return b
}
// readUint reads one number from a file; 0 for "max", absent or
// unreadable, so an unknown never reads as a bound.
func readUint(path string) uint64 {
b, err := os.ReadFile(path)
if err != nil {
return 0
}
n, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64)
if err != nil {
return 0
}
return n
}
// cgroupUint reads one number from the cgroup; 0 for "max", absent or
// unreadable, so an unknown never reads as a bound.
func cgroupUint(name string) uint64 {
dir := cgroupDir()
if dir == "" {
return 0
}
b, err := os.ReadFile(filepath.Join(dir, name))
if err != nil {
return 0
}
n, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64)
if err != nil {
return 0
}
return n
}