1// Package watchdog writes every goroutine stack to a file when a probed
2// owner loop stops answering, so a frozen session leaves the exact place
3// it is stuck on disk.
4package watchdog
5
6import (
7 "context"
8 "fmt"
9 "os"
10 "path/filepath"
11 "runtime/pprof"
12 "sync"
13 "time"
14
15 "github.com/bnema/zerowrap"
16)
17
18// Probe asks an owner loop to run an empty task: Ping returns once the
19// loop ran it. A loop that is stuck never returns.
20type Probe struct {
21 Name string
22 Ping func()
23}
24
25// Options set how often the probes run and how long a loop may take.
26type Options struct {
27 Dir string // where stall dumps go (the run log directory)
28 Every time.Duration // between two probes of a loop
29 Timeout time.Duration // a probe slower than this is a stall
30 Log zerowrap.Logger
31}
32
33// Run probes each loop probes returns until ctx ends; the set may change
34// between ticks (outputs come and go). On a stall it writes one dump per
35// stall: a loop must answer again before it is dumped again. Every Ping
36// must return once ctx ends: Run waits for the probes in flight.
37func Run(ctx context.Context, probes func() []Probe, o Options) {
38 var inflight sync.WaitGroup
39 defer inflight.Wait()
40 type state struct {
41 answer chan struct{} // the probe in flight, nil when none
42 since time.Time
43 stalled bool
44 seen bool
45 }
46 states := map[string]*state{}
47 tick := time.NewTicker(o.Every)
48 defer tick.Stop()
49 for {
50 select {
51 case <-ctx.Done():
52 return
53 case now := <-tick.C:
54 for _, s := range states {
55 s.seen = false
56 }
57 for _, p := range probes() {
58 s := states[p.Name]
59 if s == nil {
60 s = &state{}
61 states[p.Name] = s
62 }
63 s.seen = true
64 if s.answer != nil {
65 select {
66 case <-s.answer:
67 if s.stalled {
68 o.Log.Warn().Str("loop", p.Name).Dur("after", now.Sub(s.since)).Msg("loop answers again")
69 }
70 s.answer, s.stalled = nil, false
71 default:
72 if !s.stalled && now.Sub(s.since) >= o.Timeout {
73 s.stalled = true
74 path, err := dump(o.Dir, p.Name, now)
75 o.Log.Error().Err(err).Str("loop", p.Name).Dur("waited", now.Sub(s.since)).Str("dump", path).Msg("loop stalled")
76 }
77 continue // one probe in flight per loop
78 }
79 }
80 answer := make(chan struct{})
81 s.answer, s.since = answer, now
82 inflight.Go(func() { p.Ping(); close(answer) })
83 }
84 // A loop that left (an unplugged output) is forgotten; its
85 // probe goroutine ends with the loop's context.
86 for name, s := range states {
87 if !s.seen {
88 delete(states, name)
89 }
90 }
91 }
92 }
93}
94
95// dump writes every goroutine stack to dir and syncs it: a frozen machine
96// is often force rebooted, which loses what the page cache still holds.
97func dump(dir, name string, now time.Time) (string, error) {
98 path := filepath.Join(dir, fmt.Sprintf("stall-%s-%s.txt", now.Format("20060102-150405.000"), name))
99 f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
100 if err != nil {
101 return path, err
102 }
103 fmt.Fprintf(f, "loop %q stopped answering at %s\n\n", name, now.Format(time.RFC3339Nano))
104 werr := pprof.Lookup("goroutine").WriteTo(f, 2)
105 serr := f.Sync()
106 cerr := f.Close()
107 if d, err := os.Open(dir); err == nil {
108 _ = d.Sync()
109 _ = d.Close()
110 }
111 for _, err := range []error{werr, serr, cerr} {
112 if err != nil {
113 return path, err
114 }
115 }
116 return path, nil
117}