mirror of
https://github.com/henrygd/beszel.git
synced 2026-10-22 06:12:33 +00:00
fix(hub): stop keepalive pings after prolonged agent silence
The keepalive ping on a failed fetch re-arms the read deadline without needing a reply, so a dead peer (power loss, partition) kept its connection until TCP gave up. Track when the agent last sent a message and only ping while that is under maxAgentSilence (3m). A live agent always answers within interval + its own 120s deadline, so slow agents are unaffected; a dead one is now released by the read deadline and falls back through the normal down/SSH path.
This commit is contained in:
@@ -174,24 +174,11 @@ func (sys *System) update() error {
|
||||
|
||||
data, err := sys.fetchDataFromAgent(options)
|
||||
if err != nil {
|
||||
// Keep the WebSocket alive even though this fetch failed.
|
||||
//
|
||||
// Only a message from the agent resets the read deadline, and an agent
|
||||
// still collecting sends nothing. Without this ping the deadline expires
|
||||
// between polls, the hub closes the connection, the agent reconnects,
|
||||
// collects slowly again, and the cycle repeats. Pinging here is the same
|
||||
// move handlePaused already makes for a paused system, and Ping() pushes
|
||||
// the deadline out as it sends, so it works without an answer, which
|
||||
// matters because a collecting agent cannot answer.
|
||||
//
|
||||
// This is why the deadline has to exceed interval + wsDataRequestTimeout:
|
||||
// the ping only happens once the fetch gives up. See ws.deadline.
|
||||
//
|
||||
// The system is still reported down: that comes from the error we
|
||||
// return. Only the connection survives, so recovery costs no reconnect.
|
||||
// A failed ping needs no handling either, since a connection that
|
||||
// cannot be pinged runs into its deadline anyway.
|
||||
if sys.WsConn != nil && sys.WsConn.IsConnected() {
|
||||
// Ping so a slow agent's connection survives until the next poll. Ping()
|
||||
// extends the read deadline without needing a reply (see ws.deadline).
|
||||
// Once the agent has been silent for maxAgentSilence, stop pinging and let
|
||||
// the deadline close what is likely a dead connection.
|
||||
if sys.WsConn != nil && sys.WsConn.IsConnected() && sys.WsConn.TimeSinceLastMessage() < maxAgentSilence {
|
||||
_ = sys.WsConn.Ping()
|
||||
}
|
||||
return err
|
||||
@@ -820,6 +807,12 @@ func (sys *System) fetchDataFromAgent(options common.DataRequestOptions) (*syste
|
||||
// so this must be well above the request manager's 5s default.
|
||||
var wsDataRequestTimeout = 30 * time.Second
|
||||
|
||||
// maxAgentSilence is how long a WebSocket agent may go without sending anything
|
||||
// before failed polls stop keeping its connection alive. An agent whose
|
||||
// collection runs past its own 120s deadline drops the connection itself, so a
|
||||
// live agent always answers within interval + 120s.
|
||||
var maxAgentSilence = 3 * time.Minute
|
||||
|
||||
func (sys *System) fetchDataViaWebSocket(options common.DataRequestOptions) (*system.CombinedData, error) {
|
||||
if sys.WsConn == nil || !sys.WsConn.IsConnected() {
|
||||
return nil, errors.New("no websocket connection")
|
||||
|
||||
@@ -185,18 +185,26 @@ const timelineScale = 100
|
||||
// happens once the fetch gives up, which is interval + wsDataRequestTimeout into
|
||||
// the cycle, so the read deadline has to outlast that. With a deadline below
|
||||
// that sum the connection is already closed when the ping would go out, which is
|
||||
// the reconnect loop from #2294.
|
||||
// the reconnect loop from #2294. An agent silent for longer than
|
||||
// maxAgentSilence is no longer pinged, so its connection is allowed to expire.
|
||||
func TestSlowAgentKeepsConnectionAcrossPolls(t *testing.T) {
|
||||
tick := time.Duration(interval) * time.Millisecond / timelineScale
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
deadline time.Duration
|
||||
polls int
|
||||
wantConnected bool
|
||||
}{
|
||||
{"production deadline outlasts a failed poll", ws.Deadline() / timelineScale, true},
|
||||
// The first failed poll shows the ping lands, the second shows the ping
|
||||
// keeps re-arming the deadline.
|
||||
{"production deadline outlasts a failed poll", ws.Deadline() / timelineScale, 2, true},
|
||||
// 70s was the old value and sits below the 90s a failed poll needs.
|
||||
{"deadline below interval plus request timeout", 70 * time.Second / timelineScale, false},
|
||||
{"deadline below interval plus request timeout", 70 * time.Second / timelineScale, 2, false},
|
||||
// The third failed poll ends 210s after the last message, past
|
||||
// maxAgentSilence, so it skips the ping and the deadline closes the
|
||||
// connection before the fifth poll.
|
||||
{"agent silent past maxAgentSilence", ws.Deadline() / timelineScale, 5, false},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
@@ -206,6 +214,9 @@ func TestSlowAgentKeepsConnectionAcrossPolls(t *testing.T) {
|
||||
previousTimeout := wsDataRequestTimeout
|
||||
wsDataRequestTimeout = 30 * time.Second / timelineScale
|
||||
t.Cleanup(func() { wsDataRequestTimeout = previousTimeout })
|
||||
previousSilence := maxAgentSilence
|
||||
maxAgentSilence = 3 * time.Minute / timelineScale
|
||||
t.Cleanup(func() { maxAgentSilence = previousSilence })
|
||||
|
||||
connections := make(chan *ws.WsConn, 1)
|
||||
// The real handler: OnClose has to run, or a dropped connection
|
||||
@@ -244,9 +255,7 @@ func TestSlowAgentKeepsConnectionAcrossPolls(t *testing.T) {
|
||||
require.NoError(t, err, "the first poll must succeed")
|
||||
pollZero := time.Now()
|
||||
|
||||
// Two failed polls: the first shows the ping lands, the second shows
|
||||
// the ping keeps re-arming the deadline.
|
||||
for poll := 1; poll <= 2; poll++ {
|
||||
for poll := 1; poll <= tc.polls; poll++ {
|
||||
time.Sleep(time.Until(pollZero.Add(time.Duration(poll) * tick)))
|
||||
if !sys.WsConn.IsConnected() {
|
||||
break
|
||||
|
||||
@@ -36,6 +36,7 @@ type WsConn struct {
|
||||
requestManager *RequestManager
|
||||
DownChan chan struct{}
|
||||
agentVersion semver.Version
|
||||
lastMessage atomic.Int64 // unix nanoseconds of the last message from the agent
|
||||
}
|
||||
|
||||
// FingerprintRecord is fingerprints collection record data in the hub
|
||||
@@ -66,6 +67,7 @@ func NewWsConnection(conn *gws.Conn, agentVersion semver.Version) *WsConn {
|
||||
agentVersion: agentVersion,
|
||||
}
|
||||
ws.conn.Store(conn)
|
||||
ws.lastMessage.Store(time.Now().UnixNano())
|
||||
return ws
|
||||
}
|
||||
|
||||
@@ -85,6 +87,7 @@ func (h *Handler) OnMessage(conn *gws.Conn, message *gws.Message) {
|
||||
_ = conn.WriteClose(1000, nil)
|
||||
return
|
||||
}
|
||||
wsConn.(*WsConn).lastMessage.Store(time.Now().UnixNano())
|
||||
wsConn.(*WsConn).requestManager.handleResponse(message)
|
||||
}
|
||||
|
||||
@@ -126,6 +129,11 @@ func (ws *WsConn) Ping() error {
|
||||
return conn.WritePing(nil)
|
||||
}
|
||||
|
||||
// TimeSinceLastMessage returns how long it has been since the agent last sent a message.
|
||||
func (ws *WsConn) TimeSinceLastMessage() time.Duration {
|
||||
return time.Since(time.Unix(0, ws.lastMessage.Load()))
|
||||
}
|
||||
|
||||
// sendMessage encodes data to CBOR and sends it as a binary message to the agent.
|
||||
// This is kept for backwards compatibility but new actions should use RequestManager.
|
||||
func (ws *WsConn) sendMessage(data common.HubRequest[any]) error {
|
||||
|
||||
Reference in New Issue
Block a user