Merge pull request firecracker-microvm#432 from kzys/stopvm-cleanup

Close all Unix abstract sockets correctly

Author: kzys

firecracker-control/local.go

@@ -21,7 +21,9 @@ import (
 	"os/exec"
 	"strconv"
 	"strings"
+	"sync"
 	"syscall"
+	"time"
 
 	"github.com/containerd/containerd/identifiers"
 	"github.com/containerd/containerd/log"
@@ -30,6 +32,7 @@ import (
 	"github.com/containerd/containerd/runtime/v2/shim"
 	"github.com/containerd/containerd/sys"
 	"github.com/golang/protobuf/ptypes/empty"
+	"github.com/hashicorp/go-multierror"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc/codes"
@@ -47,6 +50,7 @@ import (
 var (
 	_               fccontrolTtrpc.FirecrackerService = (*local)(nil)
 	ttrpcAddressEnv                                   = "TTRPC_ADDRESS"
+	stopVMInterval                                    = 10 * time.Millisecond
 )
 
 func init() {
@@ -64,6 +68,9 @@ type local struct {
 	containerdAddress string
 	logger            *logrus.Entry
 	config            *config.Config
+
+	processesMu sync.Mutex
+	processes   map[string]int32
 }
 
 func newLocal(ic *plugin.InitContext) (*local, error) {
@@ -80,6 +87,7 @@ func newLocal(ic *plugin.InitContext) (*local, error) {
 		containerdAddress: ic.Address,
 		logger:            log.G(ic.Context),
 		config:            cfg,
+		processes:         make(map[string]int32),
 	}, nil
 }
 
@@ -116,7 +124,7 @@ func (s *local) CreateVM(requestCtx context.Context, req *proto.CreateVMRequest)
 
 	shimSocket, err := shim.NewSocket(shimSocketAddress)
 	if isEADDRINUSE(err) {
-		return nil, status.Errorf(codes.AlreadyExists, "VM with ID %q already exists", id)
+		return nil, status.Errorf(codes.AlreadyExists, "VM with ID %q already exists (socket: %q)", id, shimSocketAddress)
 	} else if err != nil {
 		err = errors.Wrapf(err, "failed to open shim socket at address %q", shimSocketAddress)
 		s.logger.WithError(err).Error()
@@ -199,9 +207,17 @@ func (s *local) CreateVM(requestCtx context.Context, req *proto.CreateVMRequest)
 		return nil, err
 	}
 
+	s.addShim(shimSocketAddress, cmd)
+
 	return resp, nil
 }
 
+func (s *local) addShim(address string, cmd *exec.Cmd) {
+	s.processesMu.Lock()
+	defer s.processesMu.Unlock()
+	s.processes[address] = int32(cmd.Process.Pid)
+}
+
 func (s *local) shimFirecrackerClient(requestCtx context.Context, vmID string) (*fcclient.Client, error) {
 	if err := identifiers.Validate(vmID); err != nil {
 		return nil, errors.Wrap(err, "invalid id")
@@ -224,16 +240,34 @@ func (s *local) StopVM(requestCtx context.Context, req *proto.StopVMRequest) (*e
 	if err != nil {
 		return nil, err
 	}
-
 	defer client.Close()
 
-	resp, err := client.StopVM(requestCtx, req)
+	resp, shimErr := client.StopVM(requestCtx, req)
+	waitErr := s.waitForShimToExit(requestCtx, req.VMID)
+
+	// Assuming the shim is returning containerd's error code, return the error as is if possible.
+	if waitErr == nil {
+		return resp, shimErr
+	}
+	return resp, multierror.Append(shimErr, waitErr).ErrorOrNil()
+}
+
+func (s *local) waitForShimToExit(ctx context.Context, vmID string) error {
+	socketAddr, err := fcShim.SocketAddress(ctx, vmID)
 	if err != nil {
-		s.logger.WithError(err).Error()
-		return nil, err
+		return err
+	}
+
+	s.processesMu.Lock()
+	defer s.processesMu.Unlock()
+
+	pid, ok := s.processes[socketAddr]
+	if !ok {
+		return errors.Errorf("failed to find a shim process for %q", socketAddr)
 	}
+	defer delete(s.processes, socketAddr)
 
-	return resp, err
+	return internal.WaitForPidToExit(ctx, stopVMInterval, pid)
 }
 
 // GetVMInfo returns metadata for the VM with the given VMID.
@@ -387,6 +421,15 @@ func (s *local) newShim(ns, vmID, containerdAddress string, shimSocket *net.Unix
 				logger.WithError(err).Error("shim has been unexpectedly terminated")
 			}
 		}
+
+		// Close all Unix abstract sockets.
+		if err := shimSocketFile.Close(); err != nil {
+			logger.WithError(err).Errorf("failed to close %q", shimSocketFile.Name())
+		}
+		if err := fcSocketFile.Close(); err != nil {
+			logger.WithError(err).Errorf("failed to close %q", fcSocketFile.Name())
+		}
+
 		if err := os.RemoveAll(shimDir.RootPath()); err != nil {
 			logger.WithError(err).Errorf("failed to remove %q", shimDir.RootPath())
 		}

runtime/service_integ_test.go

@@ -583,7 +583,7 @@ func TestLongUnixSocketPath_Isolated(t *testing.T) {
 	// default location we store state results in a path like
 	// "/run/firecracker-containerd/default/<vmID>" (with len 112).
 	const maxUnixSockLen = 108
-	vmID := strings.Repeat("x", 72)
+	vmID := strings.Repeat("x", 76)
 
 	ctx := namespaces.WithNamespace(context.Background(), "default")
 
@@ -597,16 +597,14 @@ func TestLongUnixSocketPath_Isolated(t *testing.T) {
 		{
 			name: "Without Jailer",
 			request: proto.CreateVMRequest{
-				VMID:              vmID + "noop",
+				VMID:              vmID,
 				NetworkInterfaces: []*proto.FirecrackerNetworkInterface{},
 			},
 		},
 		{
 			name: "With Jailer",
 			request: proto.CreateVMRequest{
-				// We somehow cannot use the same VM ID here.
-				// https://github.com/firecracker-microvm/firecracker-containerd/issues/409
-				VMID:              vmID + "jail",
+				VMID:              vmID,
 				NetworkInterfaces: []*proto.FirecrackerNetworkInterface{},
 				JailerConfig: &proto.JailerConfig{
 					UID: 30000,
@@ -645,6 +643,14 @@ func TestLongUnixSocketPath_Isolated(t *testing.T) {
 
 			_, err = fcClient.StopVM(ctx, &proto.StopVMRequest{VMID: vmID})
 			require.NoError(t, err)
+
+			matches, err := findProcess(ctx, findShim)
+			require.NoError(t, err)
+			require.Empty(t, matches)
+
+			matches, err = findProcess(ctx, findFirecracker)
+			require.NoError(t, err)
+			require.Empty(t, matches)
 		})
 	}
 }
@@ -1566,18 +1572,18 @@ func TestStopVM_Isolated(t *testing.T) {
 
 			test.stopFunc(ctx, fcClient, createVMRequest)
 
-			// If the function above uses StopVMRequest,
-			// shim guarantees that the underlying Firecracker process is dead
+			// If the function above uses StopVMRequest, all underlying processes must be dead
 			// (either gracefully or forcibly) at the end of the request.
 			if test.withStopVM {
 				fcExists, err := process.PidExists(fcProcess.Pid)
 				require.NoError(err, "failed to find firecracker")
 				require.False(fcExists, "firecracker %s is still there", vmID)
+
+				shimExists, err := process.PidExists(shimProcess.Pid)
+				require.NoError(err, "failed to find shim")
+				require.False(shimExists, "shim %s is still there", vmID)
 			}
 
-			// Since the shim is the one which is writing the response,
-			// it cannot guarantee that the shim itself is dead at the end of the request.
-			// But it should be dead eventually.
 			err = internal.WaitForPidToExit(ctx, time.Second, shimProcess.Pid)
 			require.NoError(err, "shim hasn't been terminated")
 		}