feat: More logging for command stream

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
 				}

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