runtime: invoke Shutdown after every task deletion

TaskService.Shutdown may be invoked multiple times. A grouped shim
itself does not terminate while it still has active tasks. It terminates only
after receiving a Shutdown request when no active tasks remain.

The shim manager should invoke TaskService.Shutdown after deleting every task,
regardless of whether the task represents sandbox or not. This ensures that the
shim receives Shutdown when the last active task is deleted.

In CRI, the pause container uses oom_score_adj=-998, making it unlikely to be
OOM-killed. However, it may still exit or be killed before the other containers
in the sandbox. SandboxService.Shutdown and TaskService.Shutdown are independent
APIs. Sandbox service should make sure there is no active containers
before shutdown. So, for the task service, we should invoke Shutdown for
each task.

Removing the supportSandboxAPIVersion check does not change behavior for shims
using task API versions below v3. Those shims were already forced down the
non-sandboxed path and received Shutdown after every task deletion. The behavior
change applies only to sandboxed tasks using task API v3, which now receive
Shutdown after deletion as well.

Signed-off-by: Wei Fu <fuweid89@gmail.com>

core/runtime/v2/shim.go

@@ -197,11 +197,11 @@ func cleanupAfterDeadShim(ctx context.Context, id string, rt *runtime.NSMap[Shim
 // delete removes it), so callers that own one must remove it on error. The shim
 // map is untouched: callers reach this having already removed the task, or never
 // added it.
-func cleanupShimTask(ctx context.Context, st *shimTask, sandboxed bool) error {
+func cleanupShimTask(ctx context.Context, st *shimTask) error {
 	dctx, cancel := timeout.WithContext(context.WithoutCancel(ctx), cleanupTimeout)
 	defer cancel()
 
-	_, err := st.delete(dctx, sandboxed, func(context.Context, string) {})
+	_, err := st.delete(dctx, func(context.Context, string) {})
 	if err == nil {
 		return nil
 	}
@@ -579,7 +579,7 @@ func (s *shimTask) PID(ctx context.Context) (uint32, error) {
 	return response.TaskPid, nil
 }
 
-func (s *shimTask) delete(ctx context.Context, sandboxed bool, removeTask func(ctx context.Context, id string)) (*runtime.Exit, error) {
+func (s *shimTask) delete(ctx context.Context, removeTask func(ctx context.Context, id string)) (*runtime.Exit, error) {
 	response, shimErr := s.task.Delete(ctx, &task.DeleteRequest{
 		ID: s.ID(),
 	})
@@ -613,21 +613,12 @@ func (s *shimTask) delete(ctx context.Context, sandboxed bool, removeTask func(c
 		removeTask(ctx, s.ID())
 	}
 
-	const supportSandboxAPIVersion = 3
-	if _, apiVer := s.ShimInstance.Endpoint(); apiVer < supportSandboxAPIVersion {
-		sandboxed = false
-	}
-
-	// Don't shutdown sandbox as there may be other containers running.
-	// Let controller decide when to shutdown.
-	if !sandboxed {
-		if err := s.waitShutdown(ctx); err != nil {
-			// FIXME(fuweid):
-			//
-			// If the error is context canceled, should we use context.TODO()
-			// to wait for it?
-			log.G(ctx).WithField("id", s.ID()).WithError(err).Error("failed to shutdown shim task and the shim might be leaked")
-		}
+	if err := s.waitShutdown(ctx); err != nil {
+		// FIXME(fuweid):
+		//
+		// If the error is context canceled, should we use context.TODO()
+		// to wait for it?
+		log.G(ctx).WithField("id", s.ID()).WithError(err).Error("failed to shutdown shim task and the shim might be leaked")
 	}
 
 	if err := s.ShimInstance.Delete(ctx); err != nil {

core/runtime/v2/shim_load.go

@@ -202,7 +202,7 @@ func (m *ShimManager) loadShim(ctx context.Context, bundle *Bundle) error {
 			logEntry = logEntry.WithError(pidErr)
 		}
 		logEntry.Info("cleaning leaked shim process")
-		if err := cleanupShimTask(ctx, shim, false); err != nil && !errdefs.IsNotFound(err) {
+		if err := cleanupShimTask(ctx, shim); err != nil && !errdefs.IsNotFound(err) {
 			// Returning an error makes loadShims remove the bundle; a shim we
 			// cannot reap would otherwise be reloaded on every start.
 			return fmt.Errorf("failed to clean up leaked shim %q: %w", id, err)

core/runtime/v2/task_manager.go

@@ -253,8 +253,7 @@ func (m *TaskManager) Create(ctx context.Context, taskID string, opts runtime.Cr
 		// NOTE: ctx contains required namespace information.
 		m.manager.shims.Delete(ctx, taskID)
 
-		_ = cleanupShimTask(ctx, shimTask, opts.SandboxID != "")
-
+		_ = cleanupShimTask(ctx, shimTask)
 		return nil, fmt.Errorf("failed to create shim task: %w", err)
 	}
 
@@ -294,7 +293,7 @@ func (m *TaskManager) Delete(ctx context.Context, taskID string) (*runtime.Exit,
 		return nil, err
 	}
 
-	container, err := m.manager.containers.Get(ctx, taskID)
+	_, err = m.manager.containers.Get(ctx, taskID)
 	if err != nil {
 		return nil, err
 	}
@@ -304,9 +303,7 @@ func (m *TaskManager) Delete(ctx context.Context, taskID string) (*runtime.Exit,
 		return nil, err
 	}
 
-	sandboxed := container.SandboxID != ""
-
-	exit, err := shimTask.delete(ctx, sandboxed, func(ctx context.Context, id string) {
+	exit, err := shimTask.delete(ctx, func(ctx context.Context, id string) {
 		m.manager.shims.Delete(ctx, id)
 	})
 

docs/sandbox-api.md

@@ -109,6 +109,8 @@ sequenceDiagram
     loop for each container in sandbox
         containerd->>shim: TaskService.Kill / Delete
         shim-->>containerd: OK
+        containerd->>shim: TaskService.Shutdown
+        shim-->>containerd: OK (exit only if no tasks remain)
     end
     containerd->>containerd: SandboxController.Stop
     containerd->>shim: SandboxService.StopSandbox
@@ -127,6 +129,17 @@ sequenceDiagram
     containerd-->>kubelet: OK
 ```
 
+### Shutdown behavior for grouped shims
+
+containerd invokes `TaskService.Shutdown` after deleting every task. It may be
+invoked multiple times for a grouped shim. It does not mean that the shim must
+terminate immediately. The shim should return without terminating while it still
+has active tasks. It should terminate only after receiving `TaskService.Shutdown`
+when no active tasks remain.
+
+`SandboxService.ShutdownSandbox` is independent from `TaskService.Shutdown` and
+shuts down the sandbox instance.
+
 ## Controller Implementations
 
 There are two `Controller` implementations today:
@@ -147,4 +160,4 @@ containerd 1.7, and improves with every release.
 
 The Sandbox API was first introduced in containerd 1.7 as an experimental API and was promoted to stable in 2.0.
 It is still evolving; ongoing work can be tracked in
-[#9431](https://github.com/containerd/containerd/issues/9431).
+[#9431](https://github.com/containerd/containerd/issues/9431).

integration/issue7496_shutdown_linux_test.go

@@ -18,12 +18,17 @@ package integration
 
 import (
 	"context"
+	"syscall"
 	"testing"
+	"time"
 
+	"github.com/containerd/errdefs"
 	"github.com/stretchr/testify/require"
 
 	apitask "github.com/containerd/containerd/api/runtime/task/v3"
+	"github.com/containerd/containerd/v2/integration/images"
 	"github.com/containerd/containerd/v2/pkg/namespaces"
+	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
 )
 
 // TestIssue7496_ShouldRetryShutdown is based on https://github.com/containerd/containerd/issues/7496.
@@ -63,3 +68,97 @@ func TestIssue7496_ShouldRetryShutdown(t *testing.T) {
 	_, err = shimCli.Connect(ctx, &apitask.ConnectRequest{})
 	require.Error(t, err, "should failed to call shim connect API")
 }
+
+func TestShutdownShimWhenPauseExitsBeforeWorkload(t *testing.T) {
+	ctx := namespaces.WithNamespace(t.Context(), "k8s.io")
+
+	t.Logf("RunPodSandbox")
+	sbConfig := PodSandboxConfig("sandbox", t.Name(), WithHostNetwork)
+	sbID, err := runtimeService.RunPodSandbox(sbConfig, "")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = runtimeService.StopPodSandbox(sbID)
+		_ = runtimeService.RemovePodSandbox(sbID)
+	})
+
+	t.Logf("Connect to the shim %s", sbID)
+	shimCli := connectToShim(ctx, t, containerdEndpoint, 3, sbID)
+
+	testImage := images.Get(images.BusyBox)
+	EnsureImageExists(t, testImage)
+
+	t.Log("Create a container - sleep 1d")
+	containerName := "test-container"
+	cnConfig := ContainerConfig(
+		containerName,
+		testImage,
+		WithCommand("sh", "-c", "sleep 1d"),
+		WithPidNamespace(runtime.NamespaceMode_CONTAINER),
+	)
+	cnID, err := runtimeService.CreateContainer(sbID, cnConfig, sbConfig)
+	require.NoError(t, err)
+
+	t.Log("Start the container")
+	require.NoError(t, runtimeService.StartContainer(cnID))
+
+	t.Log("Load pause task and wait")
+	pauseContainer, err := containerdClient.LoadContainer(ctx, sbID)
+	require.NoError(t, err)
+	pauseTask, err := pauseContainer.Task(ctx, nil)
+	require.NoError(t, err)
+	pauseExitCh, err := pauseTask.Wait(ctx)
+	require.NoError(t, err)
+
+	t.Log("Load workload task and wait")
+	workloadContainer, err := containerdClient.LoadContainer(ctx, cnID)
+	require.NoError(t, err)
+	workloadTask, err := workloadContainer.Task(ctx, nil)
+	require.NoError(t, err)
+	workloadExitCh, err := workloadTask.Wait(ctx)
+	require.NoError(t, err)
+
+	t.Log("Kill pause container by containerd client API")
+	require.NoError(t, pauseTask.Kill(ctx, syscall.SIGKILL))
+
+	select {
+	case status := <-pauseExitCh:
+		pauseExitStatus, _, err := status.Result()
+		require.NoError(t, err)
+		require.Equal(t, uint32(137), pauseExitStatus)
+	case <-time.After(30 * time.Second):
+		t.Fatal("timed out waiting for pause task exit")
+	}
+
+	t.Log("Wait for pause task deletion")
+	require.NoError(t, Eventually(func() (bool, error) {
+		_, err := pauseContainer.Task(ctx, nil)
+		if err == nil {
+			return false, nil
+		}
+
+		if errdefs.IsNotFound(err) {
+			return true, nil
+		}
+		return false, err
+	}, time.Second, 30*time.Second))
+
+	t.Log("Stop sandbox and wait for workload")
+	require.NoError(t, runtimeService.StopPodSandbox(sbID))
+
+	select {
+	case status := <-workloadExitCh:
+		workloadExitStatus, _, err := status.Result()
+		require.NoError(t, err)
+		require.Equal(t, uint32(137), workloadExitStatus)
+	case <-time.After(30 * time.Second):
+		t.Fatal("timed out waiting for workload task exit")
+	}
+
+	t.Log("Remove sandbox")
+	require.NoError(t, runtimeService.RemovePodSandbox(sbID))
+
+	t.Log("Shim should be shutdown")
+	_, err = shimCli.Connect(ctx, &apitask.ConnectRequest{})
+	require.Error(t, err)
+	require.ErrorContains(t, err, "ttrpc: closed")
+}