Merge pull request containerd#9617 from abel-von/sandbox-plugin-0109

sandbox: use sandboxService in CRI plugin instead of calling controller API directly

Author: fuweid

integration/build_local_containerd_helper_test.go

@@ -129,7 +129,6 @@ func buildLocalContainerdClient(t *testing.T, tmpDir string, tweakInitFn tweakPl
 		containerd.WithDefaultNamespace(constants.K8sContainerdNamespace),
 		containerd.WithDefaultPlatform(platforms.Default()),
 		containerd.WithInMemoryServices(lastInitContext),
-		containerd.WithInMemorySandboxControllers(lastInitContext),
 	)
 	require.NoError(t, err)
 

internal/cri/config/config.go

@@ -277,7 +277,7 @@ type ImageConfig struct {
 	//   "base": "docker.io/library/ubuntu:latest"
 	// Migrated from:
 	// (PluginConfig).SandboxImage string `toml:"sandbox_image" json:"sandboxImage"`
-	PinnedImages map[string]string
+	PinnedImages map[string]string `toml:"pinned_images" json:"pinned_images"`
 
 	// RuntimePlatforms is map between the runtime and the image platform to
 	// use for that runtime. When resolving an image for a runtime, this

internal/cri/server/container_create.go

@@ -63,12 +63,7 @@ func (c *criService) CreateContainer(ctx context.Context, r *runtime.CreateConta
 		return nil, fmt.Errorf("failed to find sandbox id %q: %w", r.GetPodSandboxId(), err)
 	}
 
-	controller, err := c.sandboxService.SandboxController(sandbox.Config, sandbox.RuntimeHandler)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get sandbox controller: %w", err)
-	}
-
-	cstatus, err := controller.Status(ctx, sandbox.ID, false)
+	cstatus, err := c.sandboxService.SandboxStatus(ctx, sandbox.Sandboxer, sandbox.ID, false)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get controller status: %w", err)
 	}
@@ -150,7 +145,7 @@ func (c *criService) CreateContainer(ctx context.Context, r *runtime.CreateConta
 		}
 	}()
 
-	platform, err := controller.Platform(ctx, sandboxID)
+	platform, err := c.sandboxService.SandboxPlatform(ctx, sandbox.Sandboxer, sandboxID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to query sandbox platform: %w", err)
 	}

internal/cri/server/container_stats_list.go

@@ -68,15 +68,12 @@ func (c *criService) getMetricsHandler(ctx context.Context, sandboxID string) (m
 	if err != nil {
 		return nil, fmt.Errorf("failed to find sandbox id %q: %w", sandboxID, err)
 	}
-	controller, err := c.sandboxService.SandboxController(sandbox.Config, sandbox.RuntimeHandler)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get sandbox controller: %w", err)
-	}
+
 	// Grab the platform that this containers sandbox advertises. Reason being, even if
 	// the host may be {insert platform}, if it virtualizes or emulates a different platform
 	// it will return stats in that format, and we need to handle the conversion logic based
 	// off of this info.
-	p, err := controller.Platform(ctx, sandboxID)
+	p, err := c.sandboxService.SandboxPlatform(ctx, sandbox.Sandboxer, sandboxID)
 	if err != nil {
 		return nil, err
 	}

internal/cri/server/podsandbox/controller.go

@@ -158,10 +158,10 @@ func (c *Controller) Wait(ctx context.Context, sandboxID string) (sandbox.ExitSt
 
 }
 
-func (c *Controller) waitSandboxExit(ctx context.Context, p *types.PodSandbox, exitCh <-chan containerd.ExitStatus) (exitStatus uint32, exitedAt time.Time, err error) {
+func (c *Controller) waitSandboxExit(ctx context.Context, p *types.PodSandbox, exitCh <-chan containerd.ExitStatus) error {
 	select {
 	case e := <-exitCh:
-		exitStatus, exitedAt, err = e.Result()
+		exitStatus, exitedAt, err := e.Result()
 		if err != nil {
 			log.G(ctx).WithError(err).Errorf("failed to get task exit status for %q", p.ID)
 			exitStatus = unknownExitCode
@@ -171,12 +171,16 @@ func (c *Controller) waitSandboxExit(ctx context.Context, p *types.PodSandbox, e
 		dctx, dcancel := context.WithTimeout(dctx, handleEventTimeout)
 		defer dcancel()
 		event := &eventtypes.TaskExit{ExitStatus: exitStatus, ExitedAt: protobuf.ToTimestamp(exitedAt)}
-		if cleanErr := handleSandboxTaskExit(dctx, p, event); cleanErr != nil {
+		if err := handleSandboxTaskExit(dctx, p, event); err != nil {
+			// TODO will backoff the event to the controller's own EventMonitor, but not cri's,
+			// because we should call handleSandboxTaskExit again the next time
+			// eventMonitor handle this event. but now it goes into cri's EventMonitor,
+			// the handleSandboxTaskExit will not be called anymore
 			c.cri.BackOffEvent(p.ID, e)
 		}
-		return
+		return nil
 	case <-ctx.Done():
-		return unknownExitCode, time.Now(), ctx.Err()
+		return ctx.Err()
 	}
 }
 
@@ -196,5 +200,8 @@ func handleSandboxTaskExit(ctx context.Context, sb *types.PodSandbox, e *eventty
 			}
 		}
 	}
+	if err := sb.Exit(e.ExitStatus, protobuf.FromTimestamp(e.ExitedAt)); err != nil {
+		return err
+	}
 	return nil
 }

internal/cri/server/podsandbox/controller_test.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	containerd "github.com/containerd/containerd/v2/client"
 	criconfig "github.com/containerd/containerd/v2/internal/cri/config"
 	"github.com/containerd/containerd/v2/internal/cri/server/podsandbox/types"
 	sandboxstore "github.com/containerd/containerd/v2/internal/cri/store/sandbox"
@@ -63,10 +62,7 @@ func Test_Status(t *testing.T) {
 		CreatedAt: createdAt,
 	})
 	sb.Metadata = sandboxstore.Metadata{ID: sandboxID}
-	err := controller.store.Save(sb)
-	if err != nil {
-		t.Fatal(err)
-	}
+	assert.NoError(t, controller.store.Save(sb))
 	s, err := controller.Status(context.Background(), sandboxID, false)
 	if err != nil {
 		t.Fatal(err)
@@ -75,7 +71,8 @@ func Test_Status(t *testing.T) {
 	assert.Equal(t, s.CreatedAt, createdAt)
 	assert.Equal(t, s.State, sandboxstore.StateReady.String())
 
-	sb.Exit(*containerd.NewExitStatus(exitStatus, exitedAt, nil))
+	assert.NoError(t, sb.Exit(exitStatus, exitedAt))
+
 	exit, err := controller.Wait(context.Background(), sandboxID)
 	if err != nil {
 		t.Fatal(err)

internal/cri/server/podsandbox/recover.go

@@ -22,16 +22,17 @@ import (
 	goruntime "runtime"
 	"time"
 
+	"github.com/containerd/errdefs"
 	"github.com/containerd/log"
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
 
 	containerd "github.com/containerd/containerd/v2/client"
 	sandbox2 "github.com/containerd/containerd/v2/core/sandbox"
+	"github.com/containerd/containerd/v2/internal/cri/config"
 	"github.com/containerd/containerd/v2/internal/cri/server/podsandbox/types"
 	sandboxstore "github.com/containerd/containerd/v2/internal/cri/store/sandbox"
 	ctrdutil "github.com/containerd/containerd/v2/internal/cri/util"
 	"github.com/containerd/containerd/v2/pkg/netns"
-	"github.com/containerd/errdefs"
 )
 
 // loadContainerTimeout is the default timeout for loading a container/sandbox.
@@ -133,8 +134,9 @@ func (c *Controller) RecoverContainer(ctx context.Context, cntr containerd.Conta
 	}
 	if ch != nil {
 		go func() {
-			code, exitTime, err := c.waitSandboxExit(ctrdutil.NamespacedContext(), podSandbox, ch)
-			podSandbox.Exit(*containerd.NewExitStatus(code, exitTime, err))
+			if err := c.waitSandboxExit(ctrdutil.NamespacedContext(), podSandbox, ch); err != nil {
+				log.G(context.Background()).Warnf("failed to wait pod sandbox exit %v", err)
+			}
 		}()
 	}
 
@@ -144,6 +146,7 @@ func (c *Controller) RecoverContainer(ctx context.Context, cntr containerd.Conta
 
 	sandbox = sandboxstore.NewSandbox(*meta, s)
 	sandbox.Container = cntr
+	sandbox.Sandboxer = string(config.ModePodSandbox)
 
 	// Load network namespace.
 	sandbox.NetNS = getNetNS(meta)

internal/cri/server/podsandbox/recover_test.go

@@ -403,7 +403,7 @@ func TestRecoverContainer(t *testing.T) {
 
 		pSb := controller.store.Get(cont.ID())
 		assert.NotNil(t, pSb)
-		assert.Equal(t, c.expectedState, pSb.State, "%s state is not expected", cont.ID())
+		assert.Equal(t, c.expectedState, pSb.Status.Get().State, "%s state is not expected", cont.ID())
 
 		if c.expectedExitCode > 0 {
 			cont.t.waitExitCh <- struct{}{}

internal/cri/server/podsandbox/sandbox_run.go

@@ -216,7 +216,6 @@ func (c *Controller) Start(ctx context.Context, id string) (cin sandbox.Controll
 	if err != nil {
 		return cin, fmt.Errorf("failed to get sandbox container info: %w", err)
 	}
-	podSandbox.CreatedAt = info.CreatedAt
 
 	// Create sandbox task in containerd.
 	log.G(ctx).Tracef("Create sandbox container (id=%q, name=%q).", id, metadata.Name)
@@ -242,7 +241,6 @@ func (c *Controller) Start(ctx context.Context, id string) (cin sandbox.Controll
 			}
 		}
 	}()
-	podSandbox.Pid = task.Pid()
 
 	// wait is a long running background request, no timeout needed.
 	exitCh, err := task.Wait(ctrdutil.NamespacedContext())
@@ -267,16 +265,25 @@ func (c *Controller) Start(ctx context.Context, id string) (cin sandbox.Controll
 	if err := task.Start(ctx); err != nil {
 		return cin, fmt.Errorf("failed to start sandbox container task %q: %w", id, err)
 	}
-	podSandbox.State = sandboxstore.StateReady
+	pid := task.Pid()
+	if err := podSandbox.Status.Update(func(status sandboxstore.Status) (sandboxstore.Status, error) {
+		status.Pid = pid
+		status.State = sandboxstore.StateReady
+		status.CreatedAt = info.CreatedAt
+		return status, nil
+	}); err != nil {
+		return cin, fmt.Errorf("failed to update status of pod sandbox %q: %w", id, err)
+	}
 
 	cin.SandboxID = id
 	cin.Pid = task.Pid()
 	cin.CreatedAt = info.CreatedAt
 	cin.Labels = labels
 
 	go func() {
-		code, exitTime, err := c.waitSandboxExit(ctrdutil.NamespacedContext(), podSandbox, exitCh)
-		podSandbox.Exit(*containerd.NewExitStatus(code, exitTime, err))
+		if err := c.waitSandboxExit(ctrdutil.NamespacedContext(), podSandbox, exitCh); err != nil {
+			log.G(context.Background()).Warnf("failed to wait pod sandbox exit %v", err)
+		}
 	}()
 
 	return

internal/cri/server/podsandbox/sandbox_status.go

@@ -36,18 +36,15 @@ func (c *Controller) Status(ctx context.Context, sandboxID string, verbose bool)
 	if sb == nil {
 		return sandbox.ControllerStatus{}, fmt.Errorf("unable to find sandbox %q: %w", sandboxID, errdefs.ErrNotFound)
 	}
-
+	status := sb.Status.Get()
 	cstatus := sandbox.ControllerStatus{
 		SandboxID: sandboxID,
-		Pid:       sb.Pid,
-		State:     sb.State.String(),
-		CreatedAt: sb.CreatedAt,
+		Pid:       status.Pid,
+		State:     status.State.String(),
+		CreatedAt: status.CreatedAt,
+		ExitedAt:  status.ExitedAt,
 		Extra:     nil,
 	}
-	exitStatus := sb.GetExitStatus()
-	if exitStatus != nil {
-		cstatus.ExitedAt = exitStatus.ExitTime()
-	}
 
 	if verbose {
 		info, err := toCRISandboxInfo(ctx, sb)
@@ -64,7 +61,7 @@ func (c *Controller) Status(ctx context.Context, sandboxID string, verbose bool)
 // toCRISandboxInfo converts internal container object information to CRI sandbox status response info map.
 func toCRISandboxInfo(ctx context.Context, sb *types.PodSandbox) (map[string]string, error) {
 	si := &critypes.SandboxInfo{
-		Pid:            sb.Pid,
+		Pid:            sb.Status.Get().Pid,
 		Config:         sb.Metadata.Config,
 		RuntimeHandler: sb.Metadata.RuntimeHandler,
 		CNIResult:      sb.Metadata.CNIResult,