diff --git a/backend/cmd/run.go b/backend/cmd/run.go index be09d6de..3f9f1c3b 100644 --- a/backend/cmd/run.go +++ b/backend/cmd/run.go @@ -35,7 +35,6 @@ import ( "github.com/tracewayapp/traceway/backend/app/synthetics" "github.com/tracewayapp/traceway/backend/static" - "github.com/coreos/go-systemd/v22/daemon" "github.com/gin-gonic/gin" "github.com/joho/godotenv" traceway "go.tracewayapp.com" @@ -83,6 +82,10 @@ func Run(opts ...Option) { } config.Init(cfg) + // Everything below runs before the service reports itself started, and + // migrations alone can outlast systemd's default 90s start timeout. + ready := beginBoot(bootExtendInterval, bootExtendWindow, bootExtendBudget) + if err := services.InitJWT(); err != nil { // A missing signing key is a configuration mistake, not a crash: print // something actionable instead of burying the message under a panic @@ -325,7 +328,7 @@ func Run(opts ...Option) { }(listener) } - notifySystemd() + ready() config.Logln("Starting server on " + listeners[0].Addr().String()) serveHTTP(router, listeners[0]) } @@ -505,25 +508,6 @@ func parsePositiveInt(s string, def int) int { return v } -func notifySystemd() { - sent, err := daemon.SdNotify(false, daemon.SdNotifyReady) - if err != nil { - config.Logf("Failed to notify systemd: %v", err) - } else if sent { - config.Logln("Notified systemd that service is ready") - } - - go func() { - defer traceway.Recover() - - ticker := time.NewTicker(15 * time.Second) - defer ticker.Stop() - for range ticker.C { - daemon.SdNotify(false, daemon.SdNotifyWatchdog) - } - }() -} - func mustSubFS(fsys fs.FS, dir string) fs.FS { sub, err := fs.Sub(fsys, dir) if err != nil { diff --git a/backend/cmd/systemd.go b/backend/cmd/systemd.go new file mode 100644 index 00000000..d2949c12 --- /dev/null +++ b/backend/cmd/systemd.go @@ -0,0 +1,145 @@ +package cmd + +import ( + "fmt" + "os" + "sync" + "time" + + "github.com/coreos/go-systemd/v22/daemon" + "github.com/tracewayapp/traceway/backend/app/config" + traceway "go.tracewayapp.com" +) + +// Boot can outlast systemd's default TimeoutStartSec=90s -- an index-building +// migration is the usual reason, which is why the Helm chart budgets 300s for +// its startup probe. Rather than ask operators to guess a TimeoutStartSec for +// their largest migration, hold the deadline open while boot runs: every +// EXTEND_TIMEOUT_USEC message pushes it out from the moment systemd receives +// it. +// +// The window is several times the interval so that one late heartbeat -- a +// stalled disk, a stop-the-world pause -- cannot let the deadline lapse. +// +// The total is bounded because these messages go out on a timer rather than on +// boot progress: without a budget, a boot wedged against an unreachable +// database would hold the deadline open forever, which deletes +// TimeoutStartSec instead of deferring it. Once the budget is spent, systemd's +// own timeout applies again. +const ( + bootExtendInterval = 10 * time.Second + bootExtendWindow = 60 * time.Second + bootExtendBudget = 30 * time.Minute +) + +// beginBoot holds the service manager's startup timeout open, and returns the +// function that ends the boot phase by reporting the service ready. +// +// Reporting ready is the same call so that no caller can leave a heartbeat +// running past it: once the service is running, EXTEND_TIMEOUT_USEC extends +// the *watchdog* deadline rather than the startup one, where a stray message +// would paper over a hung process. +// +// All of it is a no-op when the process was not started by a service manager +// (NOTIFY_SOCKET unset), which is every other way Traceway runs: Docker, +// Kubernetes, go run. +func beginBoot(interval, window, budget time.Duration) (ready func()) { + stop := startBootHeartbeat(interval, window, budget) + + return sync.OnceFunc(func() { + stop() + notifyReady() + }) +} + +func startBootHeartbeat(interval, window, budget time.Duration) (stop func()) { + // Ask the environment rather than infer it from the first send: SdNotify + // also reports "not sent" for a transient socket error, and treating that + // as "no service manager" would abandon the heartbeat for the whole boot + // and hand the process back to the 90s default this exists to survive. + if os.Getenv("NOTIFY_SOCKET") == "" { + return func() {} + } + + extendStartTimeout(window) + + stopped := make(chan struct{}) + done := make(chan struct{}) + + go func() { + defer traceway.Recover() + defer close(done) + + ticker := time.NewTicker(interval) + defer ticker.Stop() + spent := time.After(budget) + + for { + select { + case <-ticker.C: + extendStartTimeout(window) + case <-spent: + config.Logf("No longer extending the systemd startup timeout: boot has taken %s", budget) + return + case <-stopped: + return + } + } + }() + + return sync.OnceFunc(func() { + close(stopped) + <-done + }) +} + +func extendStartTimeout(window time.Duration) { + if _, err := daemon.SdNotify(false, fmt.Sprintf("EXTEND_TIMEOUT_USEC=%d", window.Microseconds())); err != nil { + config.Logf("Failed to extend the systemd startup timeout: %v", err) + } +} + +func notifyReady() { + sent, err := daemon.SdNotify(false, daemon.SdNotifyReady) + if err != nil { + config.Logf("Failed to notify systemd: %v", err) + } else if sent { + config.Logln("Notified systemd that service is ready") + } + + startWatchdog() +} + +// startWatchdog pings the watchdog at the interval systemd asked for rather +// than a fixed one: WatchdogSec=10s expects a ping every 5s, and the hardcoded +// 15s this replaces would have had systemd kill a perfectly healthy process. +func startWatchdog() { + ping, err := watchdogPingInterval() + if err != nil { + config.Logf("Failed to read the systemd watchdog interval: %v", err) + return + } + if ping <= 0 { + return + } + + config.Logf("Pinging the systemd watchdog every %s", ping) + + go func() { + defer traceway.Recover() + + ticker := time.NewTicker(ping) + defer ticker.Stop() + for range ticker.C { + daemon.SdNotify(false, daemon.SdNotifyWatchdog) + } + }() +} + +// watchdogPingInterval is half of WATCHDOG_USEC, the rate sd_notify(3) asks +// for. It is 0 when the watchdog is disabled or another process is the watched +// PID, in which case nothing should be sent at all. +func watchdogPingInterval() (time.Duration, error) { + interval, err := daemon.SdWatchdogEnabled(false) + return interval / 2, err +} diff --git a/backend/cmd/systemd_test.go b/backend/cmd/systemd_test.go new file mode 100644 index 00000000..532867f5 --- /dev/null +++ b/backend/cmd/systemd_test.go @@ -0,0 +1,199 @@ +package cmd + +import ( + "fmt" + "net" + "os" + "path/filepath" + "testing" + "time" +) + +const readTimeout = 2 * time.Second + +// notifySocket binds a unixgram socket and points NOTIFY_SOCKET at it, the way +// systemd does for a Type=notify unit. +func notifySocket(t *testing.T) *net.UnixConn { + t.Helper() + + // Keep test names short: the socket path has to fit sun_path's ~107 bytes. + path := filepath.Join(t.TempDir(), "n") + conn, err := net.ListenUnixgram("unixgram", &net.UnixAddr{Name: path, Net: "unixgram"}) + if err != nil { + t.Fatalf("bind %s: %v", path, err) + } + t.Cleanup(func() { conn.Close() }) + + t.Setenv("NOTIFY_SOCKET", path) + return conn +} + +func readMessage(t *testing.T, conn *net.UnixConn) string { + t.Helper() + + message, ok := readMessageWithin(t, conn, readTimeout) + if !ok { + t.Fatalf("no message arrived within %s", readTimeout) + } + return message +} + +func expectNoMessage(t *testing.T, conn *net.UnixConn, within time.Duration) { + t.Helper() + + if message, ok := readMessageWithin(t, conn, within); ok { + t.Fatalf("unexpected message %q", message) + } +} + +func readMessageWithin(t *testing.T, conn *net.UnixConn, within time.Duration) (string, bool) { + t.Helper() + + if err := conn.SetReadDeadline(time.Now().Add(within)); err != nil { + t.Fatalf("set deadline: %v", err) + } + buf := make([]byte, 256) + n, err := conn.Read(buf) + if err != nil { + return "", false + } + return string(buf[:n]), true +} + +func TestBootHeartbeatKeepsExtendingTheStartTimeout(t *testing.T) { + conn := notifySocket(t) + + stop := startBootHeartbeat(20*time.Millisecond, 5*time.Second, time.Minute) + defer stop() + + // The first extension goes out before Run() touches the database, so a boot + // that stalls immediately is still covered. + want := "EXTEND_TIMEOUT_USEC=5000000" + for i := range 3 { + if got := readMessage(t, conn); got != want { + t.Fatalf("message %d = %q, want %q", i, got, want) + } + } +} + +func TestNothingExtendsTheStartTimeoutAfterReady(t *testing.T) { + conn := notifySocket(t) + + ready := beginBoot(20*time.Millisecond, 5*time.Second, time.Minute) + readMessage(t, conn) + ready() + + // Anything after READY=1 extends the *watchdog* deadline instead of the + // startup one, so a heartbeat outliving ready() would hide a hung process. + for { + message := readMessage(t, conn) + if message == "READY=1" { + break + } + if message != "EXTEND_TIMEOUT_USEC=5000000" { + t.Fatalf("unexpected message before READY=1: %q", message) + } + } + expectNoMessage(t, conn, 100*time.Millisecond) + + // A second call must neither re-report ready nor start a second watchdog. + ready() + expectNoMessage(t, conn, 100*time.Millisecond) +} + +func TestBootHeartbeatStopsExtendingOnceTheBudgetIsSpent(t *testing.T) { + conn := notifySocket(t) + + // A boot wedged against an unreachable database must not hold systemd's + // start deadline open forever. + stop := startBootHeartbeat(10*time.Millisecond, 5*time.Second, 50*time.Millisecond) + defer stop() + + readMessage(t, conn) + deadline := time.Now().Add(readTimeout) + for { + if _, ok := readMessageWithin(t, conn, 150*time.Millisecond); !ok { + break + } + if time.Now().After(deadline) { + t.Fatal("heartbeat still extending well past its budget") + } + } +} + +func TestBootHeartbeatSurvivesAFailedFirstSend(t *testing.T) { + // NOTIFY_SOCKET names a socket nothing has bound yet, so the first send + // fails. Reading "no service manager" into that failure would abandon the + // heartbeat and drop boot back to the default TimeoutStartSec. + path := filepath.Join(t.TempDir(), "n") + t.Setenv("NOTIFY_SOCKET", path) + + stop := startBootHeartbeat(20*time.Millisecond, 5*time.Second, time.Minute) + defer stop() + + conn, err := net.ListenUnixgram("unixgram", &net.UnixAddr{Name: path, Net: "unixgram"}) + if err != nil { + t.Fatalf("bind %s: %v", path, err) + } + defer conn.Close() + + if got := readMessage(t, conn); got != "EXTEND_TIMEOUT_USEC=5000000" { + t.Fatalf("got %q, want the heartbeat to have retried", got) + } +} + +func TestBootHeartbeatIsANoOpWithoutAServiceManager(t *testing.T) { + t.Setenv("NOTIFY_SOCKET", "") + + done := make(chan struct{}) + go func() { + defer close(done) + beginBoot(bootExtendInterval, bootExtendWindow, bootExtendBudget)() + }() + + select { + case <-done: + case <-time.After(readTimeout): + t.Fatal("ready() blocked when NOTIFY_SOCKET is unset") + } +} + +func TestWatchdogPingIntervalIsHalfOfWhatSystemdAsksFor(t *testing.T) { + tests := []struct { + name string + usec string + pid string + want time.Duration + wantErr bool + }{ + {name: "disabled", usec: "", want: 0}, + // The interval this replaces was hardcoded at 15s, so systemd would + // have killed a healthy process one ping into a 10s watchdog. + {name: "shorter than the old hardcoded interval", usec: "10000000", want: 5 * time.Second}, + {name: "longer than the old hardcoded interval", usec: "60000000", want: 30 * time.Second}, + {name: "watched pid is this process", usec: "30000000", pid: fmt.Sprint(os.Getpid()), want: 15 * time.Second}, + {name: "watched pid is another process", usec: "30000000", pid: fmt.Sprint(os.Getpid() + 1), want: 0}, + {name: "malformed", usec: "soon", wantErr: true}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Setenv("WATCHDOG_USEC", test.usec) + t.Setenv("WATCHDOG_PID", test.pid) + + got, err := watchdogPingInterval() + if test.wantErr { + if err == nil { + t.Fatalf("got %s, want an error", got) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != test.want { + t.Fatalf("got %s, want %s", got, test.want) + } + }) + } +}