bin/ctr,integration: new runc-shim with failpoint

Added new runc shim binary in integration testing.

The shim is named by io.containerd.runc-fp.v1, which allows us to use
additional OCI annotation `io.containerd.runtime.v2.shim.failpoint.*` to
setup shim task API's failpoint. Since the shim can be shared with
multiple container, like what kubernetes pod does, the failpoint will be
initialized during setup the shim server. So, the following the
container's OCI failpoint's annotation will not work.

This commit also updates the ctr tool that we can use `--annotation` to
specify annotations when run container. For example:

```bash
➜  ctr run -d --runtime runc-fp.v1 \
     --annotation "io.containerd.runtime.v2.shim.failpoint.Kill=1*error(sorry)" \
     docker.io/library/alpine:latest testing sleep 1d

➜  ctr t ls
TASK       PID       STATUS
testing    147304    RUNNING

➜  ctr t kill -s SIGKILL testing
ctr: sorry: unknown

➜  ctr t kill -s SIGKILL testing

➜  sudo ctr t ls
TASK       PID       STATUS
testing    147304    STOPPED
```

The runc-fp.v1 shim is based on core runc.v2. We can use it to inject
failpoint during testing complicated or big transcation API, like
kubernetes PodRunPodsandbox.

Signed-off-by: Wei Fu <fuweid89@gmail.com>
(cherry picked from commit 5f9b318)
Signed-off-by: Qiutong Song <songqt01@gmail.com>

Makefile

@@ -223,6 +223,11 @@ cri-integration: binaries bin/cri-integration.test ## run cri integration tests
 	@bash -x ./script/test/cri-integration.sh
 	@rm -rf bin/cri-integration.test
 
+# build runc shimv2 with failpoint control, only used by integration test
+bin/containerd-shim-runc-fp-v1: integration/failpoint/cmd/containerd-shim-runc-fp-v1 FORCE
+	@echo "$(WHALE) $@"
+	@CGO_ENABLED=${SHIM_CGO_ENABLED} $(GO) build ${GO_BUILD_FLAGS} -o $@ ${SHIM_GO_LDFLAGS} ${GO_TAGS} ./integration/failpoint/cmd/containerd-shim-runc-fp-v1
+
 benchmark: ## run benchmarks tests
 	@echo "$(WHALE) $@"
 	@$(GO) test ${TESTFLAGS} -bench . -run Benchmark -test.root
@@ -369,6 +374,7 @@ clean-test: ## clean up debris from previously failed tests
 	@rm -rf /run/containerd/fifo/*
 	@rm -rf /run/containerd-test/*
 	@rm -rf bin/cri-integration.test
+	@rm -rf bin/containerd-shim-runc-fp-v1
 
 install: ## install binaries
 	@echo "$(WHALE) $@ $(BINARIES)"

cmd/ctr/commands/commands.go

@@ -116,6 +116,10 @@ var (
 			Name:  "label",
 			Usage: "specify additional labels (e.g. foo=bar)",
 		},
+		cli.StringSliceFlag{
+			Name:  "annotation",
+			Usage: "specify additional OCI annotations (e.g. foo=bar)",
+		},
 		cli.StringSliceFlag{
 			Name:  "mount",
 			Usage: "specify additional container mount (e.g. type=bind,src=/tmp,dst=/host,options=rbind:ro)",
@@ -227,6 +231,19 @@ func LabelArgs(labelStrings []string) map[string]string {
 	return labels
 }
 
+// AnnotationArgs returns a map of annotation key,value pairs.
+func AnnotationArgs(annoStrings []string) (map[string]string, error) {
+	annotations := make(map[string]string, len(annoStrings))
+	for _, anno := range annoStrings {
+		parts := strings.SplitN(anno, "=", 2)
+		if len(parts) != 2 {
+			return nil, fmt.Errorf("invalid key=value format annotation: %v", anno)
+		}
+		annotations[parts[0]] = parts[1]
+	}
+	return annotations, nil
+}
+
 // PrintAsJSON prints input in JSON format
 func PrintAsJSON(x interface{}) {
 	b, err := json.MarshalIndent(x, "", "    ")

cmd/ctr/commands/run/run_unix.go

@@ -213,6 +213,13 @@ func NewContainer(ctx gocontext.Context, client *containerd.Client, context *cli
 				oci.WithEnv([]string{fmt.Sprintf("HOSTNAME=%s", hostname)}),
 			)
 		}
+		if annoStrings := context.StringSlice("annotation"); len(annoStrings) > 0 {
+			annos, err := commands.AnnotationArgs(annoStrings)
+			if err != nil {
+				return nil, err
+			}
+			opts = append(opts, oci.WithAnnotations(annos))
+		}
 
 		if caps := context.StringSlice("cap-add"); len(caps) > 0 {
 			for _, cap := range caps {

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

@@ -0,0 +1,31 @@
+//go:build linux
+// +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 main
+
+import (
+	"context"
+
+	"github.com/containerd/containerd/runtime/v2/runc/manager"
+	"github.com/containerd/containerd/runtime/v2/shim"
+)
+
+func main() {
+	shim.RunManager(context.Background(), manager.NewShimManager("io.containerd.runc-fp.v1"))
+}

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

@@ -0,0 +1,144 @@
+//go:build linux
+// +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 main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/containerd/containerd/oci"
+	"github.com/containerd/containerd/pkg/failpoint"
+	"github.com/containerd/containerd/pkg/shutdown"
+	"github.com/containerd/containerd/plugin"
+	"github.com/containerd/containerd/runtime/v2/runc/task"
+	"github.com/containerd/containerd/runtime/v2/shim"
+	taskapi "github.com/containerd/containerd/runtime/v2/task"
+	"github.com/containerd/ttrpc"
+)
+
+const (
+	ociConfigFilename = "config.json"
+
+	failpointPrefixKey = "io.containerd.runtime.v2.shim.failpoint."
+)
+
+func init() {
+	plugin.Register(&plugin.Registration{
+		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
+			}
+			fps, err := newFailpointFromOCIAnnotation()
+			if err != nil {
+				return nil, err
+			}
+			service, err := task.NewTaskService(ic.Context, pp.(shim.Publisher), ss.(shutdown.Service))
+			if err != nil {
+				return nil, err
+			}
+
+			return &taskServiceWithFp{
+				fps:   fps,
+				local: service,
+			}, nil
+		},
+	})
+
+}
+
+type taskServiceWithFp struct {
+	fps   map[string]*failpoint.Failpoint
+	local taskapi.TaskService
+}
+
+func (s *taskServiceWithFp) RegisterTTRPC(server *ttrpc.Server) error {
+	taskapi.RegisterTaskService(server, s.local)
+	return nil
+}
+
+func (s *taskServiceWithFp) UnaryInterceptor() ttrpc.UnaryServerInterceptor {
+	return func(ctx context.Context, unmarshal ttrpc.Unmarshaler, info *ttrpc.UnaryServerInfo, method ttrpc.Method) (interface{}, error) {
+		methodName := filepath.Base(info.FullMethod)
+		if fp, ok := s.fps[methodName]; ok {
+			if err := fp.Evaluate(); err != nil {
+				return nil, err
+			}
+		}
+		return method(ctx, unmarshal)
+	}
+}
+
+// newFailpointFromOCIAnnotation reloads and parses the annotation from
+// bundle-path/config.json.
+//
+// The annotation controlling task API's failpoint should be like:
+//
+//	io.containerd.runtime.v2.shim.failpoint.Create = 1*off->1*error(please retry)
+//
+// The `Create` is the shim unary API and the value of annotation is the
+// failpoint control. The function will return a set of failpoint controllers.
+func newFailpointFromOCIAnnotation() (map[string]*failpoint.Failpoint, error) {
+	// NOTE: shim's current working dir is in bundle dir.
+	cwd, err := os.Getwd()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get current working dir: %w", err)
+	}
+
+	configPath := filepath.Join(cwd, ociConfigFilename)
+	data, err := os.ReadFile(configPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read %v: %w", configPath, err)
+	}
+
+	var spec oci.Spec
+	if err := json.Unmarshal(data, &spec); err != nil {
+		return nil, fmt.Errorf("failed to parse oci.Spec(%v): %w", string(data), err)
+	}
+
+	res := make(map[string]*failpoint.Failpoint)
+	for k, v := range spec.Annotations {
+		if !strings.HasPrefix(k, failpointPrefixKey) {
+			continue
+		}
+
+		methodName := strings.TrimPrefix(k, failpointPrefixKey)
+		fp, err := failpoint.NewFailpoint(methodName, v)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse failpoint %v: %w", v, err)
+		}
+		res[methodName] = fp
+	}
+	return res, nil
+}