diff --git a/core/runtime/v2/shim.go b/core/runtime/v2/shim.go index b88c9222aae53..1f13f096e2651 100644 --- a/core/runtime/v2/shim.go +++ b/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 { diff --git a/core/runtime/v2/shim_load.go b/core/runtime/v2/shim_load.go index 80bdea8ae5d45..acc62b718d48d 100644 --- a/core/runtime/v2/shim_load.go +++ b/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) diff --git a/core/runtime/v2/task_manager.go b/core/runtime/v2/task_manager.go index 03e1f65ce906b..fbc2291b16cf3 100644 --- a/core/runtime/v2/task_manager.go +++ b/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) }) diff --git a/docs/sandbox-api.md b/docs/sandbox-api.md index 0bfdc2277c3f6..64fc56006c6f4 100644 --- a/docs/sandbox-api.md +++ b/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). \ No newline at end of file diff --git a/integration/issue7496_shutdown_linux_test.go b/integration/issue7496_shutdown_linux_test.go index 7b32aa283c49c..0a4e7baa5b25d 100644 --- a/integration/issue7496_shutdown_linux_test.go +++ b/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") +}