Remove pods with uninstall

Author: vignesh-goutham

internal/containerd/install.go

@@ -7,12 +7,17 @@ import (
 	"os/exec"
 	"time"
 
+	internalapi "github.com/containerd/containerd/integration/cri-api/pkg/apis"
+	"github.com/containerd/containerd/integration/remote"
 	"github.com/pkg/errors"
+	"go.uber.org/zap"
+	v1 "k8s.io/cri-api/pkg/apis/runtime/v1"
 
 	"github.com/aws/eks-hybrid/internal/artifact"
 	"github.com/aws/eks-hybrid/internal/daemon"
 	"github.com/aws/eks-hybrid/internal/system"
 	"github.com/aws/eks-hybrid/internal/tracker"
+	"github.com/aws/eks-hybrid/internal/util"
 	"github.com/aws/eks-hybrid/internal/util/cmd"
 )
 
@@ -117,3 +122,78 @@ func areContainerdAndRuncInstalled() bool {
 	_, runcNotFoundErr := exec.LookPath(runcPackageName)
 	return containerdNotFoundErr == nil && runcNotFoundErr == nil
 }
+
+// Client is a containerd runtime client wrapper
+// Holds the internalapi.RuntimeService for pod/container operations
+type Client struct {
+	Runtime internalapi.RuntimeService
+}
+
+// NewClient creates a new Client with a real containerd runtime service
+func NewClient() (*Client, error) {
+	runtime, err := remote.NewRuntimeService(ContainerRuntimeEndpoint, 5*time.Second)
+	if err != nil {
+		return nil, err
+	}
+	return &Client{Runtime: runtime}, nil
+}
+
+// RemovePods stops and removes all pod sandboxes and containers in the k8s.io namespace on the node
+func (c *Client) RemovePods() error {
+	podSandboxes, err := c.Runtime.ListPodSandbox(&v1.PodSandboxFilter{
+		State: &v1.PodSandboxStateValue{
+			State: v1.PodSandboxState_SANDBOX_READY,
+		},
+	})
+	if err != nil {
+		return errors.Wrap(err, "listing pod sandboxes")
+	}
+
+	for _, sandbox := range podSandboxes {
+		zap.L().Info("Stopping pod..", zap.String("pod", sandbox.Metadata.Name))
+		err := util.RetryExponentialBackoff(3, 2*time.Second, func() error {
+			if err := c.Runtime.StopPodSandbox(sandbox.Id); err != nil {
+				return errors.Wrapf(err, "stopping pod %s", sandbox.Id)
+			}
+			if err := c.Runtime.RemovePodSandbox(sandbox.Id); err != nil {
+				return errors.Wrapf(err, "removing pod %s", sandbox.Id)
+			}
+			return nil
+		})
+		if err != nil {
+			zap.L().Info("ignored error stopping pod", zap.Error(err))
+		}
+	}
+
+	// If pod sandbox deletion fails, we can try to stop and remove containers individually
+	// We do not pass in a container state filter here as we want to remove all containers
+	// including stopped ones as they arent GCed by containerd post daemon stop.
+	containers, err := c.Runtime.ListContainers(nil)
+	if err != nil {
+		return errors.Wrap(err, "listing containers")
+	}
+
+	for _, container := range containers {
+		status, err := c.Runtime.ContainerStatus(container.Id)
+		if err != nil {
+			return errors.Wrapf(err, "getting container status for %s", container.Id)
+		}
+		zap.L().Info("Stopping container..", zap.String("container", container.Metadata.Name))
+		err = util.RetryExponentialBackoff(3, 2*time.Second, func() error {
+			if status.State == v1.ContainerState_CONTAINER_RUNNING {
+				if err := c.Runtime.StopContainer(container.Id, 0); err != nil {
+					return errors.Wrapf(err, "stopping container %s", container.Id)
+				}
+			}
+
+			if err := c.Runtime.RemoveContainer(container.Id); err != nil {
+				return errors.Wrapf(err, "removing container %s", container.Id)
+			}
+			return nil
+		})
+		if err != nil {
+			zap.L().Info("ignored error removing container", zap.Error(err))
+		}
+	}
+	return nil
+}

internal/containerd/install_test.go

@@ -0,0 +1,87 @@
+package containerd_test
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	v1 "k8s.io/cri-api/pkg/apis/runtime/v1"
+
+	"github.com/aws/eks-hybrid/internal/containerd"
+	"github.com/aws/eks-hybrid/internal/containerd/mocks"
+)
+
+func TestClient_RemovePods(t *testing.T) {
+	t.Run("no pods or containers", func(t *testing.T) {
+		m := new(mocks.MockRuntimeService)
+		m.On("ListPodSandbox", mock.Anything).Return([]*v1.PodSandbox{}, nil)
+		m.On("ListContainers", (*v1.ContainerFilter)(nil)).Return([]*v1.Container{}, nil)
+		c := &containerd.Client{Runtime: m}
+		err := c.RemovePods()
+		assert.NoError(t, err)
+		m.AssertExpectations(t)
+	})
+
+	t.Run("error listing pods", func(t *testing.T) {
+		m := new(mocks.MockRuntimeService)
+		m.On("ListPodSandbox", mock.Anything).Return(nil, errors.New("fail list pods"))
+		c := &containerd.Client{Runtime: m}
+		err := c.RemovePods()
+		assert.ErrorContains(t, err, "fail list pods")
+		m.AssertExpectations(t)
+	})
+
+	t.Run("error stopping/removing pod", func(t *testing.T) {
+		m := new(mocks.MockRuntimeService)
+		pod := &v1.PodSandbox{Metadata: &v1.PodSandboxMetadata{Name: "pod1"}, Id: "pod1id"}
+		m.On("ListPodSandbox", mock.Anything).Return([]*v1.PodSandbox{pod}, nil)
+		m.On("StopPodSandbox", "pod1id").Return(errors.New("fail stop pod")).Times(3)
+		m.On("ListContainers", (*v1.ContainerFilter)(nil)).Return([]*v1.Container{}, nil)
+		c := &containerd.Client{Runtime: m}
+		err := c.RemovePods()
+		assert.NoError(t, err, "RemovePods should ignore pod stop errors")
+		m.AssertExpectations(t)
+	})
+
+	t.Run("error listing containers", func(t *testing.T) {
+		m := new(mocks.MockRuntimeService)
+		m.On("ListPodSandbox", mock.Anything).Return([]*v1.PodSandbox{}, nil)
+		m.On("ListContainers", (*v1.ContainerFilter)(nil)).Return(nil, errors.New("fail list containers"))
+		c := &containerd.Client{Runtime: m}
+		err := c.RemovePods()
+		assert.ErrorContains(t, err, "fail list containers")
+		m.AssertExpectations(t)
+	})
+
+	t.Run("error stopping/removing container", func(t *testing.T) {
+		m := new(mocks.MockRuntimeService)
+		container := &v1.Container{Metadata: &v1.ContainerMetadata{Name: "c1"}, Id: "cid1"}
+		status := &v1.ContainerStatus{State: v1.ContainerState_CONTAINER_RUNNING}
+		m.On("ListPodSandbox", mock.Anything).Return([]*v1.PodSandbox{}, nil)
+		m.On("ListContainers", (*v1.ContainerFilter)(nil)).Return([]*v1.Container{container}, nil)
+		m.On("ContainerStatus", "cid1").Return(status, nil)
+		m.On("StopContainer", "cid1", int64(0)).Return(errors.New("fail stop container")).Times(3)
+		c := &containerd.Client{Runtime: m}
+		err := c.RemovePods()
+		assert.NoError(t, err, "RemovePods should ignore container stop errors")
+		m.AssertExpectations(t)
+	})
+
+	t.Run("all success", func(t *testing.T) {
+		m := new(mocks.MockRuntimeService)
+		pod := &v1.PodSandbox{Metadata: &v1.PodSandboxMetadata{Name: "pod1"}, Id: "pod1id"}
+		container := &v1.Container{Metadata: &v1.ContainerMetadata{Name: "c1"}, Id: "cid1"}
+		status := &v1.ContainerStatus{State: v1.ContainerState_CONTAINER_EXITED}
+		m.On("ListPodSandbox", mock.Anything).Return([]*v1.PodSandbox{pod}, nil)
+		m.On("StopPodSandbox", "pod1id").Return(nil)
+		m.On("RemovePodSandbox", "pod1id").Return(nil)
+		m.On("ListContainers", (*v1.ContainerFilter)(nil)).Return([]*v1.Container{container}, nil)
+		m.On("ContainerStatus", "cid1").Return(status, nil)
+		m.On("RemoveContainer", "cid1").Return(nil)
+		c := &containerd.Client{Runtime: m}
+		err := c.RemovePods()
+		assert.NoError(t, err)
+		m.AssertExpectations(t)
+	})
+}

internal/containerd/mocks/mock_runtime_service.go

@@ -0,0 +1,59 @@
+package mocks
+
+import (
+	internalapi "github.com/containerd/containerd/integration/cri-api/pkg/apis"
+	"github.com/stretchr/testify/mock"
+	"google.golang.org/grpc"
+	v1 "k8s.io/cri-api/pkg/apis/runtime/v1"
+)
+
+// MockRuntimeService is a testify mock for internalapi.RuntimeService
+// Only the methods used by RemovePods are implemented
+type MockRuntimeService struct {
+	mock.Mock
+	internalapi.RuntimeService
+}
+
+func (m *MockRuntimeService) ListPodSandbox(filter *v1.PodSandboxFilter, opts ...grpc.CallOption) ([]*v1.PodSandbox, error) {
+	args := m.Called(filter)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).([]*v1.PodSandbox), args.Error(1)
+}
+
+func (m *MockRuntimeService) StopPodSandbox(id string, opts ...grpc.CallOption) error {
+	args := m.Called(id)
+	return args.Error(0)
+}
+
+func (m *MockRuntimeService) RemovePodSandbox(id string, opts ...grpc.CallOption) error {
+	args := m.Called(id)
+	return args.Error(0)
+}
+
+func (m *MockRuntimeService) ListContainers(filter *v1.ContainerFilter, opts ...grpc.CallOption) ([]*v1.Container, error) {
+	args := m.Called(filter)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).([]*v1.Container), args.Error(1)
+}
+
+func (m *MockRuntimeService) ContainerStatus(id string, opts ...grpc.CallOption) (*v1.ContainerStatus, error) {
+	args := m.Called(id)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*v1.ContainerStatus), args.Error(1)
+}
+
+func (m *MockRuntimeService) StopContainer(id string, timeout int64, opts ...grpc.CallOption) error {
+	args := m.Called(id, timeout)
+	return args.Error(0)
+}
+
+func (m *MockRuntimeService) RemoveContainer(id string, opts ...grpc.CallOption) error {
+	args := m.Called(id)
+	return args.Error(0)
+}

internal/flows/uninstall.go

@@ -105,6 +105,14 @@ func (u *Uninstaller) uninstallDaemons(ctx context.Context) error {
 	}
 	if u.Artifacts.Containerd != tracker.ContainerdSourceNone {
 		u.Logger.Info("Uninstalling containerd...")
+		client, err := containerd.NewClient()
+		if err != nil {
+			u.Logger.Info("ignored error creating containerd client", zap.Error(err))
+		} else {
+			if err := client.RemovePods(); err != nil {
+				u.Logger.Info("ignored error stopping pods", zap.Error(err))
+			}
+		}
 		if err := u.DaemonManager.StopDaemon(containerd.ContainerdDaemonName); err != nil {
 			return err
 		}