sandbox: store endpoint in cri sandboxStore

Signed-off-by: Abel Feng <[email protected]>

Author: abel-von

core/runtime/v2/shim.go

@@ -185,8 +185,9 @@ type ShimInstance interface {
 	Client() any
 	// Delete will close the client and remove bundle from disk.
 	Delete(ctx context.Context) error
-	// Version returns shim's features compatibility version.
-	Version() int
+	// Endpoint returns shim's endpoint information,
+	// including address, protocol and version.
+	Endpoint() (string, string, int)
 }
 
 func parseStartResponse(response []byte) (client.BootstrapParams, error) {
@@ -359,9 +360,11 @@ func (gc *grpcConn) UserOnCloseWait(ctx context.Context) error {
 }
 
 type shim struct {
-	bundle  *Bundle
-	client  any
-	version int
+	bundle   *Bundle
+	client   any
+	address  string
+	protocol string
+	version  int
 }
 
 var _ ShimInstance = (*shim)(nil)
@@ -371,8 +374,8 @@ func (s *shim) ID() string {
 	return s.bundle.ID
 }
 
-func (s *shim) Version() int {
-	return s.version
+func (s *shim) Endpoint() (string, string, int) {
+	return s.address, s.protocol, s.version
 }
 
 func (s *shim) Namespace() string {
@@ -440,7 +443,8 @@ type shimTask struct {
 }
 
 func newShimTask(shim ShimInstance) (*shimTask, error) {
-	taskClient, err := NewTaskClient(shim.Client(), shim.Version())
+	_, _, version := shim.Endpoint()
+	taskClient, err := NewTaskClient(shim.Client(), version)
 	if err != nil {
 		return nil, err
 	}

core/sandbox/controller.go

@@ -117,6 +117,9 @@ type ControllerInstance struct {
 	SandboxID string
 	Pid       uint32
 	CreatedAt time.Time
+	Address   string
+	Protocol  string
+	Version   uint32
 	Labels    map[string]string
 }
 
@@ -133,4 +136,7 @@ type ControllerStatus struct {
 	CreatedAt time.Time
 	ExitedAt  time.Time
 	Extra     typeurl.Any
+	Address   string
+	Protocol  string
+	Version   uint32
 }

core/sandbox/proxy/controller.go

@@ -73,6 +73,9 @@ func (s *remoteSandboxController) Start(ctx context.Context, sandboxID string) (
 		Pid:       resp.GetPid(),
 		CreatedAt: resp.GetCreatedAt().AsTime(),
 		Labels:    resp.GetLabels(),
+		Address:   resp.GetAddress(),
+		Protocol:  resp.GetProtocol(),
+		Version:   resp.GetVersion(),
 	}, nil
 }
 
@@ -141,6 +144,9 @@ func (s *remoteSandboxController) Status(ctx context.Context, sandboxID string,
 		CreatedAt: resp.GetCreatedAt().AsTime(),
 		ExitedAt:  resp.GetExitedAt().AsTime(),
 		Extra:     resp.GetExtra(),
+		Address:   resp.GetAddress(),
+		Protocol:  resp.GetProtocol(),
+		Version:   resp.GetVersion(),
 	}, nil
 }
 

internal/cri/server/container_start.go

@@ -118,6 +118,14 @@ func (c *criService) StartContainer(ctx context.Context, r *runtime.StartContain
 	if ociRuntime.Path != "" {
 		taskOpts = append(taskOpts, containerd.WithRuntimePath(ociRuntime.Path))
 	}
+
+	// append endpoint to the options so that task manager can get task api endpoint directly
+	endpoint := sandbox.Endpoint
+	if endpoint.IsValid() {
+		taskOpts = append(taskOpts,
+			containerd.WithTaskApiEndpoint(endpoint.Address, endpoint.Protocol, endpoint.Version))
+	}
+
 	task, err := container.NewTask(ctx, ioCreation, taskOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create containerd task: %w", err)

internal/cri/server/restart.go

@@ -113,6 +113,7 @@ func (c *criService) recover(ctx context.Context) error {
 		var (
 			state      = sandboxstore.StateUnknown
 			controller = c.client.SandboxController(sbx.Sandboxer)
+			endpoint   sandboxstore.Endpoint
 		)
 
 		status, err := controller.Status(ctx, sbx.ID, false)
@@ -126,6 +127,9 @@ func (c *criService) recover(ctx context.Context) error {
 				state = sandboxstore.StateNotReady
 			}
 		} else {
+			endpoint.Protocol = status.Protocol
+			endpoint.Version = status.Version
+			endpoint.Address = status.Address
 			if code, ok := runtime.PodSandboxState_value[status.State]; ok {
 				if code == int32(runtime.PodSandboxState_SANDBOX_READY) {
 					state = sandboxstore.StateReady
@@ -137,6 +141,7 @@ func (c *criService) recover(ctx context.Context) error {
 
 		sb := sandboxstore.NewSandbox(metadata, sandboxstore.Status{State: state})
 		sb.Sandboxer = sbx.Sandboxer
+		sb.Endpoint = endpoint
 
 		// Load network namespace.
 		sb.NetNS = getNetNS(&metadata)

internal/cri/server/sandbox_run.go

@@ -265,6 +265,18 @@ func (c *criService) RunPodSandbox(ctx context.Context, r *runtime.RunPodSandbox
 		return nil, fmt.Errorf("failed to start sandbox %q: %w", id, err)
 	}
 
+	if ctrl.Protocol != "" && ctrl.Address != "" {
+		sandbox.Endpoint = sandboxstore.Endpoint{
+			Protocol: ctrl.Protocol,
+			Version:  ctrl.Version,
+			Address:  ctrl.Address,
+		}
+	}
+
+	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 !hostNetwork(config) && userNsEnabled {
 		// If userns is enabled, then the netns was created by the OCI runtime
 		// on controller.Start(). The OCI runtime needs to create the netns

internal/cri/store/sandbox/sandbox.go

@@ -47,6 +47,18 @@ type Sandbox struct {
 	*store.StopCh
 	// Stats contains (mutable) stats for the (pause) sandbox container
 	Stats *stats.ContainerStats
+	// Endpoint is the sandbox endpoint, for task or streaming api connection
+	Endpoint Endpoint
+}
+
+type Endpoint struct {
+	Address  string
+	Protocol string
+	Version  uint32
+}
+
+func (e *Endpoint) IsValid() bool {
+	return e.Protocol != "" && e.Address != ""
 }
 
 // NewSandbox creates an internally used sandbox type. This functions reminds

plugins/sandbox/controller.go

@@ -188,10 +188,13 @@ func (c *controllerLocal) Start(ctx context.Context, sandboxID string) (sandbox.
 		c.cleanupShim(ctx, sandboxID, svc)
 		return sandbox.ControllerInstance{}, fmt.Errorf("failed to start sandbox %s: %w", sandboxID, errdefs.FromGRPC(err))
 	}
-
+	address, protocol, version := shim.Endpoint()
 	return sandbox.ControllerInstance{
 		SandboxID: sandboxID,
 		Pid:       resp.GetPid(),
+		Address:   address,
+		Protocol:  protocol,
+		Version:   uint32(version),
 		CreatedAt: resp.GetCreatedAt().AsTime(),
 	}, nil
 }
@@ -302,6 +305,12 @@ func (c *controllerLocal) Status(ctx context.Context, sandboxID string, verbose
 		return sandbox.ControllerStatus{}, fmt.Errorf("failed to query sandbox %s status: %w", sandboxID, err)
 	}
 
+	shim, err := c.shims.Get(ctx, sandboxID)
+	if err != nil {
+		return sandbox.ControllerStatus{}, fmt.Errorf("unable to find sandbox %q", sandboxID)
+	}
+	address, protocol, version := shim.Endpoint()
+
 	return sandbox.ControllerStatus{
 		SandboxID: resp.GetSandboxID(),
 		Pid:       resp.GetPid(),
@@ -310,6 +319,9 @@ func (c *controllerLocal) Status(ctx context.Context, sandboxID string, verbose
 		CreatedAt: resp.GetCreatedAt().AsTime(),
 		ExitedAt:  resp.GetExitedAt().AsTime(),
 		Extra:     resp.GetExtra(),
+		Address:   address,
+		Protocol:  protocol,
+		Version:   uint32(version),
 	}, nil
 }