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), ),