378 lines
12 KiB
Go
378 lines
12 KiB
Go
package runtimestats
|
|
|
|
import (
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// A real dump, trimmed. The parser is the one part of this package that reads text a Go
|
|
// release produces, so the fixture is kept verbatim rather than idealised — the argument
|
|
// lists, the "+0x" offsets and the "created by … in goroutine N" suffix are all things the
|
|
// parser has to survive.
|
|
const sampleDump = `goroutine 1 [chan receive, 940 minutes]:
|
|
main.run(0xc000122000)
|
|
/app/cmd/memby-server/main.go:430 +0x8f4
|
|
main.main()
|
|
/app/cmd/memby-server/main.go:41 +0x1f
|
|
|
|
goroutine 18 [select]:
|
|
github.com/ponzischeme89/memby/server/internal/library.(*Ingester).Run(0xc0001a2000, {0x1a4c0d0, 0xc0000a2000})
|
|
/app/internal/library/ingest.go:118 +0x145
|
|
created by github.com/ponzischeme89/memby/server/internal/runtimestats.Go in goroutine 1
|
|
|
|
goroutine 40 [IO wait, 3 minutes]:
|
|
internal/poll.runtime_pollWait(0x7f2c1c0, 0x72)
|
|
/usr/local/go/src/runtime/netpoll.go:351 +0x85
|
|
net/http.(*conn).serve(0xc000310000, {0x1a4c0d0, 0xc0003a0000})
|
|
/usr/local/go/src/net/http/server.go:2092 +0x5db
|
|
created by net/http.(*Server).Serve in goroutine 55
|
|
|
|
goroutine 41 [GC worker (idle)]:
|
|
runtime.gopark(0x0, 0x0, 0x0, 0x0, 0x0)
|
|
/usr/local/go/src/runtime/proc.go:402 +0xce
|
|
runtime.gcBgMarkWorker(0xc0000581c0)
|
|
/usr/local/go/src/runtime/mgc.go:1310 +0xe5
|
|
created by runtime.gcBgMarkStartWorkers in goroutine 1
|
|
|
|
goroutine 77 [semacquire, 3 minutes, locked to thread]:
|
|
github.com/jackc/puddle/v2.(*Pool).acquire(0xc00019c000)
|
|
/root/go/pkg/mod/github.com/jackc/puddle/v2@v2.2.2/pool.go:481 +0x10a
|
|
created by github.com/jackc/pgx/v5/pgxpool.NewWithConfig in goroutine 1
|
|
`
|
|
|
|
func TestParseGoroutinesReadsHeaderStateAndFrames(t *testing.T) {
|
|
list := parseGoroutines(sampleDump)
|
|
if len(list) != 5 {
|
|
t.Fatalf("parsed %d goroutines, want 5", len(list))
|
|
}
|
|
|
|
first := list[0]
|
|
if first.ID != 1 || first.State != "chan receive" || first.WaitMinutes != 940 {
|
|
t.Fatalf("first = %+v", first)
|
|
}
|
|
if first.Frames[0].Function != "main.run" {
|
|
t.Fatalf("first frame function = %q", first.Frames[0].Function)
|
|
}
|
|
// The offset is stripped so two goroutines in the same place group together rather
|
|
// than differing by an address.
|
|
if first.Frames[0].File != "/app/cmd/memby-server/main.go:430" {
|
|
t.Fatalf("first frame file = %q", first.Frames[0].File)
|
|
}
|
|
|
|
if list[1].CreatedBy == "" {
|
|
t.Fatal("created-by line was dropped")
|
|
}
|
|
// A state with trailing detail after a second comma keeps its state and its wait.
|
|
last := list[4]
|
|
if last.State != "semacquire" || last.WaitMinutes != 3 {
|
|
t.Fatalf("last = %+v", last)
|
|
}
|
|
}
|
|
|
|
// The categories have to partition the total or the breakdown is misleading, which is
|
|
// worse than not offering one.
|
|
func TestBuildReportPartitionsAndAttributes(t *testing.T) {
|
|
report := buildReport(sampleDump, time.Now())
|
|
if report.Total != 5 {
|
|
t.Fatalf("total = %d", report.Total)
|
|
}
|
|
counted := 0
|
|
for _, category := range report.Categories {
|
|
counted += category.Count
|
|
}
|
|
if counted != report.Total {
|
|
t.Fatalf("categories sum to %d, total %d", counted, report.Total)
|
|
}
|
|
attributed := 0
|
|
for _, component := range report.Components {
|
|
attributed += component.Count
|
|
}
|
|
if attributed != report.Total {
|
|
t.Fatalf("components sum to %d, total %d", attributed, report.Total)
|
|
}
|
|
|
|
got := map[string]string{}
|
|
for _, group := range report.Groups {
|
|
got[group.Component] = group.Function
|
|
}
|
|
// Memby's own packages win over the libraries they call, and the frame reported is
|
|
// the one that names the work rather than the runtime plumbing every parked goroutine
|
|
// shares.
|
|
if got["Library sync"] != "github.com/ponzischeme89/memby/server/internal/library.(*Ingester).Run" {
|
|
t.Fatalf("library group = %q", got["Library sync"])
|
|
}
|
|
if got["HTTP server"] != "net/http.(*conn).serve" {
|
|
t.Fatalf("http group = %q", got["HTTP server"])
|
|
}
|
|
if got["Database pool"] != "github.com/jackc/puddle/v2.(*Pool).acquire" {
|
|
t.Fatalf("database group = %q", got["Database pool"])
|
|
}
|
|
}
|
|
|
|
func TestCategoryForKnownStates(t *testing.T) {
|
|
cases := map[string]Category{
|
|
"running": CategoryRunning,
|
|
"runnable": CategoryRunning,
|
|
"syscall": CategoryRunning,
|
|
"IO wait": CategoryIO,
|
|
"select": CategoryWaiting,
|
|
"chan receive": CategoryWaiting,
|
|
"semacquire": CategoryWaiting,
|
|
"sync.Mutex.Lock": CategoryWaiting,
|
|
"sleep": CategoryTimers,
|
|
"timer goroutine": CategoryTimers,
|
|
"GC worker (idle)": CategoryRuntime,
|
|
"force gc (idle)": CategoryRuntime,
|
|
"finalizer wait": CategoryRuntime,
|
|
// The runtime's own wait outranks the "wait" rule, or the collector would be
|
|
// reported as a component sitting on a channel.
|
|
"wait for GC cycle": CategoryRuntime,
|
|
"something new": CategoryOther,
|
|
}
|
|
for state, want := range cases {
|
|
if got := categoryFor(state); got != want {
|
|
t.Errorf("categoryFor(%q) = %q, want %q", state, got, want)
|
|
}
|
|
}
|
|
}
|
|
|
|
// A live report has to hold together too: this is the one test that would notice a Go
|
|
// release changing the dump format under the parser.
|
|
func TestCollectGoroutinesAgreesWithTheRuntime(t *testing.T) {
|
|
blocked := make(chan struct{})
|
|
var ready sync.WaitGroup
|
|
ready.Add(1)
|
|
go func() {
|
|
ready.Done()
|
|
<-blocked
|
|
}()
|
|
ready.Wait()
|
|
defer close(blocked)
|
|
|
|
report := CollectGoroutines()
|
|
if report.Total < 2 {
|
|
t.Fatalf("total = %d", report.Total)
|
|
}
|
|
counted := 0
|
|
for _, category := range report.Categories {
|
|
counted += category.Count
|
|
}
|
|
if counted != report.Total {
|
|
t.Fatalf("categories sum to %d, total %d", counted, report.Total)
|
|
}
|
|
if report.CollectedInMs < 0 {
|
|
t.Fatalf("collectedInMs = %v", report.CollectedInMs)
|
|
}
|
|
if !strings.Contains(StackDump(), "goroutine ") {
|
|
t.Fatal("stack dump does not look like a stack dump")
|
|
}
|
|
}
|
|
|
|
/* ---------- trends ---------- */
|
|
|
|
func series(values []float64, every time.Duration) ([]float64, []time.Time) {
|
|
at := make([]time.Time, len(values))
|
|
base := time.Date(2026, 8, 19, 9, 0, 0, 0, time.UTC)
|
|
for index := range values {
|
|
at[index] = base.Add(time.Duration(index) * every)
|
|
}
|
|
return values, at
|
|
}
|
|
|
|
func TestTrendRefusesToAnswerWithoutEnoughHistory(t *testing.T) {
|
|
// Four readings a minute apart is a gateway that restarted three minutes ago, not a
|
|
// trend, and "steady" would be a claim the data does not support.
|
|
points, at := series([]float64{20, 24, 30, 44}, time.Minute)
|
|
if got := trendOf(points, at, 5); got.Direction != DirectionUnknown {
|
|
t.Fatalf("direction = %q, want unknown", got.Direction)
|
|
}
|
|
}
|
|
|
|
func TestTrendIgnoresABurstAndNamesASustainedRise(t *testing.T) {
|
|
// A burst in the middle — a library import, a household arriving home — must not be
|
|
// reported as a rise; medians are what make that true where a fitted line would not.
|
|
burst := []float64{40, 41, 40, 42, 41, 300, 290, 40, 41, 40, 42, 41}
|
|
points, at := series(burst, 5*time.Minute)
|
|
if got := trendOf(points, at, 5); got.Direction != DirectionSteady {
|
|
t.Fatalf("burst direction = %q (%+v), want steady", got.Direction, got)
|
|
}
|
|
|
|
leak := make([]float64, 12)
|
|
for index := range leak {
|
|
leak[index] = float64(40 + index*5)
|
|
}
|
|
points, at = series(leak, 5*time.Minute)
|
|
rising := trendOf(points, at, 5)
|
|
if rising.Direction != DirectionRising {
|
|
t.Fatalf("leak direction = %q (%+v), want rising", rising.Direction, rising)
|
|
}
|
|
if rising.PerHour <= 0 {
|
|
t.Fatalf("perHour = %v, want a positive rate", rising.PerHour)
|
|
}
|
|
if rising.Latest != 95 || rising.First != 40 {
|
|
t.Fatalf("first/latest = %v/%v", rising.First, rising.Latest)
|
|
}
|
|
}
|
|
|
|
func TestTrendFloorKeepsAQuietGatewaySteady(t *testing.T) {
|
|
// One extra goroutine over an hour is not a leak, and without the absolute floor the
|
|
// proportional threshold would call it one on a gateway holding a handful.
|
|
points, at := series([]float64{8, 8, 9, 9, 9, 9, 9, 10}, 10*time.Minute)
|
|
if got := trendOf(points, at, 5); got.Direction != DirectionSteady {
|
|
t.Fatalf("direction = %q (%+v), want steady", got.Direction, got)
|
|
}
|
|
}
|
|
|
|
/* ---------- the verdict ---------- */
|
|
|
|
func risingTrend(first, latest, perHour float64) Trend {
|
|
return Trend{
|
|
Direction: DirectionRising, PerHour: perHour, First: first, Latest: latest,
|
|
SpanSeconds: 3600, Points: 60,
|
|
}
|
|
}
|
|
|
|
func TestAssessSaysNothingWhenThereIsNothingToSay(t *testing.T) {
|
|
health := assess(Snapshot{
|
|
Memory: Memory{MemoryLimit: 384 << 20, HeapInuse: 16 << 20, HeapShare: 0.04},
|
|
GoroutineTrend: Trend{Direction: DirectionSteady, SpanSeconds: 7200},
|
|
HeapTrend: Trend{Direction: DirectionSteady},
|
|
})
|
|
if health.Level != LevelOK || len(health.Notes) != 0 {
|
|
t.Fatalf("health = %+v", health)
|
|
}
|
|
if !strings.Contains(health.Summary, "Nothing abnormal") {
|
|
t.Fatalf("summary = %q", health.Summary)
|
|
}
|
|
}
|
|
|
|
func TestAssessNamesTheAreaToInvestigate(t *testing.T) {
|
|
health := assess(Snapshot{
|
|
Memory: Memory{MemoryLimit: 384 << 20, HeapInuse: 350 << 20, HeapShare: 0.92},
|
|
GoroutineTrend: risingTrend(60, 400, 340),
|
|
HeapTrend: Trend{Direction: DirectionSteady},
|
|
})
|
|
if health.Level != LevelBad {
|
|
t.Fatalf("level = %q", health.Level)
|
|
}
|
|
areas := strings.Join(health.Areas, ",")
|
|
if !strings.Contains(areas, "Memory") || !strings.Contains(areas, "Goroutines") {
|
|
t.Fatalf("areas = %v", health.Areas)
|
|
}
|
|
// A verdict with no area is the number that could not be acted on, which is the whole
|
|
// defect this replaced.
|
|
for _, note := range health.Notes {
|
|
if note.Area == "" {
|
|
t.Fatalf("note without an area: %+v", note)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestAssessSeparatesADriftFromAFlood(t *testing.T) {
|
|
drift := assess(Snapshot{GoroutineTrend: risingTrend(40, 60, 20)})
|
|
if drift.Level != LevelWatch {
|
|
t.Fatalf("drift level = %q", drift.Level)
|
|
}
|
|
flood := assess(Snapshot{GoroutineTrend: risingTrend(40, 400, 360)})
|
|
if flood.Level != LevelBad {
|
|
t.Fatalf("flood level = %q", flood.Level)
|
|
}
|
|
// A rise slower than the leak threshold is ordinary breathing and must not be a note
|
|
// at all, or the card cries wolf on every busy evening.
|
|
quiet := assess(Snapshot{GoroutineTrend: risingTrend(40, 44, 4)})
|
|
if quiet.Level != LevelOK || len(quiet.Notes) != 0 {
|
|
t.Fatalf("quiet = %+v", quiet)
|
|
}
|
|
}
|
|
|
|
func TestAssessNoticesAWorkerThatKeepsRestarting(t *testing.T) {
|
|
health := assess(Snapshot{Workers: []Worker{
|
|
{Name: "Library ingest", Component: "Library sync", State: WorkerRunning, Starts: 5},
|
|
{Name: "Startup library sync", Component: "Library sync", State: WorkerFinished, Starts: 1},
|
|
}})
|
|
if health.Level != LevelWatch || len(health.Notes) != 1 {
|
|
t.Fatalf("health = %+v", health)
|
|
}
|
|
if !strings.Contains(health.Notes[0].Message, "Library ingest") {
|
|
t.Fatalf("note = %q", health.Notes[0].Message)
|
|
}
|
|
}
|
|
|
|
/* ---------- the registry ---------- */
|
|
|
|
func TestWorkersRecordRunningAndFinished(t *testing.T) {
|
|
resetWorkers()
|
|
t.Cleanup(resetWorkers)
|
|
|
|
release := make(chan struct{})
|
|
var running sync.WaitGroup
|
|
running.Add(1)
|
|
Go("Library ingest", "Library sync", func() {
|
|
running.Done()
|
|
<-release
|
|
})
|
|
running.Wait()
|
|
|
|
var done sync.WaitGroup
|
|
done.Add(1)
|
|
Go("Startup library sync", "Library sync", func() { done.Done() })
|
|
done.Wait()
|
|
// The finished worker's goroutine has run its body; give the deferred end() the moment
|
|
// it needs to land before reading the registry.
|
|
waitFor(t, func() bool { return stateOf(t, "Startup library sync") == WorkerFinished })
|
|
|
|
list := Workers()
|
|
if len(list) != 2 {
|
|
t.Fatalf("workers = %+v", list)
|
|
}
|
|
// Running first, so the table does not bury the thing that is still working under the
|
|
// startup jobs that are not.
|
|
if list[0].Name != "Library ingest" || list[0].State != WorkerRunning {
|
|
t.Fatalf("first = %+v", list[0])
|
|
}
|
|
if list[1].State != WorkerFinished || list[1].Stopped.IsZero() {
|
|
t.Fatalf("second = %+v", list[1])
|
|
}
|
|
close(release)
|
|
waitFor(t, func() bool { return stateOf(t, "Library ingest") == WorkerFinished })
|
|
|
|
// A restart is counted rather than replacing the record: a worker restarted in a loop
|
|
// reads exactly like a healthy one from a single snapshot and does not from Starts.
|
|
Go("Library ingest", "Library sync", func() {})
|
|
waitFor(t, func() bool { return startsOf(t, "Library ingest") == 2 })
|
|
}
|
|
|
|
func stateOf(t *testing.T, name string) WorkerState {
|
|
t.Helper()
|
|
for _, worker := range Workers() {
|
|
if worker.Name == name {
|
|
return worker.State
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func startsOf(t *testing.T, name string) int {
|
|
t.Helper()
|
|
for _, worker := range Workers() {
|
|
if worker.Name == name {
|
|
return worker.Starts
|
|
}
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func waitFor(t *testing.T, done func() bool) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
if done() {
|
|
return
|
|
}
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
t.Fatal("condition not reached")
|
|
}
|