Backlog/v12 add keepalive connections

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

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

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

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

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

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