diff --git a/agent/internal/sync/sync.go b/agent/internal/sync/sync.go index 1db2fdd..de06199 100644 --- a/agent/internal/sync/sync.go +++ b/agent/internal/sync/sync.go @@ -118,9 +118,21 @@ func poll(client *grpcclient.Client, cfg *config.Config, version string) error { return nil } +// How long a command stream must survive before it counts as having worked. +// Past this, the next drop is treated as a fresh incident rather than as the +// continuation of a run of failures. +const streamHealthyAfter = time.Minute + func runCommandStream(ctx context.Context, cfg *config.Config) { backoff := time.Second - const maxBackoff = 2 * time.Minute + + // Two minutes was the old ceiling, and it was reached far too easily. The + // command stream is what makes this agent controllable at all: while it is + // down, workflows and console sessions fail as "agent offline" even though + // SyncKeys keeps polling happily and the fleet list still shows the server + // active. A shorter ceiling costs a few reconnect attempts; the old one cost + // two minutes of an agent that looks fine and answers nothing. + const maxBackoff = 30 * time.Second for { select { @@ -129,22 +141,40 @@ func runCommandStream(ctx context.Context, cfg *config.Config) { default: } - if err := connectAndHandleStream(ctx, cfg); err != nil { - if ctx.Err() != nil { - return - } - log.Printf("command stream error: %v, reconnecting in %s", err, backoff) - select { - case <-ctx.Done(): - return - case <-time.After(backoff): - } - if backoff < maxBackoff { - backoff *= 2 - } - } else { + started := time.Now() + err := connectAndHandleStream(ctx, cfg) + if ctx.Err() != nil { + return + } + + // A stream that stayed up is evidence the control plane is reachable, + // whatever ended it. Without this the backoff only ever climbed: + // connectAndHandleStream returns an error on *every* stream end, + // including a healthy one dropped by a routine deploy, so an agent + // pinned itself at the ceiling after a handful of ordinary restarts and + // stayed there for the rest of its life. + if time.Since(started) >= streamHealthyAfter { backoff = time.Second } + + if err != nil { + log.Printf("command stream error: %v, reconnecting in %s", err, backoff) + } else { + log.Printf("command stream closed, reconnecting in %s", backoff) + } + + select { + case <-ctx.Done(): + return + case <-time.After(backoff): + } + + if backoff < maxBackoff { + backoff *= 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + } } } diff --git a/claude.md b/claude.md index a843d26..2e1f035 100644 --- a/claude.md +++ b/claude.md @@ -195,6 +195,25 @@ marker is written and the rest is dropped. Without that cap a `yes` in a step is a database incident. **Nothing writes to `/data` any more**, which is why `server.persistence` now defaults to off and `VANTAGE_WORKFLOW_LOG_DIR` is gone. +**Shutdown order is load-bearing.** `main` traps SIGTERM, stops gRPC +(`GracefulStop`, 10s cap) and only then drains HTTP. Each `CommandStream` +handler releases its agent's presence claim on return, so a killed process +leaves `vantage:agent:` behind for the rest of its 30s TTL — during +which other replicas dispatch to a pod that has exited and the caller sees +`agent offline` for a perfectly healthy agent. Draining HTTP first would hold +those claims for the length of the drain, which is why gRPC goes first. The +chart's `server.terminationGracePeriodSeconds` (30s) must stay above the +10s + 10s the stop sequence needs, or the kubelet SIGKILLs mid-shutdown and the +handling buys nothing. + +The agent side of the same failure: `runCommandStream` resets its backoff only +after a stream that survived `streamHealthyAfter`. `connectAndHandleStream` +returns an error on *every* stream end, healthy ones included, so without that +reset the backoff only ever climbed — an agent pinned itself at the ceiling +after a handful of ordinary deploys and stayed there. The ceiling is 30s, not +minutes, because while the stream is down the agent still polls `SyncKeys` and +still reads as `active` in the fleet list while answering no commands at all. + **The leader lock is not an optimisation.** N replicas each running the monitor scheduler means each check fires N times, each incident notification reaches the customer N times, and each hourly rollup is written N times; N reapers race to diff --git a/deploy/chart/vantage/Chart.yaml b/deploy/chart/vantage/Chart.yaml index b8f4a75..4c70144 100644 --- a/deploy/chart/vantage/Chart.yaml +++ b/deploy/chart/vantage/Chart.yaml @@ -2,5 +2,5 @@ apiVersion: v2 name: vantage description: Helm chart for the Vantage stack (Redis, MongoDB, guacd, server, web) type: application -version: 1.0.6 -appVersion: "1.0.6" +version: 1.0.7 +appVersion: "1.0.7" diff --git a/deploy/chart/vantage/templates/server.yaml b/deploy/chart/vantage/templates/server.yaml index 74fe4c4..7c55814 100644 --- a/deploy/chart/vantage/templates/server.yaml +++ b/deploy/chart/vantage/templates/server.yaml @@ -48,6 +48,13 @@ spec: app.kubernetes.io/instance: {{ .Release.Name }} app.kubernetes.io/component: server spec: + # The server stops gRPC before draining HTTP, so that every CommandStream + # handler returns and releases its agent's presence claim. A claim left + # behind outlives the pod for its 30s TTL, and during that window other + # replicas dispatch commands to a process that has exited — surfacing to + # the operator as "agent offline" on an agent that is perfectly healthy. + # 10s for gRPC plus 10s for the HTTP drain, with headroom. + terminationGracePeriodSeconds: {{ .Values.server.terminationGracePeriodSeconds }} {{- if .Values.imagePullSecrets }} imagePullSecrets: {{- toYaml .Values.imagePullSecrets | nindent 8 }} diff --git a/deploy/chart/vantage/values.yaml b/deploy/chart/vantage/values.yaml index 44f6992..2cd258b 100644 --- a/deploy/chart/vantage/values.yaml +++ b/deploy/chart/vantage/values.yaml @@ -41,6 +41,10 @@ guacd: server: replicaCount: 1 + # Must exceed the server's own stop sequence (10s gRPC GracefulStop + 10s + # HTTP drain) or the kubelet SIGKILLs mid-shutdown, which is exactly the + # abrupt exit that leaves agent presence claims stranded in Redis. + terminationGracePeriodSeconds: 30 migrationJob: enabled: true backoffLimit: 0 diff --git a/server/cmd/main.go b/server/cmd/main.go index 23da9e1..ee7266e 100644 --- a/server/cmd/main.go +++ b/server/cmd/main.go @@ -2,9 +2,13 @@ package main import ( "context" + "errors" "log" + "net/http" "os" + "os/signal" "strings" + "syscall" "time" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/api" @@ -141,13 +145,15 @@ func serve() { } log.Printf("message bus ready as node %s", bus.NodeID()) - ctx := context.Background() + // Cancelled on SIGTERM/SIGINT. Everything below that takes a context — the + // housekeeping jobs, the leader lock — stops when the pod is asked to. + ctx, shutdown := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer shutdown() - go func() { - if err := grpcserver.StartGRPC(9090); err != nil { - log.Fatalf("gRPC server error: %v", err) - } - }() + stopGRPC, err := grpcserver.StartGRPC(9090) + if err != nil { + log.Fatalf("gRPC server error: %v", err) + } // Everything below runs on exactly one replica at a time. // @@ -183,12 +189,37 @@ func serve() { r.Use(corsMiddleware()) api.RegisterRoutes(r) - log.Println("REST server listening on :8080") - if err := r.Run(":8080"); err != nil { - log.Fatalf("REST server error: %v", err) + srv := &http.Server{Addr: ":8080", Handler: r} + go func() { + log.Println("REST server listening on :8080") + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Fatalf("REST server error: %v", err) + } + }() + + <-ctx.Done() + log.Println("shutdown signal received") + + // gRPC first, and this ordering is the point of the whole exercise. Stopping + // it runs each CommandStream handler's deferred release, which clears that + // agent's presence claim; until that happens another replica will keep + // dispatching commands to this process. Draining HTTP first would leave the + // claims held for the length of the drain. + stopGRPC() + + drainCtx, cancelDrain := context.WithTimeout(context.Background(), httpDrainTimeout) + defer cancelDrain() + if err := srv.Shutdown(drainCtx); err != nil { + log.Printf("REST server shutdown: %v", err) } + log.Println("shutdown complete") } +// How long in-flight REST requests are given to finish. Console tunnels are +// long-lived WebSockets that will not end on their own, so this is a ceiling +// rather than a target; the relays behind them are already gone by this point. +const httpDrainTimeout = 10 * time.Second + func corsMiddleware() gin.HandlerFunc { return func(c *gin.Context) { c.Header("Access-Control-Allow-Origin", "*") diff --git a/server/internal/grpc/server.go b/server/internal/grpc/server.go index 8974f33..8a3d123 100644 --- a/server/internal/grpc/server.go +++ b/server/internal/grpc/server.go @@ -228,10 +228,19 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err } } -func StartGRPC(port int) error { +// StartGRPC serves the agent API until stop is called. +// +// It returns a stop function rather than serving forever because an abrupt exit +// is not a neutral act here: every CommandStream handler holds an agent's +// presence claim, released by a deferred call that a killed process never runs. +// The claim then outlives its owner for the remainder of its 30s TTL, during +// which dispatch believes the agent is reachable, publishes to a channel with +// no subscriber, and fails as "agent offline" — a pod that has already exited +// still answering for an agent it can no longer reach. +func StartGRPC(port int) (stop func(), err error) { lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port)) if err != nil { - return fmt.Errorf("failed to listen: %w", err) + return nil, fmt.Errorf("failed to listen: %w", err) } s := grpc.NewServer( @@ -248,6 +257,39 @@ func StartGRPC(port int) error { ) pb.RegisterVantageServer(s, &vantageServer{}) - log.Printf("gRPC server listening on :%d", port) - return s.Serve(lis) + go func() { + log.Printf("gRPC server listening on :%d", port) + if err := s.Serve(lis); err != nil { + log.Fatalf("gRPC server error: %v", err) + } + }() + + // GracefulStop sends GOAWAY and waits for the handlers to return, which is + // what runs those deferred releases and, on the agent's side, ends the + // stream with a clean error it reconnects from immediately rather than + // waiting out a TCP timeout. + // + // It is bounded: an idle CommandStream returns as soon as its context is + // cancelled, but a console relay mid-transfer would otherwise hold the + // process past the pod's grace period and earn a SIGKILL — which is the + // abrupt exit this exists to avoid. + return func() { + done := make(chan struct{}) + go func() { + s.GracefulStop() + close(done) + }() + select { + case <-done: + log.Println("gRPC server stopped gracefully") + case <-time.After(grpcStopTimeout): + log.Printf("gRPC server did not stop within %s, forcing", grpcStopTimeout) + s.Stop() + } + }, nil } + +// How long GracefulStop is given before outstanding streams are cut. Comfortably +// inside the chart's termination grace period, so the forced stop below still +// leaves time for the HTTP server to drain. +const grpcStopTimeout = 10 * time.Second