From e0fff494bc2b10e36b0c0f0d09319dc90a834931 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alex=20S=C3=A1nchez?= Date: Wed, 2 Sep 2026 10:37:20 -0600 Subject: [PATCH 1/2] fix[backend](agentmanager): added grpc keepalive to agent-manager client to prevent idle connection resets --- backend/pkg/agentmanager/client.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/backend/pkg/agentmanager/client.go b/backend/pkg/agentmanager/client.go index a0336c6a9..8c205241c 100644 --- a/backend/pkg/agentmanager/client.go +++ b/backend/pkg/agentmanager/client.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "strconv" + "time" "github.com/utmstack/utmstack/backend/pkg/authz" @@ -14,6 +15,7 @@ import ( "github.com/utmstack/utmstack/backend/pkg/env" "google.golang.org/grpc" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" "google.golang.org/grpc/metadata" ) @@ -39,6 +41,11 @@ func NewClient() (*AgentManagerClient, error) { addr, grpc.WithTransportCredentials(tlsCreds), grpc.WithPerRPCCredentials(&internalKeyCreds{key: internalKey}), + grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: 30 * time.Second, + Timeout: 10 * time.Second, + PermitWithoutStream: true, + }), ) if err != nil { return nil, fmt.Errorf("agentmanager: dial %s: %w", addr, err) From b2e686fd6cbb0a5dd6990354f000c86d0702626f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alex=20S=C3=A1nchez?= Date: Wed, 2 Sep 2026 14:24:52 -0600 Subject: [PATCH 2/2] fix[agents-manager](grpc): add keepalive pings to agent/collector dials and agent-manager/log-input servers --- agent-manager/agent/utmgrpc.go | 14 ++++++++++++++ agent/agent/conn.go | 13 ++++++++++++- collectors/as400/conn/conn.go | 12 +++++++++++- collectors/forwarder/upstream/conn.go | 12 +++++++++++- collectors/utmstack/conn/conn.go | 12 +++++++++++- log-input/ingest/server.go | 12 ++++++++++++ 6 files changed, 71 insertions(+), 4 deletions(-) diff --git a/agent-manager/agent/utmgrpc.go b/agent-manager/agent/utmgrpc.go index 80679a5a2..0c50f0c13 100644 --- a/agent-manager/agent/utmgrpc.go +++ b/agent-manager/agent/utmgrpc.go @@ -12,6 +12,7 @@ import ( "google.golang.org/grpc/credentials" "google.golang.org/grpc/health" "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/keepalive" ) func InitGrpcServer() { @@ -48,8 +49,21 @@ func StartGrpcServer() { MinVersion: tls.VersionTLS13, }) + // Keepalive pings keep the connection to the frontend nginx alive while + // AgentStream is idle: nginx's grpc_read_timeout (900s) counts time without + // upstream bytes, and the PING ACKs we answer to client pings reset it on + // every hop. PermitWithoutStream also covers the PingService-only window + // where AgentStream is down and the agent is reconnecting it. grpcServer := grpc.NewServer( grpc.Creds(transportCredentials), + grpc.KeepaliveParams(keepalive.ServerParameters{ + Time: 30 * time.Second, + Timeout: 10 * time.Second, + }), + grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ + MinTime: 10 * time.Second, + PermitWithoutStream: true, + }), grpc.ChainUnaryInterceptor(UnaryInterceptor), grpc.StreamInterceptor(StreamInterceptor)) diff --git a/agent/agent/conn.go b/agent/agent/conn.go index ded43381c..da3b041a3 100644 --- a/agent/agent/conn.go +++ b/agent/agent/conn.go @@ -11,6 +11,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/connectivity" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" ) const ( @@ -71,10 +72,20 @@ func connectToServer(addrs, port string, skip bool) (*grpc.ClientConn, error) { return nil, fmt.Errorf("failed to connect to Server: %v", err) } + // Keepalive pings (30s) travel over the shared HTTP/2 connection and + // reset grpc_read_timeout on the nginx hops in front of the agent + // manager, so an idle AgentStream is no longer torn down after 900s. + // PermitWithoutStream matters: pings keep flowing while AgentStream is + // down and being redialed, and on the PingService-only window. conn, err = grpc.NewClient( serverAddress, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMessageSize)), - grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: skip}))) + grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: skip})), + grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: 30 * time.Second, + Timeout: 10 * time.Second, + PermitWithoutStream: true, + })) if err != nil { connectionAttemps++ utils.Logger.ErrorF("error connecting to Server, trying again in %.0f seconds", reconnectDelay.Seconds()) diff --git a/collectors/as400/conn/conn.go b/collectors/as400/conn/conn.go index ed253a427..54c29f44f 100644 --- a/collectors/as400/conn/conn.go +++ b/collectors/as400/conn/conn.go @@ -10,6 +10,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/connectivity" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" ) const ( @@ -87,10 +88,19 @@ func connectToServer(addrs, port string, skip bool) (*grpc.ClientConn, error) { return nil, utils.Logger.ErrorF("failed to connect to Server: %v", err) } + // Keepalive pings (30s) reset grpc_read_timeout on the nginx hops in + // front of the agent manager / log input, so an idle collector stream + // (no logs for 900s) is not torn down. PermitWithoutStream keeps pings + // flowing while a stream is being re-established. conn, err = grpc.NewClient( serverAddress, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMessageSize)), - grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: skip}))) + grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: skip})), + grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: 30 * time.Second, + Timeout: 10 * time.Second, + PermitWithoutStream: true, + })) if err != nil { connectionAttemps++ utils.Logger.ErrorF("error connecting to Server, trying again in %.0f seconds", reconnectDelay.Seconds()) diff --git a/collectors/forwarder/upstream/conn.go b/collectors/forwarder/upstream/conn.go index 6f20cd3c3..e91678969 100644 --- a/collectors/forwarder/upstream/conn.go +++ b/collectors/forwarder/upstream/conn.go @@ -11,6 +11,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/connectivity" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" ) const ( @@ -71,10 +72,19 @@ func connectToServer(addrs, port string, skip bool) (*grpc.ClientConn, error) { return nil, fmt.Errorf("failed to connect to Server: %v", err) } + // Keepalive pings (30s) reset grpc_read_timeout on the nginx hops in + // front of the agent manager / log input, so an idle collector stream + // (no logs for 900s) is not torn down. PermitWithoutStream keeps pings + // flowing while a stream is being re-established. conn, err = grpc.NewClient( serverAddress, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMessageSize)), - grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: skip}))) + grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: skip})), + grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: 30 * time.Second, + Timeout: 10 * time.Second, + PermitWithoutStream: true, + })) if err != nil { connectionAttemps++ utils.Logger.ErrorF("error connecting to Server, trying again in %.0f seconds", reconnectDelay.Seconds()) diff --git a/collectors/utmstack/conn/conn.go b/collectors/utmstack/conn/conn.go index 8becb047b..1977fac26 100644 --- a/collectors/utmstack/conn/conn.go +++ b/collectors/utmstack/conn/conn.go @@ -10,6 +10,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/connectivity" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" ) const ( @@ -87,10 +88,19 @@ func connectToServer(addrs, port string, skip bool) (*grpc.ClientConn, error) { return nil, utils.Logger.ErrorF("failed to connect to Server: %v", err) } + // Keepalive pings (30s) reset grpc_read_timeout on the nginx hops in + // front of the agent manager / log input, so an idle collector stream + // (no logs for 900s) is not torn down. PermitWithoutStream keeps pings + // flowing while a stream is being re-established. conn, err = grpc.NewClient( serverAddress, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMessageSize)), - grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: skip}))) + grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: skip})), + grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: 30 * time.Second, + Timeout: 10 * time.Second, + PermitWithoutStream: true, + })) if err != nil { connectionAttemps++ utils.Logger.ErrorF("error connecting to Server, trying again in %.0f seconds", reconnectDelay.Seconds()) diff --git a/log-input/ingest/server.go b/log-input/ingest/server.go index 6468debdb..f918439d2 100644 --- a/log-input/ingest/server.go +++ b/log-input/ingest/server.go @@ -13,6 +13,7 @@ import ( "google.golang.org/grpc/credentials" "google.golang.org/grpc/health" grpcHealth "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/keepalive" "github.com/utmstack/UTMStack/log-input/auth" "github.com/utmstack/UTMStack/log-input/config" @@ -61,8 +62,19 @@ func NewServer(cfg *config.Config, a *auth.Service, pub Publisher) (*Server, err s := &Server{ cfg: cfg, pub: pub, + // Keepalive pings keep the collector's idle ProcessLog stream alive + // through the frontend nginx grpc_read_timeout (900s): the PING ACKs we + // answer reset the upstream read timer on every hop. grpc: grpc.NewServer( grpc.Creds(creds), + grpc.KeepaliveParams(keepalive.ServerParameters{ + Time: 30 * time.Second, + Timeout: 10 * time.Second, + }), + grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ + MinTime: 10 * time.Second, + PermitWithoutStream: true, + }), grpc.ChainUnaryInterceptor(m.unary), grpc.ChainStreamInterceptor(m.stream), ),