Copilot commented on code in PR #1163:
URL:
https://github.com/apache/skywalking-banyandb/pull/1163#discussion_r3370381871
##########
fodc/agent/internal/proxy/client.go:
##########
@@ -1348,38 +1355,83 @@ func (c *Client) reconnect(ctx context.Context) {
c.streamsMu.Unlock()
}
+ // Each stream (re)start below can fail transiently when the new proxy
pod is not yet accepting
+ // RPCs. reconnect() has already torn down the heartbeat ticker and all
stream receive-loops -
+ // the only mechanisms that would otherwise re-trigger a reconnect - so
returning here without
+ // rescheduling leaves the agent permanently disconnected
(agents_online drops to 0 until a
+ // manual pod restart). Always schedule another attempt on failure.
if regErr := c.StartRegistrationStream(ctx); regErr != nil {
- c.logger.Error().Err(regErr).Msg("Failed to restart
registration stream")
+ c.logger.Error().Err(regErr).Msg("Failed to restart
registration stream; will retry")
+ c.scheduleReconnect(ctx)
return
}
if metricsErr := c.StartMetricsStream(ctx); metricsErr != nil {
- c.logger.Error().Err(metricsErr).Msg("Failed to restart metrics
stream")
+ c.logger.Error().Err(metricsErr).Msg("Failed to restart metrics
stream; will retry")
+ c.scheduleReconnect(ctx)
return
}
if clusterStateErr := c.StartClusterStateStream(ctx); clusterStateErr
!= nil {
- c.logger.Error().Err(clusterStateErr).Msg("Failed to restart
cluster state stream")
+ c.logger.Error().Err(clusterStateErr).Msg("Failed to restart
cluster state stream; will retry")
+ c.scheduleReconnect(ctx)
return
}
if c.lifecycleCollector != nil {
if lifecycleErr := c.StartLifecycleStream(ctx); lifecycleErr !=
nil {
- c.logger.Error().Err(lifecycleErr).Msg("Failed to
restart lifecycle stream")
+ c.logger.Error().Err(lifecycleErr).Msg("Failed to
restart lifecycle stream; will retry")
+ c.scheduleReconnect(ctx)
return
}
}
if c.collectionLister != nil {
if crashErr := c.StartCrashStream(ctx); crashErr != nil {
- c.logger.Error().Err(crashErr).Msg("Failed to restart
crash diagnostics stream")
+ c.logger.Error().Err(crashErr).Msg("Failed to restart
crash diagnostics stream; will retry")
+ c.scheduleReconnect(ctx)
return
}
}
c.logger.Info().Msg("Successfully reconnected to Proxy")
}
+// scheduleReconnect arms a delayed reconnect attempt after a transient
failure during reconnect
+// (e.g. the proxy pod not yet ready), so the agent keeps retrying instead of
stalling forever.
+// reconnect() is single-flight and re-checks ctx and c.disconnected, so a
timer that fires after
+// shutdown is a harmless no-op. Each failed reconnect arms exactly one timer,
so retries proceed at
+// roughly one per interval rather than in a tight loop.
+func (c *Client) scheduleReconnect(ctx context.Context) {
+ c.streamsMu.Lock()
+ disconnected := c.disconnected
+ interval := c.reconnectInterval
+ c.streamsMu.Unlock()
+
+ if disconnected {
+ return
+ }
+ if interval <= 0 {
+ interval = defaultReconnectRetryInterval
+ }
+
+ select {
+ case <-ctx.Done():
+ return
+ default:
+ }
+
+ reconnectFn := c.reconnect
+ if c.reconnectFn != nil {
+ reconnectFn = c.reconnectFn
+ }
+
+ c.logger.Info().Dur("retry_in", interval).Msg("Scheduling proxy
reconnect retry")
+ time.AfterFunc(interval, func() {
+ reconnectFn(ctx)
+ })
Review Comment:
scheduleReconnect uses time.AfterFunc to invoke reconnectFn directly, which
bypasses the usual recovered goroutine launcher used elsewhere in this client
(run.Go) for reconnection work. If reconnect panics, it would crash the process
without generating the expected panic diagnostics/recovery logs. Consider
launching the reconnect attempt via run.Go from the timer callback for
consistency and safety.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
