From 24dc6900ec0081ec80eb837cbc751c3e32c5351e Mon Sep 17 00:00:00 2001 From: Wei Fu Date: Sun, 23 Aug 2026 12:12:04 -0400 Subject: [PATCH] runtime: make task.Delete API retriable To make task.Delete retriable, this patch caches the delete result in memory after the shim successfully deletes the task. If a later shutdown attempt fails, a subsequent Delete retry can complete cleanup and return the cached exit information. If containerd restarts after deleting the task, the in-memory cache is lost, and containerd will publish TaskExit and TaskDelete events again to subscribers. Currently, the exit status returned by shim-delete is always 137. In a follow-up, containerd-shim-runc-v2 should store the actual exit status in the state directory, so that shim-delete can reuse it instead of returning a hard-coded value. Implements: #8981 Signed-off-by: Wei Fu --- core/runtime/v2/shim.go | 82 ++++++++--- core/runtime/v2/task_manager.go | 17 ++- integration/issue7496_shutdown_linux_test.go | 147 +++++++++++++++++-- 3 files changed, 213 insertions(+), 33 deletions(-) diff --git a/core/runtime/v2/shim.go b/core/runtime/v2/shim.go index 056c46ec2d..c70d24d754 100644 --- a/core/runtime/v2/shim.go +++ b/core/runtime/v2/shim.go @@ -27,6 +27,7 @@ import ( "os" "path/filepath" "strings" + "sync/atomic" "time" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" @@ -152,9 +153,16 @@ func cleanupAfterDeadShim(ctx context.Context, id string, rt *runtime.NSMap[Shim log.G(ctx).WithError(err).WithField("id", id).Warn("failed to clean up after shim disconnected") } - if _, err := rt.Get(ctx, id); err != nil { - // Task was never started or was already successfully deleted - // No need to publish events + s, err := rt.Get(ctx, id) + if err != nil { + // Task was never started, or its record has already been removed. + // No need to publish events. + return + } + + // If the task delete already succeeded, the shim itself has delivered the + // exit and delete events. No need to publish duplicates. + if s.(taskDeleteState).taskDeleteResult() != nil { return } @@ -274,6 +282,14 @@ type clientVersionDowngrader interface { Downgrade() error } +// taskDeleteState caches the result of a successful task delete on the shim +// instance, so that a Delete retried after a failed shutdown can still return +// the original exit. +type taskDeleteState interface { + recordTaskDeleteResult(*runtime.Exit) + taskDeleteResult() *runtime.Exit +} + func parseStartResponse(response []byte) (*bootapi.BootstrapResult, error) { var result bootapi.BootstrapResult @@ -469,11 +485,14 @@ type shim struct { // rather than decoded into fields here so that a new capability needs no // further plumbing through the shim instance. bootstrap *bootapi.BootstrapResult + + taskDeleteExit atomic.Pointer[runtime.Exit] } var _ ShimInstance = (*shim)(nil) var _ clientVersionDowngrader = (*shim)(nil) var _ shimCapabilities = (*shim)(nil) +var _ taskDeleteState = (*shim)(nil) // BootstrapResult returns what the shim advertised when it started. func (s *shim) BootstrapResult() *bootapi.BootstrapResult { @@ -554,6 +573,25 @@ func (s *shim) Delete(ctx context.Context) error { return errors.Join(result...) } +// recordTaskDeleteResult caches the result of a successful task delete. The +// value is copied both in and out so that neither the caller that recorded it +// nor a later retry can mutate the cached result. +func (s *shim) recordTaskDeleteResult(exit *runtime.Exit) { + cached := *exit + s.taskDeleteExit.Store(&cached) +} + +// taskDeleteResult returns a copy of the cached delete result, or nil if this +// containerd process has never deleted the task successfully. +func (s *shim) taskDeleteResult() *runtime.Exit { + cached := s.taskDeleteExit.Load() + if cached == nil { + return nil + } + exit := *cached + return &exit +} + var _ runtime.Task = &shimTask{} // shimTask wraps shim process and adds task service client for compatibility with existing shim manager. @@ -617,6 +655,8 @@ func (s *shimTask) delete(ctx context.Context, removeTask func(ctx context.Conte } } + deleteState := s.ShimInstance.(taskDeleteState) + // NOTE: If the shim has been killed and ttrpc connection has been // closed, the shimErr will not be nil. For this case, the event // subscriber, like moby/moby, might have received the exit or delete @@ -624,8 +664,8 @@ func (s *shimTask) delete(ctx context.Context, removeTask func(ctx context.Conte // send the exit and delete events again. And the exit status will // depend on result of shimV2.Delete. // - // If not, the shim has been delivered the exit and delete events. - // So we should remove the record and prevent duplicate events from + // If not, the shim has delivered the exit and delete events. Cache the + // delete result so a retry can return it and prevent duplicate events from // ttrpc-callback-on-close. // // TODO: It's hard to guarantee that the event is unique and sent only @@ -633,16 +673,22 @@ func (s *shimTask) delete(ctx context.Context, removeTask func(ctx context.Conte // only one exit event. The moby/moby should handle the duplicate events. // // REF: https://github.com/containerd/containerd/issues/4769 + var exit *runtime.Exit if shimErr == nil { - removeTask(ctx, s.ID()) + exit = &runtime.Exit{ + Status: response.ExitStatus, + Timestamp: protobuf.FromTimestamp(response.ExitedAt), + Pid: response.Pid, + } + deleteState.recordTaskDeleteResult(exit) } + // NOTE: Returning here deliberately leaves the shim record, the ttrpc + // client and the bundle in place, so that the caller can retry Delete. + // The result cached above lets that retry return the original exit, even + // though the shim reports NotFound for the task by then. 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") + return nil, fmt.Errorf("failed to invoke shutdown: %w", err) } if err := s.ShimInstance.Delete(ctx); err != nil { @@ -653,15 +699,15 @@ func (s *shimTask) delete(ctx context.Context, removeTask func(ctx context.Conte // this seems dirty but it cleans up the API across runtimes, tasks, and the service removeTask(ctx, s.ID()) - if shimErr != nil { + if exit == nil { + // An earlier attempt already deleted the task in the shim, so shimErr + // is NotFound. Return the exit recorded by that attempt instead. + exit = deleteState.taskDeleteResult() + } + if exit == nil { return nil, shimErr } - - return &runtime.Exit{ - Status: response.ExitStatus, - Timestamp: protobuf.FromTimestamp(response.ExitedAt), - Pid: response.Pid, - }, nil + return exit, nil } func (s *shimTask) Create(ctx context.Context, opts runtime.CreateOpts) (runtime.Task, error) { diff --git a/core/runtime/v2/task_manager.go b/core/runtime/v2/task_manager.go index 743092a951..a6c42814f4 100644 --- a/core/runtime/v2/task_manager.go +++ b/core/runtime/v2/task_manager.go @@ -320,14 +320,25 @@ func (m *TaskManager) Delete(ctx context.Context, taskID string) (*runtime.Exit, m.manager.shims.Delete(ctx, id) }) - if err != nil { + // An ErrNotFound here means the shim has no record of the task and there + // was no cached delete result to fall back to. For example, the task was + // never created successfully in the shim, or a previous containerd process + // deleted it. The runtime side has still been cleaned up, so we should + // deactivate the mounts before returning the error. + if err != nil && !errdefs.IsNotFound(err) { return nil, fmt.Errorf("failed to delete task: %w", err) } - if err := m.taskMounts.Deactivate(ctx, taskID); err != nil && !errdefs.IsNotFound(err) { - log.G(ctx).WithError(err).WithField("task", taskID).Errorf("failed to deactivate mounts") + // FIXME(fuweid): It seems that cleaning this up is best-effort because + // GC can guarantee that the mount is deleted when the container is deleted. + // What if we reuse the container and restart the task? + if merr := m.taskMounts.Deactivate(ctx, taskID); merr != nil && !errdefs.IsNotFound(merr) { + log.G(ctx).WithError(merr).WithField("task", taskID).Errorf("failed to deactivate mounts") } + if err != nil { + return nil, fmt.Errorf("failed to delete task: %w", err) + } return exit, nil } diff --git a/integration/issue7496_shutdown_linux_test.go b/integration/issue7496_shutdown_linux_test.go index 0a4e7baa5b..134ba0814e 100644 --- a/integration/issue7496_shutdown_linux_test.go +++ b/integration/issue7496_shutdown_linux_test.go @@ -17,36 +17,37 @@ package integration import ( - "context" + "fmt" "syscall" "testing" "time" "github.com/containerd/errdefs" + "github.com/containerd/typeurl/v2" "github.com/stretchr/testify/require" + eventtypes "github.com/containerd/containerd/api/events" apitask "github.com/containerd/containerd/api/runtime/task/v3" + "github.com/containerd/containerd/v2/core/events" + ctrdruntime "github.com/containerd/containerd/v2/core/runtime" "github.com/containerd/containerd/v2/integration/images" "github.com/containerd/containerd/v2/pkg/namespaces" + ptypes "github.com/containerd/containerd/v2/pkg/protobuf/types" runtime "k8s.io/cri-api/pkg/apis/runtime/v1" ) // TestIssue7496_ShouldRetryShutdown is based on https://github.com/containerd/containerd/issues/7496. // -// Assume that the shim.Delete takes almost 10 seconds and returns successfully -// and there is no container in shim. However, the context is close to be -// canceled. It will fail to call Shutdown. If we ignores the Canceled error, -// the shim will be leaked. In order to reproduce this, this case will use -// failpoint to inject error into Shutdown API, and then check whether the shim -// is leaked. +// The first task Delete succeeds, but an injected Shutdown failure prevents the +// shim from exiting. The CRI retry then sees the task as NotFound and must still +// shut down the shim without publishing duplicate exit or delete events. func TestIssue7496_ShouldRetryShutdown(t *testing.T) { - // TODO: re-enable if we can retry Shutdown API. - t.Skipf("Please re-enable me if we can retry Shutdown API") + const eventBarrierTopic = "/tests/issue7496/event-barrier" - ctx := namespaces.WithNamespace(context.Background(), "k8s.io") + ctx := namespaces.WithNamespace(t.Context(), "k8s.io") t.Logf("Create a pod config with shutdown failpoint") - sbConfig := PodSandboxConfig("sandbox", "issue7496_shouldretryshutdown") + sbConfig := PodSandboxConfig("sandbox", "issue7496_shouldretryshutdown", WithHostNetwork) injectShimFailpoint(t, sbConfig, map[string]string{ "Shutdown": "1*error(please retry)", }) @@ -60,13 +61,135 @@ func TestIssue7496_ShouldRetryShutdown(t *testing.T) { t.Logf("Log shim %s's pid: %d", sbID, shimPid(ctx, t, shimCli)) + t.Logf("Subscribe task exit/delete events") + eventCh, eventErrCh := containerdClient.EventService().Subscribe(ctx, + fmt.Sprintf(`topic=="%s",event.container_id=="%s"`, ctrdruntime.TaskExitEventTopic, sbID), + fmt.Sprintf(`topic=="%s",event.container_id=="%s"`, ctrdruntime.TaskDeleteEventTopic, sbID), + fmt.Sprintf(`topic=="%s"`, eventBarrierTopic), + ) + t.Logf("StopPodSandbox and RemovePodSandbox") require.NoError(t, runtimeService.StopPodSandbox(sbID)) require.NoError(t, runtimeService.RemovePodSandbox(sbID)) t.Logf("Check the shim connection") _, err = shimCli.Connect(ctx, &apitask.ConnectRequest{}) - require.Error(t, err, "should failed to call shim connect API") + require.Error(t, err, "should fail to call the shim Connect API") + require.ErrorContains(t, err, "ttrpc: closed") + + // The task delete succeeded on the first attempt, so the shim itself has + // already delivered the exit and delete events. When the retry finally + // shuts the shim down, ttrpc-callback-on-close must not publish them a + // second time. + // + // REF: https://github.com/containerd/containerd/issues/4769 + t.Logf("Check that task exit/delete events are not duplicated") + require.NoError(t, containerdClient.EventService().Publish(ctx, + eventBarrierTopic, &ptypes.Empty{}, + )) + exits, deletes := countTaskExitDeleteEventsUntilBarrier( + t, sbID, eventBarrierTopic, eventCh, eventErrCh, 10*time.Second, + ) + require.Equal(t, 1, exits, "task exit event should be published only once") + require.Equal(t, 1, deletes, "task delete event should be published only once") +} + +func TestKillShimPublishesTaskExitAndDeleteEvents(t *testing.T) { + ctx := namespaces.WithNamespace(t.Context(), "k8s.io") + + sbConfig := PodSandboxConfig("sandbox", t.Name(), WithHostNetwork) + sbID, err := runtimeService.RunPodSandbox(sbConfig, "") + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, runtimeService.StopPodSandbox(sbID)) + require.NoError(t, runtimeService.RemovePodSandbox(sbID)) + }) + + shimCli := connectToShim(ctx, t, containerdEndpoint, 3, sbID) + shimPID := shimPid(ctx, t, shimCli) + + eventCh, eventErrCh := containerdClient.EventService().Subscribe(ctx, + fmt.Sprintf(`topic=="%s",event.container_id=="%s"`, ctrdruntime.TaskExitEventTopic, sbID), + fmt.Sprintf(`topic=="%s",event.container_id=="%s"`, ctrdruntime.TaskDeleteEventTopic, sbID), + ) + + require.NoError(t, syscall.Kill(int(shimPID), syscall.SIGKILL)) + waitForTaskExitDeleteEvents(t, sbID, eventCh, eventErrCh, 10*time.Second) +} + +func waitForTaskExitDeleteEvents(t *testing.T, id string, + ch <-chan *events.Envelope, errCh <-chan error, timeout time.Duration, +) { + t.Helper() + + pending := map[string]struct{}{ + ctrdruntime.TaskExitEventTopic: {}, + ctrdruntime.TaskDeleteEventTopic: {}, + } + timer := time.NewTimer(timeout) + defer timer.Stop() + + for len(pending) != 0 { + select { + case <-timer.C: + t.Fatalf("timed out waiting for task exit/delete events for %q", id) + case err, ok := <-errCh: + if !ok { + t.Fatalf("event subscription closed while waiting for %q", id) + } + require.NoError(t, err, "event subscription failed") + case env, ok := <-ch: + require.True(t, ok, "event subscription closed while waiting for %q", id) + t.Logf("received event %q for %q", env.Topic, id) + delete(pending, env.Topic) + } + } +} + +// countTaskExitDeleteEventsUntilBarrier counts task exit and delete events for +// the given container ID until the event stream reaches barrierTopic. The +// barrier is published after shim deletion has waited for its close callback, +// so any duplicate events from cleanupAfterDeadShim are ordered before it. +func countTaskExitDeleteEventsUntilBarrier(t *testing.T, id, barrierTopic string, + ch <-chan *events.Envelope, errCh <-chan error, timeout time.Duration, +) (exits, deletes int) { + t.Helper() + + timer := time.NewTimer(timeout) + defer timer.Stop() + + for { + select { + case <-timer.C: + t.Fatalf("timed out waiting for event barrier %q", barrierTopic) + case err, ok := <-errCh: + if !ok { + t.Fatalf("event subscription closed before barrier %q", barrierTopic) + } + require.NoError(t, err, "event subscription failed") + case env, ok := <-ch: + require.True(t, ok, "event subscription closed before barrier %q", barrierTopic) + if env.Topic == barrierTopic { + return exits, deletes + } + + evt, err := typeurl.UnmarshalAny(env.Event) + require.NoError(t, err, "failed to unmarshal event") + + switch e := evt.(type) { + case *eventtypes.TaskExit: + if e.ContainerID == id { + t.Logf("received task exit event: %+v", e) + exits++ + } + case *eventtypes.TaskDelete: + if e.ContainerID == id { + t.Logf("received task delete event: %+v", e) + deletes++ + } + } + } + } } func TestShutdownShimWhenPauseExitsBeforeWorkload(t *testing.T) {