[management] Add a go test -json summarizer and log test store setup time

The Management / Unit (mysql) job hit the 20 minute go test budget in management/server and left only a goroutine dump of several thousand leaked goroutines, with the "running tests:" header far out of reach of the job log viewer. The last live test goroutine was runnable, not blocked: the suite was still progressing when the alarm fired. tools/gotestsummary reads a go test -json stream and prints one line per finished test, the output of failed tests, the head of a package panic (the timeout header and the running tests list), tests that never reported a result, and the slowest tests per package. NewTestStoreFromSQL now logs how long the sqlite seed store and the engine under test took to come up; the summarizer attributes those to the owning test so the per-test store cost on mysql and postgres is visible in CI. Two leaks made the dump unreadable and are fixed here: the sqlite seed store was never closed once a mysql or postgres store replaced it, and buildTestManager passed context.Background() to the metrics, caches and controllers, so every test left its P95 flushers and cleanup loops running until the process exited.
2026-09-12 17:59:06 +02:00 · 2026-09-12 12:00:37 +00:00
parent 973173be9c
commit ba9b226b4b
3 changed files with 355 additions and 6 deletions
@@ -3516,7 +3516,13 @@ func buildTestManager(t testing.TB, store store.Store, nmdataStore *networkmapdb

 	eventStore := &activity.InMemoryEventStore{}

-	metrics, err := telemetry.NewDefaultAppMetrics(context.Background())
+	// Everything built here watches this context; cancelling it on cleanup stops
+	// the metrics flushers, caches and controllers instead of leaking them for
+	// the rest of the package run.
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	metrics, err := telemetry.NewDefaultAppMetrics(ctx)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -3535,8 +3541,6 @@ func buildTestManager(t testing.TB, store store.Store, nmdataStore *networkmapdb
 		Return(nil).
 		AnyTimes()

-	ctx := context.Background()
-
 	cacheStore, err := cache.NewStore(ctx, 100*time.Millisecond, 300*time.Millisecond, 100)
 	if err != nil {
 		return nil, nil, err
@@ -685,6 +685,7 @@ func getMigrationsPostAuto(ctx context.Context) []migrationFunc {
 // NewTestStoreFromSQL is only used in tests. It will create a test database base of the store engine set in env.
 // Optionally it can load a SQL file to the database. If the filename is empty it will return an empty database
 func NewTestStoreFromSQL(ctx context.Context, filename string, dataDir string) (Store, func(), error) {
+	start := time.Now()
 	kind := getStoreEngineFromEnv()
 	if kind == "" {
 		kind = types.SqliteStoreEngine
@@ -722,10 +723,15 @@ func NewTestStoreFromSQL(ctx context.Context, filename string, dataDir string) (
 	var sqlStore Store
 	var cleanup func()

+	sqliteReady := time.Now()
 	maxRetries := 2
 	for i := 0; i < maxRetries; i++ {
 		sqlStore, cleanup, err = getSqlStoreEngine(ctx, store, kind)
 		if err == nil {
+			// Parsed by tools/gotestsummary to attribute store setup time per test.
+			log.WithContext(ctx).Infof("test store created: engine=%s total=%s sqlite=%s engine_setup=%s",
+				kind, time.Since(start).Round(time.Millisecond), sqliteReady.Sub(start).Round(time.Millisecond),
+				time.Since(sqliteReady).Round(time.Millisecond))
 			return sqlStore, cleanup, nil
 		}
 		if i < maxRetries-1 {
@@ -758,14 +764,15 @@ func addAllGroupToAccount(ctx context.Context, store Store) error {
 	return nil
 }

-func getSqlStoreEngine(ctx context.Context, store *SqlStore, kind types.Engine) (Store, func(), error) {
+func getSqlStoreEngine(ctx context.Context, sqliteStore *SqlStore, kind types.Engine) (Store, func(), error) {
+	store := sqliteStore
 	var cleanup func()
 	var err error
 	switch kind {
 	case types.PostgresStoreEngine:
-		store, cleanup, err = newReusedPostgresStore(ctx, store, kind)
+		store, cleanup, err = newReusedPostgresStore(ctx, sqliteStore, kind)
 	case types.MysqlStoreEngine:
-		store, cleanup, err = newReusedMysqlStore(ctx, store, kind)
+		store, cleanup, err = newReusedMysqlStore(ctx, sqliteStore, kind)
 	default:
 		cleanup = func() {
 			// sqlite doesn't need to be cleaned up
@@ -781,6 +788,11 @@ func getSqlStoreEngine(ctx context.Context, store *SqlStore, kind types.Engine)
 		if store.pool != nil {
 			store.pool.Close()
 		}
+		if store != sqliteStore {
+			// The sqlite store only seeded the engine under test; without this
+			// every test leaks its connection and the opener goroutines.
+			sqliteStore.Close(ctx)
+		}
 	}

 	return store, closeConnection, nil
@@ -0,0 +1,333 @@
+// Package main turns a `go test -json` stream into a readable CI log.
+//
+// It prints one line per top-level test as it finishes, the captured output of
+// every failed test, the head of a package-level panic (which is where Go
+// reports "test timed out" and the list of still-running tests), and ends with
+// the per-package durations and the slowest tests. The exit code is always zero
+// unless the input cannot be read; the `go test` exit code is what CI should act
+// on, so run the two with `set -o pipefail`.
+//
+// Usage:
+//
+//	go test -json ./... | go run ./tools/gotestsummary
+//	go run ./tools/gotestsummary -slowest 60 test-output.jsonl
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"regexp"
+	"sort"
+	"strings"
+	"time"
+)
+
+const (
+	modulePrefix = "github.com/netbirdio/netbird/"
+
+	// failedTestOutputLines bounds how much captured output a single failed
+	// test may print, so one noisy failure cannot flood the job log.
+	failedTestOutputLines = 200
+	// panicHeadLines is enough for the "panic: test timed out" header, the
+	// "running tests:" list, and the first few goroutines of the dump.
+	panicHeadLines = 150
+	// bufferedOutputLines bounds the per-test output kept in memory while the
+	// test runs; only the tail is kept once the cap is reached.
+	bufferedOutputLines = 400
+)
+
+var storeCreatedRe = regexp.MustCompile(`test store created: engine=(\S+) total=(\S+)`)
+
+type event struct {
+	Action  string  `json:"Action"`
+	Package string  `json:"Package"`
+	Test    string  `json:"Test"`
+	Output  string  `json:"Output"`
+	Elapsed float64 `json:"Elapsed"`
+}
+
+type testKey struct {
+	pkg, name string
+}
+
+type testResult struct {
+	pkg, name  string
+	action     string
+	elapsed    time.Duration
+	storeCount int
+	storeTotal time.Duration
+}
+
+type packageResult struct {
+	pkg     string
+	action  string
+	elapsed time.Duration
+}
+
+type summarizer struct {
+	out io.Writer
+
+	output    map[testKey][]string
+	dropped   map[testKey]int
+	stores    map[testKey]storeStats
+	tests     []testResult
+	packages  []packageResult
+	panicHead []string
+	panicking bool
+}
+
+type storeStats struct {
+	count int
+	total time.Duration
+}
+
+func newSummarizer(out io.Writer) *summarizer {
+	return &summarizer{
+		out:     out,
+		output:  make(map[testKey][]string),
+		dropped: make(map[testKey]int),
+		stores:  make(map[testKey]storeStats),
+	}
+}
+
+func main() {
+	slowest := flag.Int("slowest", 40, "number of slowest top-level tests to list")
+	flag.Parse()
+
+	if err := run(flag.Arg(0), *slowest); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}
+
+func run(path string, slowest int) error {
+	in := os.Stdin
+	if path != "" {
+		f, err := os.Open(path)
+		if err != nil {
+			return fmt.Errorf("open %s: %w", path, err)
+		}
+		defer f.Close()
+		in = f
+	}
+
+	s := newSummarizer(os.Stdout)
+	if err := s.consume(in); err != nil {
+		return fmt.Errorf("read input: %w", err)
+	}
+	s.printSummary(slowest)
+	return nil
+}
+
+func (s *summarizer) consume(r io.Reader) error {
+	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 0, 1024*1024), 16*1024*1024)
+	for scanner.Scan() {
+		line := scanner.Bytes()
+		var ev event
+		if err := json.Unmarshal(line, &ev); err != nil {
+			// Build errors and other non-JSON lines are passed through untouched.
+			fmt.Fprintln(s.out, string(line))
+			continue
+		}
+		s.handle(ev)
+	}
+	return scanner.Err()
+}
+
+func (s *summarizer) handle(ev event) {
+	key := testKey{pkg: ev.Package, name: ev.Test}
+	switch ev.Action {
+	case "output":
+		s.handleOutput(key, strings.TrimRight(ev.Output, "\n"))
+	case "pass", "fail", "skip":
+		if ev.Test == "" {
+			s.handlePackageResult(ev)
+			return
+		}
+		s.handleTestResult(key, ev)
+	}
+}
+
+func (s *summarizer) handleOutput(key testKey, line string) {
+	if strings.HasPrefix(line, "panic: ") || strings.HasPrefix(line, "fatal error: ") {
+		s.panicking = true
+	}
+	if s.panicking {
+		// The goroutine dump that follows a panic is kept in panicHead only;
+		// letting it flood the per-test buffers would hide the test's own output.
+		if len(s.panicHead) < panicHeadLines {
+			s.panicHead = append(s.panicHead, line)
+		}
+		return
+	}
+
+	if m := storeCreatedRe.FindStringSubmatch(line); m != nil {
+		if d, err := time.ParseDuration(m[2]); err == nil {
+			st := s.stores[key]
+			st.count++
+			st.total += d
+			s.stores[key] = st
+		}
+	}
+
+	if key.name == "" {
+		return
+	}
+	buf := s.output[key]
+	if len(buf) >= bufferedOutputLines {
+		buf = buf[1:]
+		s.dropped[key]++
+	}
+	s.output[key] = append(buf, line)
+}
+
+func (s *summarizer) handleTestResult(key testKey, ev event) {
+	elapsed := time.Duration(ev.Elapsed * float64(time.Second))
+	st := s.stores[key]
+	s.tests = append(s.tests, testResult{
+		pkg:        ev.Package,
+		name:       ev.Test,
+		action:     ev.Action,
+		elapsed:    elapsed,
+		storeCount: st.count,
+		storeTotal: st.total,
+	})
+
+	if !strings.Contains(ev.Test, "/") || ev.Action == "fail" {
+		fmt.Fprintf(s.out, "--- %s: %s.%s (%s)\n", strings.ToUpper(ev.Action), shortPkg(ev.Package), ev.Test, elapsed.Round(time.Millisecond))
+	}
+	if ev.Action == "fail" {
+		s.printTestOutput(key)
+	}
+	delete(s.output, key)
+	delete(s.dropped, key)
+}
+
+func (s *summarizer) printTestOutput(key testKey) {
+	lines := s.output[key]
+	if len(lines) == 0 {
+		return
+	}
+	skipped := s.dropped[key]
+	if len(lines) > failedTestOutputLines {
+		skipped += len(lines) - failedTestOutputLines
+		lines = lines[len(lines)-failedTestOutputLines:]
+	}
+	if skipped > 0 {
+		fmt.Fprintf(s.out, "    ... %d earlier output lines omitted ...\n", skipped)
+	}
+	for _, l := range lines {
+		fmt.Fprintf(s.out, "    %s\n", l)
+	}
+}
+
+func (s *summarizer) handlePackageResult(ev event) {
+	elapsed := time.Duration(ev.Elapsed * float64(time.Second))
+	s.packages = append(s.packages, packageResult{pkg: ev.Package, action: ev.Action, elapsed: elapsed})
+
+	label := "ok  "
+	switch ev.Action {
+	case "fail":
+		label = "FAIL"
+	case "skip":
+		label = "skip"
+	}
+	fmt.Fprintf(s.out, "%s %s %s\n", label, shortPkg(ev.Package), elapsed.Round(time.Millisecond))
+
+	if ev.Action == "fail" {
+		s.printUnfinished(ev.Package)
+	}
+	if ev.Action == "fail" && len(s.panicHead) > 0 {
+		fmt.Fprintf(s.out, "\n==== panic in %s (first %d lines) ====\n", shortPkg(ev.Package), len(s.panicHead))
+		for _, l := range s.panicHead {
+			fmt.Fprintln(s.out, l)
+		}
+		fmt.Fprintln(s.out, "==== end of panic head ====")
+		fmt.Fprintln(s.out)
+		s.panicHead = nil
+		s.panicking = false
+	}
+}
+
+// printUnfinished names the tests of a failed package that never reported a
+// result, which is what a timeout leaves behind, and shows their last output.
+func (s *summarizer) printUnfinished(pkg string) {
+	var keys []testKey
+	for key := range s.output {
+		if key.pkg == pkg && key.name != "" {
+			keys = append(keys, key)
+		}
+	}
+	if len(keys) == 0 {
+		return
+	}
+	sort.Slice(keys, func(i, j int) bool { return keys[i].name < keys[j].name })
+	fmt.Fprintf(s.out, "\n==== tests in %s that did not finish (%d) ====\n", shortPkg(pkg), len(keys))
+	for _, key := range keys {
+		fmt.Fprintf(s.out, "--- UNFINISHED: %s.%s\n", shortPkg(key.pkg), key.name)
+		s.printTestOutput(key)
+		delete(s.output, key)
+		delete(s.dropped, key)
+	}
+}
+
+func (s *summarizer) printSummary(slowest int) {
+	fmt.Fprintln(s.out)
+	fmt.Fprintln(s.out, "==== package durations ====")
+	sort.Slice(s.packages, func(i, j int) bool { return s.packages[i].elapsed > s.packages[j].elapsed })
+	for _, p := range s.packages {
+		fmt.Fprintf(s.out, "%9s  %-4s  %s\n", p.elapsed.Round(time.Millisecond), p.action, shortPkg(p.pkg))
+	}
+
+	var failed []testResult
+	for _, t := range s.tests {
+		if t.action == "fail" {
+			failed = append(failed, t)
+		}
+	}
+	if len(failed) > 0 {
+		fmt.Fprintln(s.out)
+		fmt.Fprintf(s.out, "==== failed tests (%d) ====\n", len(failed))
+		for _, t := range failed {
+			fmt.Fprintf(s.out, "%9s  %s.%s\n", t.elapsed.Round(time.Millisecond), shortPkg(t.pkg), t.name)
+		}
+	}
+
+	s.printSlowest("slowest top-level tests", slowest, func(t testResult) bool { return !strings.Contains(t.name, "/") })
+	s.printSlowest("slowest subtests", slowest/2, func(t testResult) bool { return strings.Contains(t.name, "/") })
+}
+
+func (s *summarizer) printSlowest(title string, limit int, keep func(testResult) bool) {
+	var tests []testResult
+	for _, t := range s.tests {
+		if keep(t) {
+			tests = append(tests, t)
+		}
+	}
+	if len(tests) == 0 || limit <= 0 {
+		return
+	}
+	sort.Slice(tests, func(i, j int) bool { return tests[i].elapsed > tests[j].elapsed })
+	if len(tests) > limit {
+		tests = tests[:limit]
+	}
+
+	fmt.Fprintln(s.out)
+	fmt.Fprintf(s.out, "==== %s (%d) ====\n", title, len(tests))
+	for _, t := range tests {
+		line := fmt.Sprintf("%9s  %-4s  %s.%s", t.elapsed.Round(time.Millisecond), t.action, shortPkg(t.pkg), t.name)
+		if t.storeCount > 0 {
+			line += fmt.Sprintf("  [stores: %d, %s]", t.storeCount, t.storeTotal.Round(time.Millisecond))
+		}
+		fmt.Fprintln(s.out, line)
+	}
+}
+
+func shortPkg(pkg string) string {
+	return strings.TrimPrefix(pkg, modulePrefix)
+}