You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/04/30 05:11:25 UTC
(skywalking-banyandb) 21/24: fix(fodc): heal reconnect deadlock and add error message when no agents for lifecycle request (#1101)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch v0.10.x
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 4178a498375b77d0725f0e8fc1651b6b62c8f0d4
Author: mrproliu <74...@qq.com>
AuthorDate: Wed Apr 29 10:38:22 2026 +0800
fix(fodc): heal reconnect deadlock and add error message when no agents for lifecycle request (#1101)
---
fodc/agent/internal/proxy/client.go | 65 ++++++---
.../internal/proxy/client_blocking_send_test.go | 151 +++++++++++++++++++++
fodc/agent/internal/proxy/client_test.go | 13 ++
fodc/agent/internal/proxy/wait.go | 69 ++++++++++
fodc/agent/internal/proxy/wait_test.go | 61 +++++++++
fodc/proxy/internal/api/server.go | 30 +++-
fodc/proxy/internal/api/server_test.go | 52 +++++++
fodc/proxy/internal/lifecycle/manager.go | 112 ++++++++++-----
fodc/proxy/internal/lifecycle/manager_test.go | 31 ++++-
9 files changed, 521 insertions(+), 63 deletions(-)
diff --git a/fodc/agent/internal/proxy/client.go b/fodc/agent/internal/proxy/client.go
index cf7b44203..26752365d 100644
--- a/fodc/agent/internal/proxy/client.go
+++ b/fodc/agent/internal/proxy/client.go
@@ -130,7 +130,9 @@ func (c *Client) Connect(ctx context.Context) error {
c.streamsMu.Unlock()
if wasDisconnected {
- c.heartbeatWg.Wait()
+ if !waitWaitGroupWithTimeout(&c.heartbeatWg, heartbeatWaitTimeout) {
+ c.logger.Warn().Msg("Heartbeat goroutine did not exit in time during Connect, proceeding anyway")
+ }
}
c.streamsMu.Lock()
@@ -693,8 +695,11 @@ func (c *Client) sendFilteredMetrics(
return nil
}
-// SendHeartbeat sends heartbeat to Proxy.
-func (c *Client) SendHeartbeat(_ context.Context) error {
+// SendHeartbeat sends heartbeat to Proxy. The provided context bounds the wait so that a
+// stream whose Send is wedged (for example, after the proxy entered graceful_stop and the
+// underlying TCP write queue is no longer being drained) cannot pin the heartbeat goroutine
+// indefinitely.
+func (c *Client) SendHeartbeat(ctx context.Context) error {
c.streamsMu.RLock()
if c.disconnected || c.registrationStream == nil {
c.streamsMu.RUnlock()
@@ -710,7 +715,23 @@ func (c *Client) SendHeartbeat(_ context.Context) error {
ContainerNames: c.containerNames,
}
- if sendErr := registrationStream.Send(req); sendErr != nil {
+ // gRPC client-stream Send does not accept a context, so wrap it in a goroutine and
+ // race the result against ctx.Done(). When ctx fires first the in-flight Send is
+ // orphaned. The leak is bounded: every reconnect/cleanupStreams/Disconnect path
+ // closes the registration stream after waiting on heartbeatWg with bounded timeout,
+ // which forces Send to return so the orphan goroutine can write to the (buffered)
+ // channel and exit. Without an upstream tear-down the orphan would persist, so the
+ // reliability of the heartbeatWaitTimeout-bounded cleanup paths is load-bearing here.
+ sendErrCh := make(chan error, 1)
+ go func() { sendErrCh <- registrationStream.Send(req) }()
+
+ var sendErr error
+ select {
+ case sendErr = <-sendErrCh:
+ case <-ctx.Done():
+ return fmt.Errorf("heartbeat send aborted: %w", ctx.Err())
+ }
+ if sendErr != nil {
// Check if error is due to stream being closed/disconnected
if errors.Is(sendErr, io.EOF) || errors.Is(sendErr, context.Canceled) {
return fmt.Errorf("registration stream closed")
@@ -738,15 +759,11 @@ func (c *Client) cleanupStreams() {
c.stopCh = make(chan struct{})
c.streamsMu.Unlock()
- if oldStopCh != nil {
- select {
- case <-oldStopCh:
- default:
- close(oldStopCh)
- }
- }
+ safeClose(oldStopCh)
- c.heartbeatWg.Wait()
+ if !waitWaitGroupWithTimeout(&c.heartbeatWg, heartbeatWaitTimeout) {
+ c.logger.Warn().Msg("Heartbeat goroutine did not exit in time during cleanupStreams, proceeding anyway")
+ }
c.streamsMu.Lock()
if c.registrationStream != nil {
@@ -798,8 +815,11 @@ func (c *Client) Disconnect() error {
c.streamsMu.Unlock()
- // Wait for heartbeat goroutine to exit before closing streams
- c.heartbeatWg.Wait()
+ // Wait for heartbeat goroutine to exit before closing streams. Bounded so a wedged
+ // in-flight Send cannot keep Disconnect from progressing.
+ if !waitWaitGroupWithTimeout(&c.heartbeatWg, heartbeatWaitTimeout) {
+ c.logger.Warn().Msg("Heartbeat goroutine did not exit in time during Disconnect, proceeding anyway")
+ }
c.streamsMu.Lock()
if c.registrationStream != nil {
@@ -1017,15 +1037,22 @@ func (c *Client) reconnect(ctx context.Context) {
c.heartbeatTicker = nil
}
+ // Swap the stop channel under the lock so subsequent callers see the new one, then
+ // release the lock and close the old channel from outside the critical section.
+ // This is what allows the heartbeat goroutine to exit via <-stopCh BEFORE we wait on
+ // heartbeatWg; closing stopCh AFTER the wait (the previous behavior) deadlocks any
+ // heartbeat goroutine that is parked on the select.
+ oldStopCh := c.stopCh
+ c.stopCh = make(chan struct{})
c.streamsMu.Unlock()
- c.heartbeatWg.Wait()
- c.streamsMu.Lock()
+ safeClose(oldStopCh)
- if !c.disconnected {
- close(c.stopCh)
+ if !waitWaitGroupWithTimeout(&c.heartbeatWg, heartbeatWaitTimeout) {
+ c.logger.Warn().Msg("Heartbeat goroutine did not exit in time during reconnect, proceeding anyway")
}
- c.stopCh = make(chan struct{})
+
+ c.streamsMu.Lock()
if c.registrationStream != nil {
_ = c.registrationStream.CloseSend()
c.registrationStream = nil
diff --git a/fodc/agent/internal/proxy/client_blocking_send_test.go b/fodc/agent/internal/proxy/client_blocking_send_test.go
new file mode 100644
index 000000000..203739a67
--- /dev/null
+++ b/fodc/agent/internal/proxy/client_blocking_send_test.go
@@ -0,0 +1,151 @@
+// 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"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/fodc/agent/internal/flightrecorder"
+)
+
+// newBlockingMockRegisterAgentClient returns a mock whose Send blocks until blockCh is
+// closed, mimicking a gRPC client stream that received GOAWAY but whose underlying TCP
+// write queue is no longer being drained. The entered channel is closed once the first
+// time Send is called so tests can synchronize with the heartbeat goroutine reaching the
+// blocking call.
+func newBlockingMockRegisterAgentClient(ctx context.Context) *mockRegisterAgentClient {
+ mock := newMockRegisterAgentClient(ctx)
+ mock.sendBlock = make(chan struct{})
+ mock.sendEntered = make(chan struct{})
+ return mock
+}
+
+// TestProxyClient_SendHeartbeat_RespectsCtxOnBlockingSend reproduces the bug where
+// SendHeartbeat discards its ctx parameter. With a registration stream whose Send blocks
+// indefinitely, the caller's context timeout has no effect: SendHeartbeat hangs forever
+// instead of returning when the deadline elapses.
+//
+// Pre-fix behavior: SendHeartbeat blocks far beyond the 100ms deadline; the test reaches
+// its 2s ceiling and calls t.Fatal.
+// Post-fix behavior: SendHeartbeat returns shortly after the 100ms deadline with an error
+// that wraps context.DeadlineExceeded.
+func TestProxyClient_SendHeartbeat_RespectsCtxOnBlockingSend(t *testing.T) {
+ testLogger := initTestLogger(t)
+ fr := flightrecorder.NewFlightRecorder(1000000)
+ pc := NewProxyClient(
+ "localhost:8080", "datanode-hot", "192.168.1.1", []string{"data"},
+ nil, 5*time.Second, 10*time.Second, fr, nil, nil, testLogger,
+ )
+
+ streamCtx, streamCancel := context.WithCancel(context.Background())
+ defer streamCancel()
+ stream := newBlockingMockRegisterAgentClient(streamCtx)
+ defer close(stream.sendBlock)
+
+ pc.streamsMu.Lock()
+ pc.registrationStream = stream
+ pc.streamsMu.Unlock()
+
+ callCtx, callCancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+ defer callCancel()
+
+ done := make(chan error, 1)
+ go func() { done <- pc.SendHeartbeat(callCtx) }()
+
+ select {
+ case err := <-done:
+ require.Error(t, err, "SendHeartbeat must return an error when its ctx times out")
+ assert.Truef(t,
+ errors.Is(err, context.DeadlineExceeded),
+ "expected error wrapping context.DeadlineExceeded, got: %v", err,
+ )
+ case <-time.After(2 * time.Second):
+ t.Fatal("SendHeartbeat ignored its context deadline: blocked Send hangs forever " +
+ "(root cause: SendHeartbeat declares ctx as `_` and never applies it to Send)")
+ }
+}
+
+// TestProxyClient_reconnect_DeadlocksWhenHeartbeatBlockedOnSend reproduces the central bug
+// from the production incident: when a heartbeat goroutine is in the middle of
+// registrationStream.Send and that Send is stuck (e.g. proxy entered graceful_stop and the
+// stream's underlying TCP write queue is no longer being drained), the reconnect path's
+// c.heartbeatWg.Wait() never returns. Combined with the fact that reconnect closes stopCh
+// AFTER Wait rather than before, the heartbeat goroutine has no way out: its Send is stuck,
+// ctx is alive, stopCh is still open, ticker is stopped. The result in production is the
+// agent silently turning into a zombie that no longer talks to the proxy.
+//
+// Pre-fix behavior: pc.reconnect blocks indefinitely; the test reaches its 2s ceiling and
+// calls t.Fatal (heartbeatWaitTimeout is overridden to 200ms below to keep the post-fix
+// assertion fast; without the fix the call still hangs because Wait has no timeout at all).
+// Post-fix behavior: pc.reconnect returns within roughly the overridden heartbeatWaitTimeout.
+func TestProxyClient_reconnect_DeadlocksWhenHeartbeatBlockedOnSend(t *testing.T) {
+ originalTimeout := heartbeatWaitTimeout
+ heartbeatWaitTimeout = 200 * time.Millisecond
+ defer func() { heartbeatWaitTimeout = originalTimeout }()
+
+ testLogger := initTestLogger(t)
+ fr := flightrecorder.NewFlightRecorder(1000000)
+ pc := NewProxyClient(
+ "localhost:8080", "datanode-hot", "192.168.1.1", []string{"data"},
+ nil, 10*time.Millisecond, 100*time.Millisecond, fr, nil, nil, testLogger,
+ )
+
+ bgCtx, bgCancel := context.WithCancel(context.Background())
+ defer bgCancel()
+
+ streamCtx, streamCancel := context.WithCancel(context.Background())
+ defer streamCancel()
+ stream := newBlockingMockRegisterAgentClient(streamCtx)
+ defer close(stream.sendBlock)
+
+ pc.streamsMu.Lock()
+ pc.registrationStream = stream
+ pc.streamsMu.Unlock()
+
+ pc.connManager.start(bgCtx)
+ defer pc.connManager.stop()
+
+ pc.startHeartbeat(bgCtx)
+
+ select {
+ case <-stream.sendEntered:
+ case <-time.After(2 * time.Second):
+ t.Fatal("heartbeat goroutine never entered Send within 2s; cannot reproduce blocking-Send scenario")
+ }
+
+ done := make(chan struct{})
+ go func() {
+ pc.reconnect(bgCtx)
+ close(done)
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("reconnect deadlocked: heartbeatWg.Wait() never returned while a heartbeat " +
+ "goroutine was blocked inside registrationStream.Send (root causes: SendHeartbeat " +
+ "ignores ctx + reconnect waits on heartbeatWg without a bounded timeout + reconnect " +
+ "closes stopCh AFTER Wait instead of before)")
+ }
+}
diff --git a/fodc/agent/internal/proxy/client_test.go b/fodc/agent/internal/proxy/client_test.go
index b37229ef5..0c5346c3d 100644
--- a/fodc/agent/internal/proxy/client_test.go
+++ b/fodc/agent/internal/proxy/client_test.go
@@ -85,13 +85,20 @@ func (m *mockFODCServiceClient) StreamLifecycle(_ context.Context, _ ...grpc.Cal
}
// mockRegisterAgentClient implements fodcv1.FODCService_RegisterAgentClient for testing.
+// When sendBlock is non-nil, Send blocks on it until the channel is closed (used by tests
+// that reproduce the wedged-Send scenario). When sendEntered is non-nil, it is closed once
+// the first time Send is invoked, letting tests synchronize with the goroutine reaching
+// the blocking call.
type mockRegisterAgentClient struct {
ctx context.Context
recvChan chan *fodcv1.RegisterAgentResponse
+ sendBlock chan struct{}
+ sendEntered chan struct{}
sendErr error
recvErr error
sentRequests []*fodcv1.RegisterAgentRequest
recvResponses []*fodcv1.RegisterAgentResponse
+ enterOnce sync.Once
mu sync.RWMutex
}
@@ -105,6 +112,12 @@ func newMockRegisterAgentClient(ctx context.Context) *mockRegisterAgentClient {
}
func (m *mockRegisterAgentClient) Send(req *fodcv1.RegisterAgentRequest) error {
+ if m.sendEntered != nil {
+ m.enterOnce.Do(func() { close(m.sendEntered) })
+ }
+ if m.sendBlock != nil {
+ <-m.sendBlock
+ }
m.mu.Lock()
defer m.mu.Unlock()
if m.sendErr != nil {
diff --git a/fodc/agent/internal/proxy/wait.go b/fodc/agent/internal/proxy/wait.go
new file mode 100644
index 000000000..3e578f36d
--- /dev/null
+++ b/fodc/agent/internal/proxy/wait.go
@@ -0,0 +1,69 @@
+// 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 (
+ "sync"
+ "time"
+)
+
+// heartbeatWaitTimeout bounds how long reconnect, cleanupStreams and Disconnect will wait
+// for the heartbeat goroutine to exit. It is a package-level variable rather than a const
+// so that tests reproducing blocking-Send scenarios can shorten it without dragging real
+// production wait times down. The default of 10s tracks the conn_manager's own 5s heartbeat
+// check timeout plus enough slack for one Send round-trip.
+var heartbeatWaitTimeout = 10 * time.Second
+
+// waitWaitGroupWithTimeout waits for wg to reach zero, but no longer than d. It returns
+// true when wg completed before the timeout fired, false when the timeout fired first.
+//
+// On the timeout branch the inner waiter goroutine remains parked on wg.Wait() until wg
+// eventually reaches zero (sync.WaitGroup offers no cancellation). Callers should ensure
+// that whatever holds the wg counter up will eventually be released; in this package the
+// reconnect / cleanupStreams / Disconnect paths close the registration stream right after
+// calling this helper, which unblocks any Send the heartbeat goroutine is parked in and
+// lets it run to completion.
+func waitWaitGroupWithTimeout(wg *sync.WaitGroup, d time.Duration) bool {
+ done := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(done)
+ }()
+ timer := time.NewTimer(d)
+ defer timer.Stop()
+ select {
+ case <-done:
+ return true
+ case <-timer.C:
+ return false
+ }
+}
+
+// safeClose closes ch if it is non-nil and has not already been closed. The caller is
+// responsible for ensuring no other goroutine closes ch concurrently; this helper only
+// guards against the "already closed by an earlier call from this same path" case.
+func safeClose(ch chan struct{}) {
+ if ch == nil {
+ return
+ }
+ select {
+ case <-ch:
+ default:
+ close(ch)
+ }
+}
diff --git a/fodc/agent/internal/proxy/wait_test.go b/fodc/agent/internal/proxy/wait_test.go
new file mode 100644
index 000000000..9c4ee0c1a
--- /dev/null
+++ b/fodc/agent/internal/proxy/wait_test.go
@@ -0,0 +1,61 @@
+// 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 (
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestWaitWaitGroupWithTimeout_CompletesBeforeTimeout(t *testing.T) {
+ var wg sync.WaitGroup
+ wg.Add(1)
+ go func() {
+ time.Sleep(20 * time.Millisecond)
+ wg.Done()
+ }()
+
+ completed := waitWaitGroupWithTimeout(&wg, 500*time.Millisecond)
+
+ assert.True(t, completed, "expected helper to report completion before the timeout fired")
+}
+
+func TestWaitWaitGroupWithTimeout_TimesOut(t *testing.T) {
+ var wg sync.WaitGroup
+ wg.Add(1)
+ defer wg.Done()
+
+ start := time.Now()
+ completed := waitWaitGroupWithTimeout(&wg, 50*time.Millisecond)
+ elapsed := time.Since(start)
+
+ assert.False(t, completed, "expected helper to report timeout when wg never reaches zero")
+ assert.GreaterOrEqual(t, elapsed, 50*time.Millisecond, "helper returned earlier than the configured timeout")
+ assert.Less(t, elapsed, 500*time.Millisecond, "helper waited significantly past the configured timeout")
+}
+
+func TestWaitWaitGroupWithTimeout_AlreadyDone(t *testing.T) {
+ var wg sync.WaitGroup
+
+ completed := waitWaitGroupWithTimeout(&wg, 100*time.Millisecond)
+
+ assert.True(t, completed, "expected helper to return immediately for an already-zero WaitGroup")
+}
diff --git a/fodc/proxy/internal/api/server.go b/fodc/proxy/internal/api/server.go
index d823752ce..3ff5de920 100644
--- a/fodc/proxy/internal/api/server.go
+++ b/fodc/proxy/internal/api/server.go
@@ -476,6 +476,15 @@ type timeSeriesMetric struct {
}
// handleClusterLifecycle handles GET /cluster/lifecycle endpoint.
+//
+// The handler adds an "error" field to the response body whenever the proxy could not
+// gather any lifecycle data — that is, when no agent was registered (Total == 0), no
+// registered agent supported the lifecycle stream (Requested == 0), or none of the
+// requested agents responded within the collection window (Responded == 0). The error
+// message describes which of these three sub-cases occurred so callers can distinguish an
+// infrastructure-layer outage (e.g. proxy pod restart with every agent failing to
+// reconnect) from a cluster that genuinely has no groups. The HTTP status stays 200 in
+// every case; the distinguishing signal is in the body, not in the status code.
func (s *Server) handleClusterLifecycle(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
@@ -485,10 +494,27 @@ func (s *Server) handleClusterLifecycle(w http.ResponseWriter, r *http.Request)
http.Error(w, "Lifecycle manager not available", http.StatusServiceUnavailable)
return
}
- lifecycleData := s.lifecycleManager.CollectLifecycle(r.Context())
+ lifecycleData, agentSummary := s.lifecycleManager.CollectLifecycle(r.Context())
+
+ body := map[string]interface{}{
+ "groups": lifecycleData.Groups,
+ "lifecycle_statuses": lifecycleData.LifecycleStatuses,
+ "agent_summary": agentSummary,
+ }
+ switch {
+ case agentSummary.Total == 0:
+ body["error"] = "FODC unavailable: no agents registered"
+ case agentSummary.Requested == 0:
+ body["error"] = fmt.Sprintf("FODC unavailable: 0/%d registered agents support the lifecycle stream",
+ agentSummary.Total)
+ case agentSummary.Responded == 0:
+ body["error"] = fmt.Sprintf("FODC unavailable: 0/%d requested agents responded",
+ agentSummary.Requested)
+ }
+
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
- if encodeErr := json.NewEncoder(w).Encode(lifecycleData); encodeErr != nil {
+ if encodeErr := json.NewEncoder(w).Encode(body); encodeErr != nil {
s.logger.Error().Err(encodeErr).Msg("Failed to encode lifecycle response")
}
}
diff --git a/fodc/proxy/internal/api/server_test.go b/fodc/proxy/internal/api/server_test.go
index e6182bb73..b39d8c609 100644
--- a/fodc/proxy/internal/api/server_test.go
+++ b/fodc/proxy/internal/api/server_test.go
@@ -781,3 +781,55 @@ func TestHandleClusterLifecycle_NoLifecycleManager(t *testing.T) {
assert.Equal(t, http.StatusServiceUnavailable, resp.Code)
}
+
+// TestHandleClusterLifecycle_NoAgents_FlagsErrorInBody reproduces the bug where the proxy
+// silently returns 200 + an empty payload when no agents are registered. This makes the
+// caller (e.g. the SRE inspection agent) unable to distinguish two semantically very
+// different states:
+// - the BanyanDB cluster genuinely has no groups (a real cluster-side problem); or
+// - the FODC proxy lost all of its agents (an infrastructure-layer outage caused by a
+// proxy pod restart, every agent stuck on reconnect, etc.).
+//
+// In production the second case is what triggered the lifecycle critical alert, but the
+// HTTP layer presented it identically to the first.
+//
+// Pre-fix behavior: handleClusterLifecycle writes 200 with body
+// `{"groups":[],"lifecycle_statuses":[]}` and no way to tell the two cases apart.
+// Post-fix behavior: HTTP status is still 200, but the body now carries an "error" field
+// when agent_summary reports total=0 or responded=0, plus the agent_summary itself.
+func TestHandleClusterLifecycle_NoAgents_FlagsErrorInBody(t *testing.T) {
+ initTestLogger(t)
+ testLogger := logger.GetLogger("test", "api")
+ testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
+ defer testRegistry.Stop()
+ mockSender := &mockRequestSender{}
+ aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)
+
+ mockLifecycleRequester := &mockLifecycleDataRequester{
+ dataByAgent: make(map[string]*fodcv1.LifecycleData),
+ podByAgent: make(map[string]string),
+ }
+ lifecycleMgr := lifecycle.NewManager(testRegistry, mockLifecycleRequester, testLogger)
+ mockLifecycleRequester.lifecycleMgr = lifecycleMgr
+
+ server := NewServer(aggregator, nil, lifecycleMgr, testRegistry, testLogger)
+
+ req := httptest.NewRequest(http.MethodGet, "/cluster/lifecycle", nil)
+ resp := httptest.NewRecorder()
+ server.handleClusterLifecycle(resp, req)
+
+ assert.Equal(t, http.StatusOK, resp.Code, "status code stays 200 even when zero agents are registered")
+
+ var body map[string]interface{}
+ require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
+
+ errMsg, hasError := body["error"]
+ assert.Truef(t, hasError, "body must include an error field when zero agents are registered, got: %v", body)
+ assert.Containsf(t, errMsg, "FODC unavailable",
+ "error field should describe the infrastructure-layer failure, got: %v", errMsg)
+
+ summary, hasSummary := body["agent_summary"].(map[string]interface{})
+ require.Truef(t, hasSummary, "body must include agent_summary, got: %v", body)
+ assert.EqualValues(t, 0, summary["total"], "agent_summary.total should be 0 when no agents are registered")
+ assert.EqualValues(t, 0, summary["responded"], "agent_summary.responded should be 0 when no agents are registered")
+}
diff --git a/fodc/proxy/internal/lifecycle/manager.go b/fodc/proxy/internal/lifecycle/manager.go
index 66cb9ce81..207354730 100644
--- a/fodc/proxy/internal/lifecycle/manager.go
+++ b/fodc/proxy/internal/lifecycle/manager.go
@@ -42,6 +42,18 @@ type InspectionResult struct {
LifecycleStatuses []*PodLifecycleStatus `json:"lifecycle_statuses"`
}
+// AgentSummary describes how many agents the most recent CollectLifecycle invocation saw,
+// requested data from, and actually got data back from. It lets callers tell apart "the
+// cluster has nothing to report" (cluster-side empty) from "the proxy could not reach any
+// agent" (infrastructure-side empty), which look identical when only the InspectionResult
+// is observed.
+type AgentSummary struct {
+ Total int `json:"total"`
+ Requested int `json:"requested"`
+ Responded int `json:"responded"`
+ NotResponded int `json:"not_responded"`
+}
+
func emptyResult() *InspectionResult {
return &InspectionResult{
Groups: make([]*fodcv1.GroupLifecycleInfo, 0),
@@ -106,67 +118,95 @@ func (m *Manager) UpdateLifecycle(agentID, podName string, data *fodcv1.Lifecycl
}
}
-// CollectLifecycle requests and collects lifecycle data from all agents.
-func (m *Manager) CollectLifecycle(ctx context.Context) *InspectionResult {
+func (m *Manager) registerSession(agentID string, collectChs map[string]chan *agentLifecycleData) {
+ collectCh := make(chan *agentLifecycleData, 1)
+ collectChs[agentID] = collectCh
+ m.collectingMu.Lock()
+ m.collecting[agentID] = collectCh
+ m.collectingMu.Unlock()
+}
+
+func (m *Manager) unregisterSession(agentID string, collectChs map[string]chan *agentLifecycleData) {
+ m.collectingMu.Lock()
+ if ch, exists := m.collecting[agentID]; exists {
+ close(ch)
+ delete(m.collecting, agentID)
+ }
+ m.collectingMu.Unlock()
+ delete(collectChs, agentID)
+}
+
+// CollectLifecycle requests and collects lifecycle data from all agents and returns
+// both the aggregated result and the agent summary captured atomically during the same
+// invocation. Returning the summary as a second value (rather than via a separate
+// "LastSummary" accessor) avoids a read-after-write race between the result and the
+// summary when concurrent HTTP requests trigger overlapping collections.
+func (m *Manager) CollectLifecycle(ctx context.Context) (*InspectionResult, AgentSummary) {
m.collectingOp.Lock()
defer m.collectingOp.Unlock()
- if m.registry == nil {
- m.log.Info().Msg("CollectLifecycle: registry is nil, returning empty")
- return emptyResult()
+ summary := AgentSummary{}
+
+ if m.registry == nil || m.grpcService == nil {
+ m.log.Info().Msg("CollectLifecycle: registry or grpcService is nil, returning empty")
+ return emptyResult(), summary
}
agents := m.registry.ListAgents()
+ summary.Total = len(agents)
if len(agents) == 0 {
m.log.Info().Msg("CollectLifecycle: no agents registered, returning empty")
- return emptyResult()
+ return emptyResult(), summary
}
m.log.Info().Int("agent_count", len(agents)).Msg("CollectLifecycle: starting collection")
collectChs := make(map[string]chan *agentLifecycleData)
+ defer m.cleanupSessions(collectChs)
- defer func() {
- m.collectingMu.Lock()
- for agentID, collectCh := range collectChs {
- close(collectCh)
- delete(m.collecting, agentID)
- }
- m.collectingMu.Unlock()
- }()
+ summary.Requested = m.requestAllAgents(ctx, agents, collectChs)
+ m.log.Info().Int("requested", summary.Requested).Int("waiting_for", len(collectChs)).
+ Msg("CollectLifecycle: requests sent, waiting for responses")
+
+ allData := m.waitForResponses(ctx, collectChs)
+ summary.Responded = len(allData)
+ if summary.Requested >= summary.Responded {
+ summary.NotResponded = summary.Requested - summary.Responded
+ }
+ m.log.Info().Int("responses_with_data", len(allData)).
+ Msg("CollectLifecycle: all responses collected, aggregating")
+ return m.buildInspectionResult(allData), summary
+}
+
+func (m *Manager) requestAllAgents(ctx context.Context, agents []*registry.AgentInfo,
+ collectChs map[string]chan *agentLifecycleData,
+) int {
requestedCount := 0
for _, agentInfo := range agents {
select {
case <-ctx.Done():
m.log.Info().Msg("CollectLifecycle: context canceled during request phase")
- return emptyResult()
+ return requestedCount
default:
}
- if m.grpcService == nil {
- m.log.Info().Str("agent_id", agentInfo.AgentID).Msg("CollectLifecycle: grpcService is nil, skipping request")
- continue
- }
+ m.registerSession(agentInfo.AgentID, collectChs)
if err := m.grpcService.RequestLifecycleData(agentInfo.AgentID); err != nil {
m.log.Info().Err(err).
Str("agent_id", agentInfo.AgentID).
Msg("Agent does not support lifecycle stream, skipping")
+ m.unregisterSession(agentInfo.AgentID, collectChs)
continue
}
- collectCh := make(chan *agentLifecycleData, 1)
- collectChs[agentInfo.AgentID] = collectCh
- m.collectingMu.Lock()
- m.collecting[agentInfo.AgentID] = collectCh
- m.collectingMu.Unlock()
requestedCount++
}
+ return requestedCount
+}
- m.log.Info().Int("requested", requestedCount).Int("waiting_for", len(collectChs)).Msg("CollectLifecycle: requests sent, waiting for responses")
-
+func (m *Manager) waitForResponses(ctx context.Context, collectChs map[string]chan *agentLifecycleData) []*agentLifecycleData {
allData := make([]*agentLifecycleData, 0, len(collectChs))
var dataMu sync.Mutex
var wg sync.WaitGroup
-
for agentID, collectCh := range collectChs {
wg.Add(1)
go func(id string, ch chan *agentLifecycleData) {
@@ -175,9 +215,7 @@ func (m *Manager) CollectLifecycle(ctx context.Context) *InspectionResult {
defer agentCancel()
select {
case <-agentCtx.Done():
- m.log.Warn().
- Str("agent_id", id).
- Msg("Timeout waiting for lifecycle data from agent")
+ m.log.Warn().Str("agent_id", id).Msg("Timeout waiting for lifecycle data from agent")
case data := <-ch:
if data != nil {
m.log.Info().
@@ -189,17 +227,21 @@ func (m *Manager) CollectLifecycle(ctx context.Context) *InspectionResult {
dataMu.Lock()
allData = append(allData, data)
dataMu.Unlock()
- } else {
- m.log.Info().Str("agent_id", id).Msg("CollectLifecycle: received nil data from agent")
}
}
}(agentID, collectCh)
}
-
wg.Wait()
- m.log.Info().Int("responses_with_data", len(allData)).Msg("CollectLifecycle: all responses collected, aggregating")
+ return allData
+}
- return m.buildInspectionResult(allData)
+func (m *Manager) cleanupSessions(collectChs map[string]chan *agentLifecycleData) {
+ m.collectingMu.Lock()
+ for agentID, collectCh := range collectChs {
+ close(collectCh)
+ delete(m.collecting, agentID)
+ }
+ m.collectingMu.Unlock()
}
func (m *Manager) buildInspectionResult(allData []*agentLifecycleData) *InspectionResult {
diff --git a/fodc/proxy/internal/lifecycle/manager_test.go b/fodc/proxy/internal/lifecycle/manager_test.go
index 7df3a2811..55a76f166 100644
--- a/fodc/proxy/internal/lifecycle/manager_test.go
+++ b/fodc/proxy/internal/lifecycle/manager_test.go
@@ -115,21 +115,24 @@ func TestManager_UpdateLifecycle_NoActiveCollection(t *testing.T) {
func TestManager_CollectLifecycle_NoAgents(t *testing.T) {
log := initTestLogger(t)
testRegistry := registry.NewAgentRegistry(log, 5*time.Second, 10*time.Second, 100)
+ defer testRegistry.Stop()
mgr := NewManager(testRegistry, nil, log)
- result := mgr.CollectLifecycle(context.Background())
+ result, summary := mgr.CollectLifecycle(context.Background())
require.NotNil(t, result)
assert.Empty(t, result.Groups)
assert.Empty(t, result.LifecycleStatuses)
+ assert.Equal(t, AgentSummary{}, summary, "summary must be zero when no agents are registered")
}
func TestManager_CollectLifecycle_NilRegistry(t *testing.T) {
log := initTestLogger(t)
mgr := NewManager(nil, nil, log)
- result := mgr.CollectLifecycle(context.Background())
+ result, summary := mgr.CollectLifecycle(context.Background())
require.NotNil(t, result)
assert.Empty(t, result.Groups)
+ assert.Equal(t, AgentSummary{}, summary, "summary must be zero when registry is nil")
}
func TestManager_CollectLifecycle_MultipleAgents(t *testing.T) {
@@ -169,9 +172,12 @@ func TestManager_CollectLifecycle_MultipleAgents(t *testing.T) {
},
}
+ var capturedSummary AgentSummary
done := make(chan *InspectionResult)
go func() {
- done <- mgr.CollectLifecycle(ctx)
+ result, summary := mgr.CollectLifecycle(ctx)
+ capturedSummary = summary
+ done <- result
}()
time.Sleep(50 * time.Millisecond)
@@ -194,6 +200,11 @@ func TestManager_CollectLifecycle_MultipleAgents(t *testing.T) {
podNames := []string{result.LifecycleStatuses[0].PodName, result.LifecycleStatuses[1].PodName}
assert.Contains(t, podNames, "pod-1")
assert.Contains(t, podNames, "pod-2")
+ // Verify summary reflects what actually happened
+ assert.Equal(t, 2, capturedSummary.Total, "summary.Total should match registered agents")
+ assert.Equal(t, 2, capturedSummary.Requested, "summary.Requested should match agents that accepted the request")
+ assert.Equal(t, 2, capturedSummary.Responded, "summary.Responded should match agents that returned data")
+ assert.Equal(t, 0, capturedSummary.NotResponded, "summary.NotResponded should be zero when all agents respond")
case <-time.After(5 * time.Second):
t.Fatal("Collection timed out")
@@ -219,7 +230,7 @@ func TestManager_CollectLifecycle_RequestError(t *testing.T) {
mockSender.SetRequestError(agentID, fmt.Errorf("connection refused"))
- result := mgr.CollectLifecycle(ctx)
+ result, _ := mgr.CollectLifecycle(ctx)
require.NotNil(t, result)
// Should return empty data since request failed
assert.Empty(t, result.Groups)
@@ -298,7 +309,10 @@ func TestManager_CollectLifecycle_WithGroups(t *testing.T) {
require.NoError(t, err)
done := make(chan *InspectionResult)
- go func() { done <- mgr.CollectLifecycle(ctx) }()
+ go func() {
+ result, _ := mgr.CollectLifecycle(ctx)
+ done <- result
+ }()
time.Sleep(50 * time.Millisecond)
mgr.UpdateLifecycle(agentID, "liaison-pod", &fodcv1.LifecycleData{
@@ -339,7 +353,10 @@ func TestManager_CollectLifecycle_GroupsFromFirstAgent(t *testing.T) {
require.NoError(t, err)
done := make(chan *InspectionResult)
- go func() { done <- mgr.CollectLifecycle(ctx) }()
+ go func() {
+ result, _ := mgr.CollectLifecycle(ctx)
+ done <- result
+ }()
time.Sleep(50 * time.Millisecond)
mgr.UpdateLifecycle(agentID1, "liaison-1", &fodcv1.LifecycleData{
@@ -374,6 +391,6 @@ func TestManager_CollectLifecycle_ContextCanceled(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
cancel() // Cancel immediately
- result := mgr.CollectLifecycle(ctx)
+ result, _ := mgr.CollectLifecycle(ctx)
require.NotNil(t, result)
}