This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 65d951f1c fix(fodc): keep retrying agent registration after proxy
restart (#1163)
65d951f1c is described below
commit 65d951f1c464bd85770c51b4d69a0a28ed861a1e
Author: Gao Hongtao <[email protected]>
AuthorDate: Mon Jun 8 16:39:49 2026 +0800
fix(fodc): keep retrying agent registration after proxy restart (#1163)
---
fodc/agent/internal/proxy/client.go | 70 ++++++++++++++++---
.../internal/proxy/reconnect_schedule_test.go | 79 ++++++++++++++++++++++
2 files changed, 140 insertions(+), 9 deletions(-)
diff --git a/fodc/agent/internal/proxy/client.go
b/fodc/agent/internal/proxy/client.go
index b18305caa..855194aca 100644
--- a/fodc/agent/internal/proxy/client.go
+++ b/fodc/agent/internal/proxy/client.go
@@ -44,6 +44,11 @@ import (
"github.com/apache/skywalking-banyandb/pkg/run"
)
+// defaultReconnectRetryInterval is the fallback delay scheduleReconnect uses
when the client was
+// created without a positive reconnect interval. In practice
c.reconnectInterval is always set from
+// the --reconnect-interval flag; this only guards a zero value.
+const defaultReconnectRetryInterval = 10 * time.Second
+
// MetricsRequestFilter defines filters for metrics requests.
type MetricsRequestFilter struct {
StartTime *time.Time
@@ -68,6 +73,7 @@ type Client struct {
lifecycleStream fodcv1.FODCService_StreamLifecycleClient
crashStream fodcv1.FODCService_StreamCrashDiagnosticsClient
client fodcv1.FODCServiceClient
+ reconnectFn func(context.Context) // overridable in tests; nil
means use reconnect
proxyAddr string
nodeRole string
@@ -1335,10 +1341,11 @@ func (c *Client) reconnect(ctx context.Context) {
}
if connResult.err != nil {
- c.logger.Error().Err(connResult.err).Msg("Failed to reconnect
to Proxy")
- if disconnectErr := c.Disconnect(); disconnectErr != nil {
- c.logger.Warn().Err(disconnectErr).Msg("Failed to
disconnect")
- }
+ // Transient failure (e.g. the proxy pod is mid-rollout). Do
NOT call Disconnect here: that
+ // latches c.disconnected and the guard at the top of reconnect
would suppress every future
+ // attempt. Schedule another try so the agent recovers once the
proxy is ready again.
+ c.logger.Error().Err(connResult.err).Msg("Failed to reconnect
to Proxy; will retry")
+ c.scheduleReconnect(ctx)
return
}
@@ -1348,31 +1355,41 @@ 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
}
}
@@ -1380,6 +1397,41 @@ func (c *Client) reconnect(ctx context.Context) {
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)
+ })
+}
+
// startHeartbeat starts the heartbeat ticker.
func (c *Client) startHeartbeat(ctx context.Context) {
c.streamsMu.Lock()
diff --git a/fodc/agent/internal/proxy/reconnect_schedule_test.go
b/fodc/agent/internal/proxy/reconnect_schedule_test.go
new file mode 100644
index 000000000..a3a49b7eb
--- /dev/null
+++ b/fodc/agent/internal/proxy/reconnect_schedule_test.go
@@ -0,0 +1,79 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package proxy
+
+import (
+ "context"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+)
+
+// TestScheduleReconnect_ArmsRetryWhenHealthy is the regression guard for the
reconnect-stall root
+// cause: a transient stream-establishment failure must arm another reconnect
attempt rather than
+// leaving the agent permanently disconnected.
+func TestScheduleReconnect_ArmsRetryWhenHealthy(t *testing.T) {
+ var calls int32
+ c := &Client{
+ logger: initTestLogger(t),
+ reconnectInterval: 20 * time.Millisecond,
+ reconnectFn: func(context.Context) {
atomic.AddInt32(&calls, 1) },
+ }
+
+ c.scheduleReconnect(context.Background())
+
+ assert.Eventually(t, func() bool {
+ return atomic.LoadInt32(&calls) == 1
+ }, time.Second, 5*time.Millisecond, "scheduleReconnect should arm
exactly one retry")
+}
+
+// TestScheduleReconnect_SkipsWhenDisconnected confirms an intentional
disconnect suppresses retries
+// so a shut-down agent does not keep dialing the proxy.
+func TestScheduleReconnect_SkipsWhenDisconnected(t *testing.T) {
+ var calls int32
+ c := &Client{
+ logger: initTestLogger(t),
+ reconnectInterval: 10 * time.Millisecond,
+ disconnected: true,
+ reconnectFn: func(context.Context) {
atomic.AddInt32(&calls, 1) },
+ }
+
+ c.scheduleReconnect(context.Background())
+
+ time.Sleep(60 * time.Millisecond)
+ assert.Equal(t, int32(0), atomic.LoadInt32(&calls), "no retry should be
armed when disconnected")
+}
+
+// TestScheduleReconnect_SkipsWhenContextDone confirms a canceled context
suppresses retries.
+func TestScheduleReconnect_SkipsWhenContextDone(t *testing.T) {
+ var calls int32
+ c := &Client{
+ logger: initTestLogger(t),
+ reconnectInterval: 10 * time.Millisecond,
+ reconnectFn: func(context.Context) {
atomic.AddInt32(&calls, 1) },
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ c.scheduleReconnect(ctx)
+
+ time.Sleep(60 * time.Millisecond)
+ assert.Equal(t, int32(0), atomic.LoadInt32(&calls), "no retry should be
armed when context is canceled")
+}
