Merge pull request #4889 from mn-ram/fix/guestagent-client-leak

hostagent: close stale GuestAgentClient on reconnect to stop ClientConn leak

Author: unsuman

pkg/guestagent/api/client/client.go

@@ -18,7 +18,8 @@ import (
 )
 
 type GuestAgentClient struct {
-	cli api.GuestServiceClient
+	cli  api.GuestServiceClient
+	conn *grpc.ClientConn
 }
 
 func NewGuestAgentClient(dialFn func(ctx context.Context) (net.Conn, error)) (*GuestAgentClient, error) {
@@ -44,10 +45,20 @@ func NewGuestAgentClient(dialFn func(ctx context.Context) (net.Conn, error)) (*G
 	}
 	client := api.NewGuestServiceClient(clientConn)
 	return &GuestAgentClient{
-		cli: client,
+		cli:  client,
+		conn: clientConn,
 	}, nil
 }
 
+// Close releases the underlying gRPC connection. It is safe to call on a nil
+// receiver or a client whose connection has already been closed.
+func (c *GuestAgentClient) Close() error {
+	if c == nil || c.conn == nil {
+		return nil
+	}
+	return c.conn.Close()
+}
+
 func (c *GuestAgentClient) Info(ctx context.Context) (*api.Info, error) {
 	return c.cli.GetInfo(ctx, &emptypb.Empty{})
 }

pkg/hostagent/hostagent.go

@@ -728,7 +728,8 @@ func (a *HostAgent) watchGuestAgentEvents(ctx context.Context) {
 
 	go func() {
 		if a.instConfig.MountInotify != nil && *a.instConfig.MountInotify {
-			if a.client == nil || !isGuestAgentSocketAccessible(ctx, a.client) {
+			client := a.getClient()
+			if client == nil || !isGuestAgentSocketAccessible(ctx, client) {
 				if a.driver.ForwardGuestAgent() {
 					sshAddress, sshPort := a.sshAddressPort()
 					_ = forwardSSH(ctx, a.sshConfig, sshAddress, sshPort, localUnix, remoteUnix, verbForward, false)
@@ -743,9 +744,20 @@ func (a *HostAgent) watchGuestAgentEvents(ctx context.Context) {
 
 	// ensure close before ctx is cancelled
 	a.cleanUp(a.grpcPortForwarder.Close)
+	a.cleanUp(func() error {
+		a.clientMu.Lock()
+		defer a.clientMu.Unlock()
+		if a.client == nil {
+			return nil
+		}
+		err := a.client.Close()
+		a.client = nil
+		return err
+	})
 
 	for {
-		if a.client == nil || !isGuestAgentSocketAccessible(ctx, a.client) {
+		client := a.getClient()
+		if client == nil || !isGuestAgentSocketAccessible(ctx, client) {
 			if a.driver.ForwardGuestAgent() {
 				sshAddress, sshPort := a.sshAddressPort()
 				_ = forwardSSH(ctx, a.sshConfig, sshAddress, sshPort, localUnix, remoteUnix, verbForward, false)
@@ -820,12 +832,30 @@ func isGuestAgentSocketAccessible(ctx context.Context, client *guestagentclient.
 	return err == nil
 }
 
+// getClient returns the current guest agent client under the clientMu lock,
+// so callers observe a consistent (non-mid-transition) snapshot.
+func (a *HostAgent) getClient() *guestagentclient.GuestAgentClient {
+	a.clientMu.Lock()
+	defer a.clientMu.Unlock()
+	return a.client
+}
+
 func (a *HostAgent) getOrCreateClient(ctx context.Context) (*guestagentclient.GuestAgentClient, error) {
 	a.clientMu.Lock()
 	defer a.clientMu.Unlock()
 	if a.client != nil && isGuestAgentSocketAccessible(ctx, a.client) {
 		return a.client, nil
 	}
+	// The previous client (if any) is unreachable: close its underlying gRPC
+	// ClientConn before replacing it, otherwise the resolver/transport
+	// goroutines and the dialed net.Conn (e.g. forwarded ga.sock) leak across
+	// every guest agent restart or VM reboot.
+	if a.client != nil {
+		if err := a.client.Close(); err != nil {
+			logrus.WithError(err).Debug("failed to close stale guest agent client")
+		}
+		a.client = nil
+	}
 	var err error
 	a.client, err = guestagentclient.NewGuestAgentClient(a.createConnection)
 	return a.client, err