sandbox: add methods to sandboxService

abel-von · abel-von · commit 0f1d27412f23 · 2024-02-26T10:10:11.000+08:00
so that we cri service don't have to get sandbox controller everytime it
needs to call sandbox controller api.

Signed-off-by: Abel Feng &lt;fshb1988@gmail.com&gt;
diff --git a/integration/build_local_containerd_helper_test.go b/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)
 
diff --git a/internal/cri/server/container_create.go b/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)
 	}
diff --git a/internal/cri/server/container_stats_list.go b/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
 	}
diff --git a/internal/cri/server/podsandbox/recover.go b/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.
@@ -144,6 +145,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)
diff --git a/internal/cri/server/restart.go b/internal/cri/server/restart.go
@@ -58,8 +58,10 @@ func (c *criService) recover(ctx context.Context) error {
 		return fmt.Errorf("failed to list sandbox containers: %w", err)
 	}
 
-	podSandboxController := c.client.SandboxController(string(criconfig.ModePodSandbox))
-
+	podSandboxController, err := c.sandboxService.SandboxController(string(criconfig.ModePodSandbox))
+	if err != nil {
+		return fmt.Errorf("failed to get podsanbox controller %v", err)
+	}
 	podSandboxLoader, ok := podSandboxController.(podSandboxRecover)
 	if !ok {
 		log.G(ctx).Fatal("pod sandbox controller doesn't support recovery")
@@ -134,6 +136,7 @@ func (c *criService) recover(ctx context.Context) error {
 		}
 
 		sb := sandboxstore.NewSandbox(metadata, sandboxstore.Status{State: state})
+		sb.Sandboxer = sbx.Sandboxer
 
 		// Load network namespace.
 		sb.NetNS = getNetNS(&metadata)
@@ -149,20 +152,11 @@ func (c *criService) recover(ctx context.Context) error {
 		if status.State == sandboxstore.StateNotReady {
 			continue
 		}
-		controller, err := c.sandboxService.SandboxController(sb.Config, sb.RuntimeHandler)
+		exitCh, err := c.sandboxService.WaitSandbox(ctrdutil.NamespacedContext(), sb.Sandboxer, sb.ID)
 		if err != nil {
-			log.G(ctx).WithError(err).Error("failed to get sandbox controller while waiting sandbox")
+			log.G(ctx).WithError(err).Error("failed to wait sandbox")
 			continue
 		}
-		exitCh := make(chan containerd.ExitStatus, 1)
-		go func() {
-			exit, err := controller.Wait(ctrdutil.NamespacedContext(), sb.ID)
-			if err != nil {
-				log.G(ctx).WithError(err).Error("failed to wait for sandbox exit")
-				exitCh <- *containerd.NewExitStatus(containerd.UnknownExitStatus, time.Time{}, err)
-			}
-			exitCh <- *containerd.NewExitStatus(exit.ExitStatus, exit.ExitedAt, nil)
-		}()
 		c.eventMonitor.startSandboxExitMonitor(context.Background(), sb.ID, exitCh)
 	}
 	// Recover all containers.
diff --git a/internal/cri/server/sandbox_remove.go b/internal/cri/server/sandbox_remove.go
@@ -79,13 +79,7 @@ func (c *criService) RemovePodSandbox(ctx context.Context, r *runtime.RemovePodS
 		}
 	}
 
-	// Use sandbox controller to delete sandbox
-	controller, err := c.sandboxService.SandboxController(sandbox.Config, sandbox.RuntimeHandler)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get sandbox controller: %w", err)
-	}
-
-	if err := controller.Shutdown(ctx, id); err != nil && !errdefs.IsNotFound(err) {
+	if err := c.sandboxService.ShutdownSandbox(ctx, sandbox.Sandboxer, id); err != nil && !errdefs.IsNotFound(err) {
 		return nil, fmt.Errorf("failed to delete sandbox %q: %w", id, err)
 	}
 
diff --git a/internal/cri/server/sandbox_run.go b/internal/cri/server/sandbox_run.go
@@ -31,7 +31,6 @@ import (
 	"github.com/containerd/typeurl/v2"
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
 
-	containerd "github.com/containerd/containerd/v2/client"
 	sb "github.com/containerd/containerd/v2/core/sandbox"
 	"github.com/containerd/containerd/v2/internal/cri/annotations"
 	"github.com/containerd/containerd/v2/internal/cri/bandwidth"
@@ -124,6 +123,7 @@ func (c *criService) RunPodSandbox(ctx context.Context, r *runtime.RunPodSandbox
 			CreatedAt: time.Now().UTC(),
 		},
 	)
+	sandbox.Sandboxer = ociRuntime.Sandboxer
 
 	if _, err := c.client.SandboxStore().Create(ctx, sandboxInfo); err != nil {
 		return nil, fmt.Errorf("failed to save sandbox metadata: %w", err)
@@ -240,21 +240,16 @@ func (c *criService) RunPodSandbox(ctx context.Context, r *runtime.RunPodSandbox
 		return nil, fmt.Errorf("unable to save sandbox %q to store: %w", id, err)
 	}
 
-	controller, err := c.sandboxService.SandboxController(config, r.GetRuntimeHandler())
-	if err != nil {
-		return nil, fmt.Errorf("failed to get sandbox controller: %w", err)
-	}
-
 	// Save sandbox metadata to store
 	if sandboxInfo, err = c.client.SandboxStore().Update(ctx, sandboxInfo, "extensions"); err != nil {
 		return nil, fmt.Errorf("unable to update extensions for sandbox %q: %w", id, err)
 	}
 
-	if err := controller.Create(ctx, sandboxInfo, sb.WithOptions(config), sb.WithNetNSPath(sandbox.NetNSPath)); err != nil {
+	if err := c.sandboxService.CreateSandbox(ctx, sandboxInfo, sb.WithOptions(config), sb.WithNetNSPath(sandbox.NetNSPath)); err != nil {
 		return nil, fmt.Errorf("failed to create sandbox %q: %w", id, err)
 	}
 
-	ctrl, err := controller.Start(ctx, id)
+	ctrl, err := c.sandboxService.StartSandbox(ctx, sandbox.Sandboxer, id)
 	if err != nil {
 		var cerr podsandbox.CleanupErr
 		if errors.As(err, &cerr) {
@@ -401,21 +396,10 @@ func (c *criService) RunPodSandbox(ctx context.Context, r *runtime.RunPodSandbox
 	// SandboxStatus from the store and include it in the event.
 	c.generateAndSendContainerEvent(ctx, id, id, runtime.ContainerEventType_CONTAINER_CREATED_EVENT)
 
-	// TODO: Use sandbox client instead
-	exitCh := make(chan containerd.ExitStatus, 1)
-	go func() {
-		defer close(exitCh)
-
-		ctx := util.NamespacedContext()
-		resp, err := controller.Wait(ctx, id)
-		if err != nil {
-			log.G(ctx).WithError(err).Error("failed to wait for sandbox exit")
-			exitCh <- *containerd.NewExitStatus(containerd.UnknownExitStatus, time.Time{}, err)
-			return
-		}
-
-		exitCh <- *containerd.NewExitStatus(resp.ExitStatus, resp.ExitedAt, nil)
-	}()
+	exitCh, err := c.sandboxService.WaitSandbox(util.NamespacedContext(), sandbox.Sandboxer, id)
+	if err != nil {
+		return nil, fmt.Errorf("failed to wait sandbox %s: %v", id, err)
+	}
 
 	// start the monitor after adding sandbox into the store, this ensures
 	// that sandbox is in the store, when event monitor receives the TaskExit event.
diff --git a/internal/cri/server/sandbox_service.go b/internal/cri/server/sandbox_service.go
@@ -17,31 +17,103 @@
 package server
 
 import (
+	"context"
 	"fmt"
+	"time"
 
-	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
+	"github.com/containerd/platforms"
 
 	"github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/core/sandbox"
 	criconfig "github.com/containerd/containerd/v2/internal/cri/config"
 )
 
 type criSandboxService struct {
-	cli    *client.Client
-	config *criconfig.Config
+	sandboxControllers map[string]sandbox.Controller
+	config             *criconfig.Config
 }
 
-func newCriSandboxService(config *criconfig.Config, c *client.Client) *criSandboxService {
+func newCriSandboxService(config *criconfig.Config, sandboxers map[string]sandbox.Controller) *criSandboxService {
 	return &criSandboxService{
-		cli:    c,
-		config: config,
+		sandboxControllers: sandboxers,
+		config:             config,
 	}
 }
 
-func (c *criSandboxService) SandboxController(config *runtime.PodSandboxConfig, runtimeHandler string) (sandbox.Controller, error) {
-	ociRuntime, err := c.config.GetSandboxRuntime(config, runtimeHandler)
+func (c *criSandboxService) SandboxController(sandboxer string) (sandbox.Controller, error) {
+	sbController, ok := c.sandboxControllers[sandboxer]
+	if !ok {
+		return nil, fmt.Errorf("failed to get sandbox controller by %s", sandboxer)
+	}
+	return sbController, nil
+}
+
+func (c *criSandboxService) CreateSandbox(ctx context.Context, info sandbox.Sandbox, opts ...sandbox.CreateOpt) error {
+	ctrl, err := c.SandboxController(info.Sandboxer)
+	if err != nil {
+		return err
+	}
+	return ctrl.Create(ctx, info, opts...)
+}
+
+func (c *criSandboxService) StartSandbox(ctx context.Context, sandboxer string, sandboxID string) (sandbox.ControllerInstance, error) {
+	ctrl, err := c.SandboxController(sandboxer)
+	if err != nil {
+		return sandbox.ControllerInstance{}, err
+	}
+	return ctrl.Start(ctx, sandboxID)
+}
+
+func (c *criSandboxService) WaitSandbox(ctx context.Context, sandboxer string, sandboxID string) (<-chan client.ExitStatus, error) {
+	ctrl, err := c.SandboxController(sandboxer)
+	if err != nil {
+		return nil, err
+	}
+
+	ch := make(chan client.ExitStatus, 1)
+	go func() {
+		defer close(ch)
+
+		exitStatus, err := ctrl.Wait(ctx, sandboxID)
+		if err != nil {
+			ch <- *client.NewExitStatus(client.UnknownExitStatus, time.Time{}, err)
+			return
+		}
+
+		ch <- *client.NewExitStatus(exitStatus.ExitStatus, exitStatus.ExitedAt, nil)
+	}()
+
+	return ch, nil
+}
+
+func (c *criSandboxService) SandboxStatus(ctx context.Context, sandboxer string, sandboxID string, verbose bool) (sandbox.ControllerStatus, error) {
+	ctrl, err := c.SandboxController(sandboxer)
+	if err != nil {
+		return sandbox.ControllerStatus{}, err
+	}
+	return ctrl.Status(ctx, sandboxID, verbose)
+}
+
+func (c *criSandboxService) SandboxPlatform(ctx context.Context, sandboxer string, sandboxID string) (platforms.Platform, error) {
+	ctrl, err := c.SandboxController(sandboxer)
+	if err != nil {
+		return platforms.Platform{}, err
+	}
+	return ctrl.Platform(ctx, sandboxID)
+}
+
+func (c *criSandboxService) ShutdownSandbox(ctx context.Context, sandboxer string, sandboxID string) error {
+	ctrl, err := c.SandboxController(sandboxer)
+	if err != nil {
+		return err
+	}
+	return ctrl.Shutdown(ctx, sandboxID)
+}
+
+func (c *criSandboxService) StopSandbox(ctx context.Context, sandboxer, sandboxID string, opts ...sandbox.StopOpt) error {
+	ctrl, err := c.SandboxController(sandboxer)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get sandbox runtime: %w", err)
+		return err
 	}
-	return c.cli.SandboxController(ociRuntime.Sandboxer), nil
+	return ctrl.Stop(ctx, sandboxID, opts...)
 }
diff --git a/internal/cri/server/sandbox_status.go b/internal/cri/server/sandbox_status.go
@@ -41,17 +41,12 @@ func (c *criService) PodSandboxStatus(ctx context.Context, r *runtime.PodSandbox
 		return nil, fmt.Errorf("failed to get sandbox ip: %w", err)
 	}
 
-	controller, err := c.sandboxService.SandboxController(sandbox.Config, sandbox.RuntimeHandler)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get sandbox controller: %w", err)
-	}
-
 	var (
 		createdAt time.Time
 		state     string
 		info      map[string]string
 	)
-	cstatus, err := controller.Status(ctx, sandbox.ID, r.GetVerbose())
+	cstatus, err := c.sandboxService.SandboxStatus(ctx, sandbox.Sandboxer, sandbox.ID, r.GetVerbose())
 	if err != nil {
 		// If the shim died unexpectedly (segfault etc.) let's set the state as
 		// NOTREADY and not just error out to make k8s and clients like crictl
diff --git a/internal/cri/server/sandbox_stop.go b/internal/cri/server/sandbox_stop.go
@@ -68,13 +68,7 @@ func (c *criService) stopPodSandbox(ctx context.Context, sandbox sandboxstore.Sa
 	// Only stop sandbox container when it's running or unknown.
 	state := sandbox.Status.Get().State
 	if state == sandboxstore.StateReady || state == sandboxstore.StateUnknown {
-		// Use sandbox controller to stop sandbox
-		controller, err := c.sandboxService.SandboxController(sandbox.Config, sandbox.RuntimeHandler)
-		if err != nil {
-			return fmt.Errorf("failed to get sandbox controller: %w", err)
-		}
-
-		if err := controller.Stop(ctx, id); err != nil {
+		if err := c.sandboxService.StopSandbox(ctx, sandbox.Sandboxer, id); err != nil {
 			// Log and ignore the error if controller already removed the sandbox
 			if errdefs.IsNotFound(err) {
 				log.G(ctx).Warnf("sandbox %q is not found when stopping it", id)
diff --git a/internal/cri/server/service.go b/internal/cri/server/service.go
@@ -28,9 +28,9 @@ import (
 
 	"github.com/containerd/go-cni"
 	"github.com/containerd/log"
+	"github.com/containerd/platforms"
 	"github.com/containerd/typeurl/v2"
 	"github.com/opencontainers/runtime-spec/specs-go/features"
-
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
 	"k8s.io/kubelet/pkg/cri/streaming"
 
@@ -74,7 +74,14 @@ type CRIService interface {
 }
 
 type sandboxService interface {
-	SandboxController(config *runtime.PodSandboxConfig, runtimeHandler string) (sandbox.Controller, error)
+	CreateSandbox(ctx context.Context, info sandbox.Sandbox, opts ...sandbox.CreateOpt) error
+	StartSandbox(ctx context.Context, sandboxer string, sandboxID string) (sandbox.ControllerInstance, error)
+	WaitSandbox(ctx context.Context, sandboxer string, sandboxID string) (<-chan containerd.ExitStatus, error)
+	StopSandbox(ctx context.Context, sandboxer, sandboxID string, opts ...sandbox.StopOpt) error
+	ShutdownSandbox(ctx context.Context, sandboxer string, sandboxID string) error
+	SandboxStatus(ctx context.Context, sandboxer string, sandboxID string, verbose bool) (sandbox.ControllerStatus, error)
+	SandboxPlatform(ctx context.Context, sandboxer string, sandboxID string) (platforms.Platform, error)
+	SandboxController(sandboxer string) (sandbox.Controller, error)
 }
 
 // RuntimeService specifies dependencies to runtime service which provides
@@ -188,7 +195,7 @@ func NewCRIService(options *CRIServiceOptions) (CRIService, runtime.RuntimeServi
 		sandboxNameIndex:   registrar.NewRegistrar(),
 		containerNameIndex: registrar.NewRegistrar(),
 		netPlugin:          make(map[string]cni.CNI),
-		sandboxService:     newCriSandboxService(&config, options.Client),
+		sandboxService:     newCriSandboxService(&config, options.SandboxControllers),
 	}
 
 	// TODO: Make discard time configurable
@@ -231,8 +238,7 @@ func NewCRIService(options *CRIServiceOptions) (CRIService, runtime.RuntimeServi
 	}
 
 	// Initialize pod sandbox controller
-	// TODO: Get this from options, NOT client
-	podSandboxController := options.Client.SandboxController(string(criconfig.ModePodSandbox)).(*podsandbox.Controller)
+	podSandboxController := options.SandboxControllers[string(criconfig.ModePodSandbox)].(*podsandbox.Controller)
 	podSandboxController.Init(c)
 
 	c.nri = options.NRI
diff --git a/internal/cri/server/service_test.go b/internal/cri/server/service_test.go
diff --git a/internal/cri/store/sandbox/sandbox.go b/internal/cri/store/sandbox/sandbox.go
diff --git a/plugins/cri/cri.go b/plugins/cri/cri.go

Original file line number	Diff line number	Diff line change
`@@ -129,7 +129,6 @@ func buildLocalContainerdClient(t *testing.T, tmpDir string, tweakInitFn tweakPl`
`129`	`129`	`containerd.WithDefaultNamespace(constants.K8sContainerdNamespace),`
`130`	`130`	`containerd.WithDefaultPlatform(platforms.Default()),`
`131`	`131`	`containerd.WithInMemoryServices(lastInitContext),`
`132`		`- containerd.WithInMemorySandboxControllers(lastInitContext),`
`133`	`132`	`)`
`134`	`133`	`require.NoError(t, err)`
`135`	`134`
Original file line number	Diff line number	Diff line change
`@@ -63,12 +63,7 @@ func (c criService) CreateContainer(ctx context.Context, r runtime.CreateConta`
`63`	`63`	`return nil, fmt.Errorf("failed to find sandbox id %q: %w", r.GetPodSandboxId(), err)`
`64`	`64`	`}`
`65`	`65`
`66`		`- controller, err := c.sandboxService.SandboxController(sandbox.Config, sandbox.RuntimeHandler)`
`67`		`- if err != nil {`
`68`		`- return nil, fmt.Errorf("failed to get sandbox controller: %w", err)`
`69`		`- }`
`70`		`-`
`71`		`- cstatus, err := controller.Status(ctx, sandbox.ID, false)`
	`66`	`+ cstatus, err := c.sandboxService.SandboxStatus(ctx, sandbox.Sandboxer, sandbox.ID, false)`
`72`	`67`	`if err != nil {`
`73`	`68`	`return nil, fmt.Errorf("failed to get controller status: %w", err)`
`74`	`69`	`}`
`@@ -150,7 +145,7 @@ func (c criService) CreateContainer(ctx context.Context, r runtime.CreateConta`
`150`	`145`	`}`
`151`	`146`	`}()`
`152`	`147`
`153`		`- platform, err := controller.Platform(ctx, sandboxID)`
	`148`	`+ platform, err := c.sandboxService.SandboxPlatform(ctx, sandbox.Sandboxer, sandboxID)`
`154`	`149`	`if err != nil {`
`155`	`150`	`return nil, fmt.Errorf("failed to query sandbox platform: %w", err)`
`156`	`151`	`}`
Original file line number	Diff line number	Diff line change
`@@ -79,13 +79,7 @@ func (c criService) RemovePodSandbox(ctx context.Context, r runtime.RemovePodS`
`79`	`79`	`}`
`80`	`80`	`}`
`81`	`81`
`82`		`- // Use sandbox controller to delete sandbox`
`83`		`- controller, err := c.sandboxService.SandboxController(sandbox.Config, sandbox.RuntimeHandler)`
`84`		`- if err != nil {`
`85`		`- return nil, fmt.Errorf("failed to get sandbox controller: %w", err)`
`86`		`- }`
`87`		`-`
`88`		`- if err := controller.Shutdown(ctx, id); err != nil && !errdefs.IsNotFound(err) {`
	`82`	`+ if err := c.sandboxService.ShutdownSandbox(ctx, sandbox.Sandboxer, id); err != nil && !errdefs.IsNotFound(err) {`
`89`	`83`	`return nil, fmt.Errorf("failed to delete sandbox %q: %w", id, err)`
`90`	`84`	`}`
`91`	`85`