Skip to content
quarry Artifact [65bd3b7a7a]
Login

Artifact [65bd3b7a7a]

Artifact 65bd3b7a7ac4a5b92576176b2673438c218f55dc4cd2461114d0af8e6a4f2c8c:


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.At, Warnings(all, cgroupUint("memory.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: cgroupUint("memory.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: cgroupUint("memory.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 ""
})

// 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
}