Merge pull request #8373 from Iceber/shim_run

runtime/shim: rename RunManager to Run and remove `runc/v2/services` package

cmd/containerd-shim-runc-v2/main.go

@@ -28,5 +28,5 @@ import (
 )
 
 func main() {
-	shim.RunManager(context.Background(), manager.NewShimManager("io.containerd.runc.v2"))
+	shim.Run(context.Background(), manager.NewShimManager("io.containerd.runc.v2"))
 }

integration/failpoint/cmd/containerd-shim-runc-fp-v1/main_linux.go

@@ -25,5 +25,5 @@ import (
 )
 
 func main() {
-	shim.RunManager(context.Background(), manager.NewShimManager("io.containerd.runc-fp.v1"))
+	shim.Run(context.Background(), manager.NewShimManager("io.containerd.runc-fp.v1"))
 }

runtime/v2/example/cmd/main.go

@@ -19,11 +19,14 @@
 package main
 
 import (
-	"github.com/containerd/containerd/runtime/v2/example"
+	"context"
+
+	_ "github.com/containerd/containerd/runtime/v2/example"
+	"github.com/containerd/containerd/runtime/v2/runc/manager"
 	"github.com/containerd/containerd/runtime/v2/shim"
 )
 
 func main() {
 	// init and execute the shim
-	shim.Run("io.containerd.example.v1", example.New)
+	shim.Run(context.Background(), manager.NewShimManager("io.containerd.example.v1"))
 }

runtime/v2/example/example.go

@@ -24,115 +24,141 @@ import (
 
 	taskAPI "github.com/containerd/containerd/api/runtime/task/v2"
 	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/pkg/shutdown"
+	"github.com/containerd/containerd/plugin"
 	ptypes "github.com/containerd/containerd/protobuf/types"
 	"github.com/containerd/containerd/runtime/v2/shim"
 )
 
 var (
-	// check to make sure the *service implements the GRPC API
+	// check to make sure the *exampleTaskService implements the GRPC API
-	_ = (taskAPI.TaskService)(&service{})
+	_ = (taskAPI.TaskService)(&exampleTaskService{})
 )
 
-// New returns a new shim service
+func init() {
-func New(ctx context.Context, id string, publisher shim.Publisher, shutdown func()) (shim.Shim, error) {
+	plugin.Register(&plugin.Registration{
-	return &service{}, nil
+		Type: plugin.TTRPCPlugin,
+		ID:   "task",
+		Requires: []plugin.Type{
+			plugin.EventPlugin,
+			plugin.InternalPlugin,
+		},
+		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
+			pp, err := ic.GetByID(plugin.EventPlugin, "publisher")
+			if err != nil {
+				return nil, err
+			}
+			ss, err := ic.GetByID(plugin.InternalPlugin, "shutdown")
+			if err != nil {
+				return nil, err
+			}
+			return NewTaskService(ic.Context, pp.(shim.Publisher), ss.(shutdown.Service))
+		},
+	})
 }
 
-type service struct {
+// NewTaskService creates a new instance of a task service
+func NewTaskService(ctx context.Context, publisher shim.Publisher, sd shutdown.Service) (taskAPI.TaskService, error) {
+	// The shim.Publisher and shutdown.Service are usually useful for your task service,
+	// but we don't need them in the exampleTaskService.
+	return &exampleTaskService{}, nil
+}
+
+type exampleTaskService struct {
 }
 
 // StartShim is a binary call that executes a new shim returning the address
-func (s *service) StartShim(ctx context.Context, opts shim.StartOpts) (string, error) {
+func (s *exampleTaskService) StartShim(ctx context.Context, opts shim.StartOpts) (string, error) {
 	return "", nil
 }
 
 // Cleanup is a binary call that cleans up any resources used by the shim when the service crashes
-func (s *service) Cleanup(ctx context.Context) (*taskAPI.DeleteResponse, error) {
+func (s *exampleTaskService) Cleanup(ctx context.Context) (*taskAPI.DeleteResponse, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Create a new container
-func (s *service) Create(ctx context.Context, r *taskAPI.CreateTaskRequest) (_ *taskAPI.CreateTaskResponse, err error) {
+func (s *exampleTaskService) Create(ctx context.Context, r *taskAPI.CreateTaskRequest) (_ *taskAPI.CreateTaskResponse, err error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Start the primary user process inside the container
-func (s *service) Start(ctx context.Context, r *taskAPI.StartRequest) (*taskAPI.StartResponse, error) {
+func (s *exampleTaskService) Start(ctx context.Context, r *taskAPI.StartRequest) (*taskAPI.StartResponse, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Delete a process or container
-func (s *service) Delete(ctx context.Context, r *taskAPI.DeleteRequest) (*taskAPI.DeleteResponse, error) {
+func (s *exampleTaskService) Delete(ctx context.Context, r *taskAPI.DeleteRequest) (*taskAPI.DeleteResponse, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Exec an additional process inside the container
-func (s *service) Exec(ctx context.Context, r *taskAPI.ExecProcessRequest) (*ptypes.Empty, error) {
+func (s *exampleTaskService) Exec(ctx context.Context, r *taskAPI.ExecProcessRequest) (*ptypes.Empty, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // ResizePty of a process
-func (s *service) ResizePty(ctx context.Context, r *taskAPI.ResizePtyRequest) (*ptypes.Empty, error) {
+func (s *exampleTaskService) ResizePty(ctx context.Context, r *taskAPI.ResizePtyRequest) (*ptypes.Empty, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // State returns runtime state of a process
-func (s *service) State(ctx context.Context, r *taskAPI.StateRequest) (*taskAPI.StateResponse, error) {
+func (s *exampleTaskService) State(ctx context.Context, r *taskAPI.StateRequest) (*taskAPI.StateResponse, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Pause the container
-func (s *service) Pause(ctx context.Context, r *taskAPI.PauseRequest) (*ptypes.Empty, error) {
+func (s *exampleTaskService) Pause(ctx context.Context, r *taskAPI.PauseRequest) (*ptypes.Empty, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Resume the container
-func (s *service) Resume(ctx context.Context, r *taskAPI.ResumeRequest) (*ptypes.Empty, error) {
+func (s *exampleTaskService) Resume(ctx context.Context, r *taskAPI.ResumeRequest) (*ptypes.Empty, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Kill a process
-func (s *service) Kill(ctx context.Context, r *taskAPI.KillRequest) (*ptypes.Empty, error) {
+func (s *exampleTaskService) Kill(ctx context.Context, r *taskAPI.KillRequest) (*ptypes.Empty, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Pids returns all pids inside the container
-func (s *service) Pids(ctx context.Context, r *taskAPI.PidsRequest) (*taskAPI.PidsResponse, error) {
+func (s *exampleTaskService) Pids(ctx context.Context, r *taskAPI.PidsRequest) (*taskAPI.PidsResponse, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // CloseIO of a process
-func (s *service) CloseIO(ctx context.Context, r *taskAPI.CloseIORequest) (*ptypes.Empty, error) {
+func (s *exampleTaskService) CloseIO(ctx context.Context, r *taskAPI.CloseIORequest) (*ptypes.Empty, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Checkpoint the container
-func (s *service) Checkpoint(ctx context.Context, r *taskAPI.CheckpointTaskRequest) (*ptypes.Empty, error) {
+func (s *exampleTaskService) Checkpoint(ctx context.Context, r *taskAPI.CheckpointTaskRequest) (*ptypes.Empty, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Connect returns shim information of the underlying service
-func (s *service) Connect(ctx context.Context, r *taskAPI.ConnectRequest) (*taskAPI.ConnectResponse, error) {
+func (s *exampleTaskService) Connect(ctx context.Context, r *taskAPI.ConnectRequest) (*taskAPI.ConnectResponse, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Shutdown is called after the underlying resources of the shim are cleaned up and the service can be stopped
-func (s *service) Shutdown(ctx context.Context, r *taskAPI.ShutdownRequest) (*ptypes.Empty, error) {
+func (s *exampleTaskService) Shutdown(ctx context.Context, r *taskAPI.ShutdownRequest) (*ptypes.Empty, error) {
 	os.Exit(0)
 	return &ptypes.Empty{}, nil
 }
 
 // Stats returns container level system stats for a container and its processes
-func (s *service) Stats(ctx context.Context, r *taskAPI.StatsRequest) (*taskAPI.StatsResponse, error) {
+func (s *exampleTaskService) Stats(ctx context.Context, r *taskAPI.StatsRequest) (*taskAPI.StatsResponse, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Update the live container
-func (s *service) Update(ctx context.Context, r *taskAPI.UpdateTaskRequest) (*ptypes.Empty, error) {
+func (s *exampleTaskService) Update(ctx context.Context, r *taskAPI.UpdateTaskRequest) (*ptypes.Empty, error) {
 	return nil, errdefs.ErrNotImplemented
 }
 
 // Wait for a process to exit
-func (s *service) Wait(ctx context.Context, r *taskAPI.WaitRequest) (*taskAPI.WaitResponse, error) {
+func (s *exampleTaskService) Wait(ctx context.Context, r *taskAPI.WaitRequest) (*taskAPI.WaitResponse, error) {
 	return nil, errdefs.ErrNotImplemented
 }

runtime/v2/runc/v2/service.go

@@ -1,79 +0,0 @@
-//go:build linux
-
-/*
-   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 v2
-
-import (
-	"context"
-
-	shimapi "github.com/containerd/containerd/api/runtime/task/v2"
-	"github.com/containerd/containerd/pkg/shutdown"
-	"github.com/containerd/containerd/protobuf"
-	"github.com/containerd/containerd/runtime/v2/runc/manager"
-	"github.com/containerd/containerd/runtime/v2/runc/task"
-	"github.com/containerd/containerd/runtime/v2/shim"
-)
-
-// TODO(2.0): Remove this package
-
-type shimTaskManager struct {
-	shimapi.TaskService
-	id      string
-	manager shim.Manager
-}
-
-func (stm *shimTaskManager) Cleanup(ctx context.Context) (*shimapi.DeleteResponse, error) {
-	ss, err := stm.manager.Stop(ctx, stm.id)
-	if err != nil {
-		return nil, err
-	}
-	return &shimapi.DeleteResponse{
-		Pid:        uint32(ss.Pid),
-		ExitStatus: uint32(ss.ExitStatus),
-		ExitedAt:   protobuf.ToTimestamp(ss.ExitedAt),
-	}, nil
-}
-
-func (stm *shimTaskManager) StartShim(ctx context.Context, opts shim.StartOpts) (string, error) {
-	return stm.manager.Start(ctx, opts.ID, opts)
-}
-
-// New returns a new shim service that can be used for
-// - serving the task service over grpc/ttrpc
-// - shim management
-// This function is deprecated in favor direct creation
-// of shim manager and registering task service via plugins.
-func New(ctx context.Context, id string, publisher shim.Publisher, fn func()) (shim.Shim, error) {
-	sd, ok := ctx.(shutdown.Service)
-	if !ok {
-		ctx, sd = shutdown.WithShutdown(ctx)
-		sd.RegisterCallback(func(context.Context) error {
-			fn()
-			return nil
-		})
-	}
-	ts, err := task.NewTaskService(ctx, publisher, sd)
-	if err != nil {
-		return nil, err
-	}
-	return &shimTaskManager{
-		TaskService: ts,
-		id:          id,
-		manager:     manager.NewShimManager("runc"),
-	}, nil
-}

runtime/v2/shim/shim.go

@@ -50,11 +50,9 @@ type Publisher interface {
 
 // StartOpts describes shim start configuration received from containerd
 type StartOpts struct {
-	ID               string // TODO(2.0): Remove ID, passed directly to start for call symmetry
+	Address      string
-	ContainerdBinary string // TODO(2.0): Remove ContainerdBinary, use the TTRPC_ADDRESS env to forward events
+	TTRPCAddress string
-	Address          string
+	Debug        bool
-	TTRPCAddress     string
-	Debug            bool
 }
 
 // BootstrapParams is a JSON payload returned in stdout from shim.Start call.
@@ -73,18 +71,6 @@ type StopStatus struct {
 	ExitedAt   time.Time
 }
 
-// Init func for the creation of a shim server
-// TODO(2.0): Remove init function
-type Init func(context.Context, string, Publisher, func()) (Shim, error)
-
-// Shim server interface
-// TODO(2.0): Remove unified shim interface
-type Shim interface {
-	shimapi.TaskService
-	Cleanup(ctx context.Context) (*shimapi.DeleteResponse, error)
-	StartShim(ctx context.Context, opts StartOpts) (string, error)
-}
-
 // Manager is the interface which manages the shim process
 type Manager interface {
 	Name() string
@@ -124,15 +110,6 @@ type ttrpcServerOptioner interface {
 	UnaryInterceptor() ttrpc.UnaryServerInterceptor
 }
 
-type taskService struct {
-	shimapi.TaskService
-}
-
-func (t taskService) RegisterTTRPC(server *ttrpc.Server) error {
-	shimapi.RegisterTaskService(server, t.TaskService)
-	return nil
-}
-
 var (
 	debugFlag            bool
 	versionFlag          bool
@@ -200,54 +177,8 @@ func setLogger(ctx context.Context, id string) (context.Context, error) {
 	return log.WithLogger(ctx, l), nil
 }
 
-// Run initializes and runs a shim server
+// Run initializes and runs a shim server.
-// TODO(2.0): Remove function
+func Run(ctx context.Context, manager Manager, opts ...BinaryOpts) {
-func Run(name string, initFunc Init, opts ...BinaryOpts) {
-	var config Config
-	for _, o := range opts {
-		o(&config)
-	}
-
-	ctx := context.Background()
-	ctx = log.WithLogger(ctx, log.G(ctx).WithField("runtime", name))
-
-	if err := run(ctx, nil, initFunc, name, config); err != nil {
-		fmt.Fprintf(os.Stderr, "%s: %s", name, err)
-		os.Exit(1)
-	}
-}
-
-// TODO(2.0): Remove this type
-type shimToManager struct {
-	shim Shim
-	name string
-}
-
-func (stm shimToManager) Name() string {
-	return stm.name
-}
-
-func (stm shimToManager) Start(ctx context.Context, id string, opts StartOpts) (string, error) {
-	opts.ID = id
-	return stm.shim.StartShim(ctx, opts)
-}
-
-func (stm shimToManager) Stop(ctx context.Context, id string) (StopStatus, error) {
-	// shim must already have id
-	dr, err := stm.shim.Cleanup(ctx)
-	if err != nil {
-		return StopStatus{}, err
-	}
-	return StopStatus{
-		Pid:        int(dr.Pid),
-		ExitStatus: int(dr.ExitStatus),
-		ExitedAt:   protobuf.FromTimestamp(dr.ExitedAt),
-	}, nil
-}
-
-// RunManager initializes and runs a shim server.
-// TODO(2.0): Rename to Run
-func RunManager(ctx context.Context, manager Manager, opts ...BinaryOpts) {
 	var config Config
 	for _, o := range opts {
 		o(&config)
@@ -255,13 +186,13 @@ func RunManager(ctx context.Context, manager Manager, opts ...BinaryOpts) {
 
 	ctx = log.WithLogger(ctx, log.G(ctx).WithField("runtime", manager.Name()))
 
-	if err := run(ctx, manager, nil, "", config); err != nil {
+	if err := run(ctx, manager, "", config); err != nil {
 		fmt.Fprintf(os.Stderr, "%s: %s", manager.Name(), err)
 		os.Exit(1)
 	}
 }
 
-func run(ctx context.Context, manager Manager, initFunc Init, name string, config Config) error {
+func run(ctx context.Context, manager Manager, name string, config Config) error {
 	parseFlags()
 	if versionFlag {
 		fmt.Printf("%s:\n", filepath.Base(os.Args[0]))
@@ -301,27 +232,6 @@ func run(ctx context.Context, manager Manager, initFunc Init, name string, confi
 	ctx, sd := shutdown.WithShutdown(ctx)
 	defer sd.Shutdown()
 
-	if manager == nil {
-		service, err := initFunc(ctx, id, publisher, sd.Shutdown)
-		if err != nil {
-			return err
-		}
-		plugin.Register(&plugin.Registration{
-			Type: plugin.TTRPCPlugin,
-			ID:   "task",
-			Requires: []plugin.Type{
-				plugin.EventPlugin,
-			},
-			InitFn: func(ic *plugin.InitContext) (interface{}, error) {
-				return taskService{service}, nil
-			},
-		})
-		manager = shimToManager{
-			shim: service,
-			name: name,
-		}
-	}
-
 	// Handle explicit actions
 	switch action {
 	case "delete":