diff --git a/agent/internal/sync/sync.go b/agent/internal/sync/sync.go index 9ba6001..44b845f 100644 --- a/agent/internal/sync/sync.go +++ b/agent/internal/sync/sync.go @@ -130,6 +130,12 @@ const streamHealthyAfter = time.Minute const ( streamStaleAfter = 70 * time.Second streamStaleCheck = 10 * time.Second + + // How often a healthy stream reports itself. Also the interval at which an + // agent talking to a control plane too old to send heartbeats says so — + // that agent is running without a watchdog, and the journal should not be + // silent about it. + pingSummaryInterval = 5 * time.Minute ) func runCommandStream(ctx context.Context, cfg *config.Config) { @@ -166,10 +172,15 @@ func runCommandStream(ctx context.Context, cfg *config.Config) { backoff = time.Second } + // The uptime is in the line because it is what distinguishes a stream + // that never worked from one that ran for hours and was dropped by a + // deploy — and it is the same measure that decides whether the backoff + // resets, so a reader can see why the delay is what it is. + up := time.Since(started).Truncate(time.Second) if err != nil { - log.Printf("command stream error: %v, reconnecting in %s", err, backoff) + log.Printf("command stream error after %s: %v, reconnecting in %s", up, err, backoff) } else { - log.Printf("command stream closed, reconnecting in %s", backoff) + log.Printf("command stream closed after %s, reconnecting in %s", up, backoff) } select { @@ -213,7 +224,7 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error { return fmt.Errorf("send auth: %w", err) } - log.Println("command stream connected") + log.Printf("command stream connected to %s", cfg.ServerURL) var sendMu sync.Mutex send := func(msg *pb.AgentMessage) error { @@ -238,12 +249,20 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error { lastMu sync.Mutex lastRecv = time.Now() pinged bool + beats int ) markRecv := func(isPing bool) { lastMu.Lock() lastRecv = time.Now() if isPing { - pinged = true + beats++ + // Logged once per stream, because it is the moment the agent starts + // holding the control plane to account: before this the watchdog is + // disarmed and a dead stream would go unnoticed indefinitely. + if !pinged { + pinged = true + log.Printf("command stream heartbeat detected, watchdog armed (%s threshold)", streamStaleAfter) + } } lastMu.Unlock() } @@ -251,16 +270,37 @@ func connectAndHandleStream(ctx context.Context, cfg *config.Config) error { go func() { t := time.NewTicker(streamStaleCheck) defer t.Stop() + + // Reported periodically rather than per beat: at one every 20s the + // journal would be nothing else. The count is what makes a partial + // failure visible — beats arriving but fewer than expected is a + // different problem from beats stopping altogether. + summary := time.NewTicker(pingSummaryInterval) + defer summary.Stop() + for { select { case <-streamCtx.Done(): return + case <-summary.C: + lastMu.Lock() + n, armed := beats, pinged + beats = 0 + lastMu.Unlock() + if armed { + log.Printf("command stream healthy, %d heartbeats in the last %s", n, pingSummaryInterval) + } else { + log.Printf("command stream up but sending no heartbeats; "+ + "control plane predates them, watchdog stays disarmed (last message %s ago)", + time.Since(lastRecv).Truncate(time.Second)) + } case <-t.C: lastMu.Lock() idle, armed := time.Since(lastRecv), pinged lastMu.Unlock() if armed && idle > streamStaleAfter { - log.Printf("command stream silent for %s, assuming it is dead", idle.Truncate(time.Second)) + log.Printf("command stream silent for %s (threshold %s), assuming it is dead and reconnecting", + idle.Truncate(time.Second), streamStaleAfter) abandon() return } diff --git a/server/internal/grpc/server.go b/server/internal/grpc/server.go index 23c29f1..207873f 100644 --- a/server/internal/grpc/server.go +++ b/server/internal/grpc/server.go @@ -221,18 +221,37 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err ping := time.NewTicker(pingInterval) defer ping.Stop() + // Beats are counted and reported periodically rather than logged one by + // one: at one every 20s per agent, a fleet of any size would drown every + // other line in the log. What is worth a line of its own is the first beat + // (it tells the operator this stream's watchdog is now armed on the agent + // side) and any failure to send one. + var beats int + summary := time.NewTicker(pingSummaryInterval) + defer summary.Stop() + ctx := stream.Context() for { select { case <-ctx.Done(): return nil + case <-summary.C: + log.Printf("agent %s command stream healthy, %d beats in the last %s", + srv.ServerID, beats, pingSummaryInterval) + beats = 0 case <-ping.C: // A failed send is the point: it is how this side learns the stream // is gone, which runs the deferred release and frees the agent's // presence claim for whichever pod it reconnects to. if err := stream.Send(&pb.ServerCommand{Ping: &pb.PingCmd{}}); err != nil { + log.Printf("agent %s command stream beat failed after %d beats: %v", + srv.ServerID, beats, err) return err } + beats++ + if beats == 1 { + log.Printf("agent %s command stream beating every %s", srv.ServerID, pingInterval) + } case cmd, ok := <-ch: if !ok { return nil @@ -249,6 +268,11 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err // reconnect. const pingInterval = 20 * time.Second +// How often an otherwise silent healthy stream says so. Long enough that a +// large fleet does not fill the log, short enough that "this pod is still +// serving that agent" is answerable from the log rather than by inference. +const pingSummaryInterval = 5 * time.Minute + // StartGRPC serves the agent API until stop is called. // // It returns a stop function rather than serving forever because an abrupt exit