Merge pull request containerd#9463 from abel-von/sandbox-plugin-1205

sandbox: remove sandboxStore from podsandbox controller

Author: fuweid

integration/main_test.go

@@ -34,6 +34,15 @@ import (
 	"testing"
 	"time"
 
+	"github.com/containerd/log"
+	"github.com/opencontainers/selinux/go-selinux"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
+	"k8s.io/klog/v2"
+
 	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/containers"
 	cri "github.com/containerd/containerd/v2/integration/cri-api/pkg/apis"
@@ -42,16 +51,8 @@ import (
 	dialer "github.com/containerd/containerd/v2/integration/remote/util"
 	criconfig "github.com/containerd/containerd/v2/pkg/cri/config"
 	"github.com/containerd/containerd/v2/pkg/cri/constants"
-	"github.com/containerd/containerd/v2/pkg/cri/server"
+	"github.com/containerd/containerd/v2/pkg/cri/server/base"
 	"github.com/containerd/containerd/v2/pkg/cri/util"
-	"github.com/containerd/log"
-	"github.com/opencontainers/selinux/go-selinux"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
-	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
-	"k8s.io/klog/v2"
 )
 
 const (
@@ -676,7 +677,7 @@ func CRIConfig() (*criconfig.Config, error) {
 }
 
 // SandboxInfo gets sandbox info.
-func SandboxInfo(id string) (*runtime.PodSandboxStatus, *server.SandboxInfo, error) {
+func SandboxInfo(id string) (*runtime.PodSandboxStatus, *base.SandboxInfo, error) {
 	client, err := RawRuntimeClient()
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to get raw runtime client: %w", err)
@@ -689,7 +690,7 @@ func SandboxInfo(id string) (*runtime.PodSandboxStatus, *server.SandboxInfo, err
 		return nil, nil, fmt.Errorf("failed to get sandbox status: %w", err)
 	}
 	status := resp.GetStatus()
-	var info server.SandboxInfo
+	var info base.SandboxInfo
 	if err := json.Unmarshal([]byte(resp.GetInfo()["info"]), &info); err != nil {
 		return nil, nil, fmt.Errorf("failed to unmarshal sandbox info: %w", err)
 	}

integration/sandbox_run_rollback_test.go

@@ -35,7 +35,7 @@ import (
 	"github.com/stretchr/testify/require"
 	criapiv1 "k8s.io/cri-api/pkg/apis/runtime/v1"
 
-	"github.com/containerd/containerd/v2/pkg/cri/server/podsandbox"
+	"github.com/containerd/containerd/v2/pkg/cri/server/base"
 	"github.com/containerd/containerd/v2/pkg/failpoint"
 )
 
@@ -299,7 +299,7 @@ func TestRunPodSandboxAndTeardownCNISlow(t *testing.T) {
 }
 
 // sbserverSandboxInfo gets sandbox info.
-func sbserverSandboxInfo(id string) (*criapiv1.PodSandboxStatus, *podsandbox.SandboxInfo, error) {
+func sbserverSandboxInfo(id string) (*criapiv1.PodSandboxStatus, *base.SandboxInfo, error) {
 	client, err := RawRuntimeClient()
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to get raw runtime client: %w", err)
@@ -312,7 +312,7 @@ func sbserverSandboxInfo(id string) (*criapiv1.PodSandboxStatus, *podsandbox.San
 		return nil, nil, fmt.Errorf("failed to get sandbox status: %w", err)
 	}
 	status := resp.GetStatus()
-	var info podsandbox.SandboxInfo
+	var info base.SandboxInfo
 	if err := json.Unmarshal([]byte(resp.GetInfo()["info"]), &info); err != nil {
 		return nil, nil, fmt.Errorf("failed to unmarshal sandbox info: %w", err)
 	}

pkg/cri/server/base/sandbox_info.go

@@ -0,0 +1,47 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package base
+
+import (
+	"github.com/containerd/go-cni"
+	"github.com/opencontainers/runtime-spec/specs-go"
+	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
+
+	"github.com/containerd/containerd/v2/pkg/cri/store/sandbox"
+)
+
+// SandboxInfo is extra information for sandbox.
+// TODO (mikebrow): discuss predefining constants structures for some or all of these field names in CRI
+type SandboxInfo struct {
+	Pid         uint32 `json:"pid"`
+	Status      string `json:"processStatus"`
+	NetNSClosed bool   `json:"netNamespaceClosed"`
+	Image       string `json:"image"`
+	SnapshotKey string `json:"snapshotKey"`
+	Snapshotter string `json:"snapshotter"`
+	// Note: a new field `RuntimeHandler` has been added into the CRI PodSandboxStatus struct, and
+	// should be set. This `RuntimeHandler` field will be deprecated after containerd 1.3 (tracked
+	// in https://github.com/containerd/cri/issues/1064).
+	RuntimeHandler string                    `json:"runtimeHandler"` // see the Note above
+	RuntimeType    string                    `json:"runtimeType"`
+	RuntimeOptions interface{}               `json:"runtimeOptions"`
+	Config         *runtime.PodSandboxConfig `json:"config"`
+	// Note: RuntimeSpec may not be populated if the sandbox has not been fully created.
+	RuntimeSpec *specs.Spec       `json:"runtimeSpec"`
+	CNIResult   *cni.Result       `json:"cniResult"`
+	Metadata    *sandbox.Metadata `json:"sandboxMetadata"`
+}

pkg/cri/server/events.go

@@ -23,6 +23,11 @@ import (
 	"sync"
 	"time"
 
+	"github.com/containerd/log"
+	"github.com/containerd/typeurl/v2"
+	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
+	"k8s.io/utils/clock"
+
 	eventtypes "github.com/containerd/containerd/v2/api/events"
 	apitasks "github.com/containerd/containerd/v2/api/services/tasks/v1"
 	containerdio "github.com/containerd/containerd/v2/cio"
@@ -34,10 +39,6 @@ import (
 	sandboxstore "github.com/containerd/containerd/v2/pkg/cri/store/sandbox"
 	ctrdutil "github.com/containerd/containerd/v2/pkg/cri/util"
 	"github.com/containerd/containerd/v2/protobuf"
-	"github.com/containerd/log"
-	"github.com/containerd/typeurl/v2"
-	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
-	"k8s.io/utils/clock"
 )
 
 const (
@@ -108,7 +109,7 @@ func (em *eventMonitor) subscribe(subscriber events.Subscriber) {
 }
 
 // startSandboxExitMonitor starts an exit monitor for a given sandbox.
-func (em *eventMonitor) startSandboxExitMonitor(ctx context.Context, id string, pid uint32, exitCh <-chan containerd.ExitStatus) <-chan struct{} {
+func (em *eventMonitor) startSandboxExitMonitor(ctx context.Context, id string, exitCh <-chan containerd.ExitStatus) <-chan struct{} {
 	stopCh := make(chan struct{})
 	go func() {
 		defer close(stopCh)

pkg/cri/server/podsandbox/controller.go

@@ -34,8 +34,8 @@ import (
 	"github.com/containerd/containerd/v2/pkg/cri/constants"
 	"github.com/containerd/containerd/v2/pkg/cri/server/base"
 	"github.com/containerd/containerd/v2/pkg/cri/server/images"
+	"github.com/containerd/containerd/v2/pkg/cri/server/podsandbox/types"
 	imagestore "github.com/containerd/containerd/v2/pkg/cri/store/image"
-	sandboxstore "github.com/containerd/containerd/v2/pkg/cri/store/sandbox"
 	ctrdutil "github.com/containerd/containerd/v2/pkg/cri/util"
 	osinterface "github.com/containerd/containerd/v2/pkg/os"
 	"github.com/containerd/containerd/v2/platforms"
@@ -116,8 +116,6 @@ type Controller struct {
 	client *containerd.Client
 	// imageService is a dependency to CRI image service.
 	imageService ImageService
-	// sandboxStore stores all resources associated with sandboxes.
-	sandboxStore *sandboxstore.Store
 	// os is an interface for all required os operations.
 	os osinterface.OS
 	// cri is CRI service that provides missing gaps needed by controller.
@@ -129,11 +127,9 @@ type Controller struct {
 }
 
 func (c *Controller) Init(
-	sandboxStore *sandboxstore.Store,
 	cri CRIService,
 ) {
 	c.cri = cri
-	c.sandboxStore = sandboxStore
 }
 
 var _ sandbox.Controller = (*Controller)(nil)
@@ -143,63 +139,46 @@ func (c *Controller) Platform(_ctx context.Context, _sandboxID string) (platform
 }
 
 func (c *Controller) Wait(ctx context.Context, sandboxID string) (sandbox.ExitStatus, error) {
-	status := c.store.Get(sandboxID)
-	if status == nil {
+	podSandbox := c.store.Get(sandboxID)
+	if podSandbox == nil {
 		return sandbox.ExitStatus{}, fmt.Errorf("failed to get exit channel. %q", sandboxID)
-	}
-
-	exitStatus, exitedAt, err := c.waitSandboxExit(ctx, sandboxID, status.Waiter)
 
+	}
+	exit, err := podSandbox.Wait(ctx)
+	if err != nil {
+		return sandbox.ExitStatus{}, fmt.Errorf("failed to wait pod sandbox, %w", err)
+	}
 	return sandbox.ExitStatus{
-		ExitStatus: exitStatus,
-		ExitedAt:   exitedAt,
+		ExitStatus: exit.ExitCode(),
+		ExitedAt:   exit.ExitTime(),
 	}, err
+
 }
 
-func (c *Controller) waitSandboxExit(ctx context.Context, id string, exitCh <-chan containerd.ExitStatus) (exitStatus uint32, exitedAt time.Time, err error) {
-	exitStatus = unknownExitCode
-	exitedAt = time.Now()
+func (c *Controller) waitSandboxExit(ctx context.Context, p *types.PodSandbox, exitCh <-chan containerd.ExitStatus) (exitStatus uint32, exitedAt time.Time, err error) {
 	select {
-	case exitRes := <-exitCh:
-		log.G(ctx).Debugf("received sandbox exit %+v", exitRes)
-
-		exitStatus, exitedAt, err = exitRes.Result()
+	case e := <-exitCh:
+		exitStatus, exitedAt, err = e.Result()
 		if err != nil {
-			log.G(ctx).WithError(err).Errorf("failed to get task exit status for %q", id)
+			log.G(ctx).WithError(err).Errorf("failed to get task exit status for %q", p.ID)
 			exitStatus = unknownExitCode
 			exitedAt = time.Now()
 		}
-
-		err = func() error {
-			dctx := ctrdutil.NamespacedContext()
-			dctx, dcancel := context.WithTimeout(dctx, handleEventTimeout)
-			defer dcancel()
-
-			sb, err := c.sandboxStore.Get(id)
-			if err == nil {
-				if err := handleSandboxExit(dctx, sb, &eventtypes.TaskExit{ExitStatus: exitStatus, ExitedAt: protobuf.ToTimestamp(exitedAt)}); err != nil {
-					return err
-				}
-				return nil
-			} else if !errdefs.IsNotFound(err) {
-				return fmt.Errorf("failed to get sandbox %s: %w", id, err)
-			}
-			return nil
-		}()
-		if err != nil {
-			log.G(ctx).WithError(err).Errorf("failed to handle sandbox TaskExit %s", id)
-			// Don't backoff, the caller is responsible for.
-			return
+		dctx := ctrdutil.NamespacedContext()
+		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 {
+			c.cri.BackOffEvent(p.ID, e)
 		}
+		return
 	case <-ctx.Done():
-		return exitStatus, exitedAt, ctx.Err()
+		return unknownExitCode, time.Now(), ctx.Err()
 	}
-	return
 }
 
-// handleSandboxExit handles TaskExit event for sandbox.
-// TODO https://github.com/containerd/containerd/issues/7548
-func handleSandboxExit(ctx context.Context, sb sandboxstore.Sandbox, e *eventtypes.TaskExit) error {
+// handleSandboxTaskExit handles TaskExit event for sandbox.
+func handleSandboxTaskExit(ctx context.Context, sb *types.PodSandbox, e *eventtypes.TaskExit) error {
 	// No stream attached to sandbox container.
 	task, err := sb.Container.Task(ctx, nil)
 	if err != nil {
@@ -212,17 +191,7 @@ func handleSandboxExit(ctx context.Context, sb sandboxstore.Sandbox, e *eventtyp
 			if !errdefs.IsNotFound(err) {
 				return fmt.Errorf("failed to stop sandbox: %w", err)
 			}
-			// Move on to make sure container status is updated.
 		}
 	}
-	sb.Status.Update(func(status sandboxstore.Status) (sandboxstore.Status, error) {
-		status.State = sandboxstore.StateNotReady
-		status.Pid = 0
-		status.ExitStatus = e.ExitStatus
-		status.ExitedAt = e.ExitedAt.AsTime()
-		return status, nil
-	})
-	// Using channel to propagate the information of sandbox stop
-	sb.Stop()
 	return nil
 }

pkg/cri/server/podsandbox/controller_test.go

@@ -21,11 +21,13 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
+	containerd "github.com/containerd/containerd/v2/client"
 	criconfig "github.com/containerd/containerd/v2/pkg/cri/config"
-	"github.com/containerd/containerd/v2/pkg/cri/store/label"
+	"github.com/containerd/containerd/v2/pkg/cri/server/podsandbox/types"
 	sandboxstore "github.com/containerd/containerd/v2/pkg/cri/store/sandbox"
 	ostesting "github.com/containerd/containerd/v2/pkg/os/testing"
-	"github.com/stretchr/testify/assert"
 )
 
 const (
@@ -48,32 +50,25 @@ var testConfig = criconfig.Config{
 
 // newControllerService creates a fake criService for test.
 func newControllerService() *Controller {
-	labels := label.NewStore()
 	return &Controller{
-		config:       testConfig,
-		os:           ostesting.NewFakeOS(),
-		sandboxStore: sandboxstore.NewStore(labels),
+		config: testConfig,
+		os:     ostesting.NewFakeOS(),
+		store:  NewStore(),
 	}
 }
 
 func Test_Status(t *testing.T) {
 	sandboxID, pid, exitStatus := "1", uint32(1), uint32(0)
 	createdAt, exitedAt := time.Now(), time.Now()
 	controller := newControllerService()
-	status := sandboxstore.Status{
-		Pid:        pid,
-		CreatedAt:  createdAt,
-		ExitStatus: exitStatus,
-		ExitedAt:   exitedAt,
-		State:      sandboxstore.StateReady,
-	}
-	sb := sandboxstore.Sandbox{
-		Metadata: sandboxstore.Metadata{
-			ID: sandboxID,
-		},
-		Status: sandboxstore.StoreStatus(status),
-	}
-	err := controller.sandboxStore.Add(sb)
+
+	sb := types.NewPodSandbox(sandboxID, sandboxstore.Status{
+		State:     sandboxstore.StateReady,
+		Pid:       pid,
+		CreatedAt: createdAt,
+	})
+	sb.Metadata = sandboxstore.Metadata{ID: sandboxID}
+	err := controller.store.Save(sb)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -82,6 +77,20 @@ func Test_Status(t *testing.T) {
 		t.Fatal(err)
 	}
 	assert.Equal(t, s.Pid, pid)
-	assert.Equal(t, s.ExitedAt, exitedAt)
+	assert.Equal(t, s.CreatedAt, createdAt)
 	assert.Equal(t, s.State, sandboxstore.StateReady.String())
+
+	sb.Exit(*containerd.NewExitStatus(exitStatus, exitedAt, nil))
+	exit, err := controller.Wait(context.Background(), sandboxID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, exit.ExitStatus, exitStatus)
+	assert.Equal(t, exit.ExitedAt, exitedAt)
+
+	s, err = controller.Status(context.Background(), sandboxID, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, s.State, sandboxstore.StateNotReady.String())
 }