runtime.go

/*
Copyright 2021 The Everoute 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 client

import (
	"archive/tar"
	"bytes"
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"os"
	"path"
	"strings"
	"time"

	"github.com/containerd/containerd"
	"github.com/containerd/containerd/cio"
	"github.com/containerd/containerd/containers"
	"github.com/containerd/containerd/content"
	"github.com/containerd/containerd/errdefs"
	"github.com/containerd/containerd/images"
	"github.com/containerd/containerd/namespaces"
	"github.com/containerd/containerd/oci"
	"github.com/containerd/containerd/platforms"
	"github.com/containerd/containerd/plugin"
	"github.com/containerd/containerd/runtime/restart"
	"github.com/containerd/containerd/runtime/v2/runc/options"
	jsonpatch "github.com/evanphx/json-patch"
	"github.com/opencontainers/go-digest"
	"github.com/opencontainers/image-spec/identity"
	v1 "github.com/opencontainers/image-spec/specs-go/v1"
	"github.com/opencontainers/runtime-spec/specs-go"
	"github.com/samber/lo"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/credentials/insecure"
	"k8s.io/apimachinery/pkg/util/rand"
	"k8s.io/apimachinery/pkg/util/sets"

	"github.com/everoute/container/model"
	"github.com/everoute/container/remotes"
)

type runtime struct {
	platform  platforms.MatchComparer
	namespace string
	client    *containerd.Client
	resolver  remotes.Resolver
	runcPath  string
}

// Options to build a new Runtime
type Options struct {
	Endpoint  string             // containerd endpoint
	Client    *containerd.Client // containerd client
	Namespace string             // containerd namespace
	TLSConfig *tls.Config        // containerd endpoint tls config
	Timeout   time.Duration      // containerd connect timeout
	Provider  remotes.Provider   // containerd image provider
}

// NewRuntime create a new instance of Runtime
func NewRuntime(ctx context.Context, opt Options) (Runtime, error) {
	var client *containerd.Client
	var err error
	var platform platforms.MatchComparer

	if opt.Endpoint == "" {
		opt.Endpoint = "/run/containerd/containerd.sock"
	}

	if opt.Namespace == "" {
		opt.Namespace = "default"
	}

	if opt.Timeout == 0 {
		opt.Timeout = 3 * time.Second
	}

	if opt.Provider == nil {
		opt.Provider = remotes.NewComposeProvider()
	}

	ctx, cancel := context.WithTimeout(ctx, opt.Timeout)
	defer cancel()

	// close connection when the client create in NewRuntime
	defer func() {
		if err != nil && opt.Client == nil && client != nil {
			client.Close()
		}
	}()

	switch {
	case opt.Client != nil: // use client from options
		client = opt.Client
	case strings.HasPrefix(opt.Endpoint, "/"): // unix socket
		client, err = containerd.New(opt.Endpoint, containerd.WithTimeout(opt.Timeout))
	default:
		client, err = newTCPClient(ctx, opt.Endpoint, opt.TLSConfig, opt.Timeout)
	}
	if err != nil {
		return nil, err
	}

	// get platform that the containerd support
	platform, err = client.GetSnapshotterSupportedPlatforms(ctx, containerd.DefaultSnapshotter)
	if err != nil {
		return nil, err
	}

	if injectable, ok := opt.Provider.(remotes.ContainerdClientInjectable); ok {
		err = injectable.WithContainerdClient(ctx, client)
		if err != nil {
			return nil, err
		}
	}

	r := &runtime{
		platform:  platform,
		namespace: opt.Namespace,
		client:    client,
		resolver:  remotes.ProviderResolver{Provider: opt.Provider},
	}
	return r, nil
}

func newTCPClient(ctx context.Context, endpoint string, tlsConfig *tls.Config, timeout time.Duration) (*containerd.Client, error) {
	var opts = []grpc.DialOption{grpc.WithBlock()}
	if tlsConfig == nil {
		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
	} else {
		opts = append(opts, grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)))
	}

	conn, err := grpc.DialContext(ctx, endpoint, opts...)
	if err != nil {
		return nil, err
	}

	// containerd.NewWithConn always use io.containerd.runtime.v1/linux as runtime
	// so must specify runtime(io.containerd.runc.v2) when create container.
	return containerd.NewWithConn(conn, containerd.WithTimeout(timeout))
}

func (r *runtime) Platform() platforms.MatchComparer    { return r.platform }
func (r *runtime) ContainerdClient() *containerd.Client { return r.client }
func (r *runtime) Namespace() string                    { return r.namespace }

func (r *runtime) NodeExecute(ctx context.Context, name string, commands ...string) error {
	return r.execHostCommand(ctx, name, commands...)
}

func (r *runtime) ConfigRuntime(ctx context.Context) error {
	return r.doConfig(ctx)
}

func (r *runtime) ImportImages(ctx context.Context, refs ...string) error {
	ctx = namespaces.WithNamespace(ctx, r.namespace)

	ctx, cancel, err := r.client.WithLease(ctx)
	if err != nil {
		return fmt.Errorf("add lease: %w", err)
	}
	defer func() { _ = cancel(ctx) }()

	for _, ref := range refs {
		// fix: pull with unpack do not fetch missing contents
		img, err := r.client.Fetch(ctx, ref, containerd.WithPlatformMatcher(r.platform), containerd.WithResolver(r.resolver))
		if err != nil {
			return fmt.Errorf("import %s: %w", ref, err)
		}

		err = containerd.NewImageWithPlatform(r.client, img, r.platform).Unpack(ctx, containerd.DefaultSnapshotter)
		if err != nil {
			return fmt.Errorf("unpack %s: %w", ref, err)
		}
	}
	return nil
}

func (r *runtime) ListImages(ctx context.Context) ([]images.Image, error) {
	ctx = namespaces.WithNamespace(ctx, r.namespace)
	return r.client.ImageService().List(ctx)
}

func (r *runtime) RemoveImage(ctx context.Context, ref string) error {
	ctx = namespaces.WithNamespace(ctx, r.namespace)
	err := r.client.ImageService().Delete(ctx, ref, images.SynchronousDelete())
	return ignoreNotFoundError(err)
}

func (r *runtime) GetImage(ctx context.Context, ref string) (*images.Image, bool, error) {
	ctx = namespaces.WithNamespace(ctx, r.namespace)

	img, err := r.getImage(ctx, ref)
	if err != nil {
		return nil, false, ignoreNotFoundError(err)
	}

	var imgMetadata = img.Metadata()
	return &imgMetadata, true, nil
}

func (r *runtime) getImage(ctx context.Context, ref string) (containerd.Image, error) {
	i, err := r.client.ImageService().Get(ctx, ref)
	if err != nil {
		return nil, err
	}
	return containerd.NewImageWithPlatform(r.client, i, r.platform), nil
}

func (r *runtime) RecommendedRuntimeInfo(ctx context.Context, container *model.Container) *containers.RuntimeInfo {
	cc := &containers.Container{}
	lo.Must0(withRuntime(r.runcPath, container)(ctx, r.client, cc))
	return &cc.Runtime
}

func (r *runtime) CreateContainer(ctx context.Context, container *model.Container, following bool) error {
	ctx = namespaces.WithNamespace(ctx, r.namespace)

	image, err := r.getImage(ctx, container.Image)
	if err != nil {
		return fmt.Errorf("get image %s: %w", container.Image, err)
	}

	nc, err := r.client.NewContainer(ctx, container.Name,
		containerd.WithImageName(container.Image),
		withNewSnapshotAndConfig(image, container.ConfigContent),
		withLogPath(container.Process.LogPath),
		withRuntime(r.runcPath, container),
		containerd.WithNewSpec(containerSpecOpts(r.namespace, image, container)...),
	)
	if err != nil {
		return fmt.Errorf("create container: %w", err)
	}

	creator := cio.LogFile(container.Process.LogPath)
	if following && container.Process.LogPath == model.StdOutputStream {
		creator = cio.NewCreator(cio.WithStreams(nil, os.Stdout, os.Stderr))
	}

	task, err := r.newTask(ctx, nc, creator)
	if err != nil {
		return fmt.Errorf("create task: %w", err)
	}

	if following {
		return HandleTaskResult(ExecTask(ctx, task))
	}

	err = task.Start(ctx)
	if err != nil {
		return err
	}

	if container.Process.RestartPolicy == model.RestartPolicyAlways {
		err = nc.Update(ctx, withLogPath(container.Process.LogPath), restart.WithStatus(containerd.Running))
		if err != nil {
			return err
		}
	}

	return nil
}

func (r *runtime) UpdateContainer(ctx context.Context, container *model.Container, opts *ContainerUpdateOptions) error {
	ctx = namespaces.WithNamespace(ctx, r.namespace)

	image, err := r.getImage(ctx, container.Image)
	if err != nil {
		return fmt.Errorf("get image %s: %w", container.Image, err)
	}

	updateOptions := []containerd.UpdateContainerOpts{
		containerd.UpdateContainerOpts(containerd.WithImageName(container.Image)),
		containerd.UpdateContainerOpts(withLogPath(container.Process.LogPath)),
		// containerd.UpdateContainerOpts(withRuntime(r.runcPath, container)), fixme: runtime donot support update
		containerd.UpdateContainerOpts(containerd.WithNewSpec(containerSpecOpts(r.namespace, image, container)...)),
	}
	if container.Process.RestartPolicy == model.RestartPolicyAlways {
		updateOptions = append(updateOptions, restart.WithStatus(containerd.Running))
	}
	if opts.UpdateSnapshot {
		updateOptions = append(updateOptions, containerd.UpdateContainerOpts(withNewSnapshotAndConfig(image, container.ConfigContent)))
	}

	c, err := r.client.LoadContainer(ctx, container.Name)
	if err != nil {
		return fmt.Errorf("load container: %w", err)
	}
	err = c.Update(ctx, updateOptions...)
	if err != nil {
		return fmt.Errorf("update container: %w", err)
	}

	task, err := c.Task(ctx, nil)
	if err != nil {
		if errdefs.IsNotFound(err) {
			return nil
		}
		return fmt.Errorf("load task: %w", err)
	}

	spec, err := c.Spec(ctx)
	if err != nil {
		return fmt.Errorf("get container spec: %w", err)
	}
	return task.Update(ctx, containerd.WithResources(spec.Linux.Resources))
}

func (r *runtime) RemoveContainer(ctx context.Context, containerID string) error {
	ctx = namespaces.WithNamespace(ctx, r.namespace)

	container, err := r.client.LoadContainer(ctx, containerID)
	if err != nil {
		if errdefs.IsNotFound(err) {
			return nil
		}
		return err
	}

	_ = container.Update(ctx, func(_ context.Context, _ *containerd.Client, c *containers.Container) error {
		delete(c.Labels, restart.StatusLabel)
		return nil
	})

	task, err := container.Task(ctx, nil)
	if err != nil && !errdefs.IsNotFound(err) {
		return err
	}
	if err == nil {
		_, err = task.Delete(ctx, containerd.WithProcessKill)
		if err != nil && !errdefs.IsNotFound(err) {
			return err
		}
	}

	return container.Delete(ctx, containerd.WithSnapshotCleanup)
}

func (r *runtime) GetContainer(ctx context.Context, containerID string) (*model.Container, error) {
	ctx = namespaces.WithNamespace(ctx, r.namespace)
	container, err := r.client.ContainerService().Get(ctx, containerID)
	if err != nil {
		return nil, err
	}
	return parseContainer(container)
}

func (r *runtime) ListContainers(ctx context.Context) ([]*model.Container, error) {
	ctx = namespaces.WithNamespace(ctx, r.namespace)
	cs, err := r.client.ContainerService().List(ctx)
	if err != nil {
		return nil, err
	}
	containerList := make([]*model.Container, 0, len(cs))
	for _, c := range cs {
		parsedContainer, err := parseContainer(c)
		if err != nil {
			return nil, fmt.Errorf("parse %s: %w", c.ID, err)
		}
		containerList = append(containerList, parsedContainer)
	}
	return containerList, nil
}

func (r *runtime) RemoveNamespace(ctx context.Context) error {
	err := r.client.NamespaceService().Delete(ctx, r.namespace)
	return ignoreNotFoundError(err)
}

func (r *runtime) GetContainerStatus(ctx context.Context, containerID string) (ContainerStatus, error) {
	ctx = namespaces.WithNamespace(ctx, r.namespace)

	c, err := r.client.LoadContainer(ctx, containerID)
	if err != nil {
		return ContainerStatus{}, fmt.Errorf("load container: %w", err)
	}

	task, err := c.Task(ctx, nil)
	if err != nil {
		return ContainerStatus{}, fmt.Errorf("load task: %w", err)
	}

	status, err := task.Status(ctx)
	if err != nil {
		return ContainerStatus{}, fmt.Errorf("get task status: %w", err)
	}

	return ContainerStatus{
		Status:    status,
		Task:      task,
		Container: lo.Must(c.Info(ctx, containerd.WithoutRefreshedMetadata)),
	}, nil
}

func (r *runtime) ExecCommand(ctx context.Context, containerID string, commands []string) (*containerd.ExitStatus, error) {
	ctx = namespaces.WithNamespace(ctx, r.namespace)

	c, err := r.client.LoadContainer(ctx, containerID)
	if err != nil {
		return nil, fmt.Errorf("load container: %w", err)
	}

	task, err := c.Task(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("load task: %w", err)
	}

	spec, err := c.Spec(ctx)
	if err != nil {
		return nil, fmt.Errorf("load task spec: %w", err)
	}

	taskExecID := "exec-" + rand.String(10)
	progressSpec := spec.Process
	progressSpec.Terminal = false
	progressSpec.Args = commands

	progress, err := task.Exec(ctx, taskExecID, progressSpec, cio.LogFile(os.DevNull))
	if err != nil {
		return nil, fmt.Errorf("exec command: %w", err)
	}
	return ExecTask(ctx, progress)
}

func (r *runtime) Close() error {
	return r.client.Close()
}

func (r *runtime) doConfig(ctx context.Context) error {
	if err := r.doPlatformConfig(ctx); err != nil {
		return err
	}
	return nil
}

func ContainerSpecOpts(namespace string, img containerd.Image, container *model.Container) []oci.SpecOpts {
	return containerSpecOpts(namespace, img, container)
}

func containerSpecOpts(namespace string, img containerd.Image, container *model.Container) []oci.SpecOpts {
	var specOpts []oci.SpecOpts
	specOpts = append(specOpts, oci.WithProcessCwd(container.Process.WorkingDir))
	specOpts = append(specOpts, oci.WithProcessArgs(container.Process.Args...))
	specOpts = append(specOpts, withCgroupParent(namespace, container))
	specOpts = append(specOpts, oci.WithEnv(container.Process.Env))
	specOpts = append(specOpts, oci.WithDefaultPathEnv)
	specOpts = append(specOpts, oci.WithMounts(container.Mounts))
	specOpts = append(specOpts, oci.WithHostname("localhost"))
	specOpts = append(specOpts, oci.WithHostNamespace(specs.NetworkNamespace))
	mountTargets := sets.NewString(lo.Map(container.Mounts, func(m specs.Mount, _ int) string { return m.Destination })...)
	if !mountTargets.Has("/etc/hosts") {
		specOpts = append(specOpts, oci.WithHostHostsFile)
	}
	if !mountTargets.Has("/etc/resolv.conf") {
		specOpts = append(specOpts, oci.WithHostResolvconf)
	}
	if container.Privilege {
		specOpts = append(specOpts, oci.WithPrivileged)
	}
	specOpts = append(specOpts, oci.WithAddedCapabilities(container.Capabilities))
	for _, device := range container.Devices {
		devicePath, containerPath := device, ""
		if strings.Contains(device, ":") {
			devicePath, containerPath = strings.Split(device, ":")[0], strings.Split(device, ":")[1]
		}
		specOpts = append(specOpts, oci.WithDevices(devicePath, containerPath, "rwm"))
	}
	if img != nil {
		specOpts = append(specOpts, withImageENV(img))
	}
	if container.MemoryLimit > 0 {
		specOpts = append(specOpts, oci.WithMemoryLimit(container.MemoryLimit))
	}
	if container.CPUQuota > 0 && container.CPUPeriod > 0 {
		specOpts = append(specOpts, oci.WithCPUCFS(container.CPUQuota, container.CPUPeriod))
	}
	specOpts = append(specOpts, withRlimits(container.Rlimits))
	specOpts = append(specOpts, withSpecPatches(container.SpecPatches))
	specOpts = append(specOpts, withRuntimeENV(namespace, container))
	return specOpts
}

func parseContainer(container containers.Container) (*model.Container, error) {
	spec := &specs.Spec{}

	if err := json.Unmarshal(container.Spec.Value, spec); err != nil {
		return nil, err
	}

	c := &model.Container{
		Name:   container.ID,
		Image:  container.Image,
		Mounts: spec.Mounts,
		Process: model.Process{
			Args:       spec.Process.Args,
			Env:        spec.Process.Env,
			WorkingDir: spec.Process.Cwd,
			LogPath:    container.Labels[restart.LogPathLabel],
		},
	}
	return c, nil
}

func toRawConfig(config []model.ConfigFile) []byte {
	var rawData bytes.Buffer
	tw := tar.NewWriter(&rawData)
	defer tw.Close()

	for _, file := range config {
		_ = tw.WriteHeader(&tar.Header{
			Name: file.Path,
			Size: int64(len(file.FileContent)),
			Mode: int64(file.FileMode),
		})
		_, _ = tw.Write(file.FileContent)
	}

	return rawData.Bytes()
}

func withoutAnyMounts() oci.SpecOpts {
	return func(_ context.Context, _ oci.Client, _ *containers.Container, spec *oci.Spec) error {
		spec.Mounts = nil
		return nil
	}
}

func withNewSnapshotAndConfig(img containerd.Image, configContent []model.ConfigFile) containerd.NewContainerOpts {
	return func(ctx context.Context, client *containerd.Client, c *containers.Container) error {
		var (
			snapshotID = rand.String(10)
			data       = toRawConfig(configContent)
			descriptor = v1.Descriptor{
				MediaType: v1.MediaTypeImageLayer,
				Digest:    digest.SHA256.FromBytes(data),
				Size:      int64(len(data)),
			}
			ref = fmt.Sprintf("ingest-%s", descriptor.Digest)
		)

		diffIDs, err := img.RootFS(ctx)
		if err != nil {
			return err
		}

		mounts, err := client.SnapshotService(containerd.DefaultSnapshotter).Prepare(ctx, snapshotID, identity.ChainID(diffIDs).String())
		if err != nil {
			return err
		}

		err = content.WriteBlob(ctx, client.ContentStore(), ref, bytes.NewReader(data), descriptor)
		if err != nil {
			return fmt.Errorf("write config content: %w", err)
		}

		if _, err = client.DiffService().Apply(ctx, descriptor, mounts); err != nil {
			return err
		}

		c.Snapshotter = containerd.DefaultSnapshotter
		c.SnapshotKey = snapshotID
		return nil
	}
}

func withImageENV(img containerd.Image) oci.SpecOpts {
	return func(ctx context.Context, _ oci.Client, _ *containers.Container, s *oci.Spec) error {
		ic, err := img.Config(ctx)
		if err != nil {
			return err
		}
		var (
			ociimage v1.Image
			config   v1.ImageConfig
		)
		switch ic.MediaType {
		case v1.MediaTypeImageConfig, images.MediaTypeDockerSchema2Config:
			p, err := content.ReadBlob(ctx, img.ContentStore(), ic)
			if err != nil {
				return err
			}

			if err := json.Unmarshal(p, &ociimage); err != nil {
				return err
			}
			config = ociimage.Config
		default:
			return fmt.Errorf("unknown image config media type %s", ic.MediaType)
		}

		s.Process.Env = sets.NewString(append(config.Env, s.Process.Env...)...).List()
		return nil
	}
}

func withLogPath(logPath string) func(ctx context.Context, client *containerd.Client, c *containers.Container) error {
	return func(_ context.Context, _ *containerd.Client, c *containers.Container) error {
		if c.Labels == nil {
			c.Labels = make(map[string]string)
		}

		if logPath == model.StdOutputStream {
			return nil
		}

		uri, err := cio.LogURIGenerator("file", logPath, nil)
		if err != nil {
			return err
		}

		c.Labels[restart.LogPathLabel] = logPath
		c.Labels[restart.LogURILabel] = uri.String()
		return nil
	}
}

func GetLogPath(c *containers.Container) string {
	if c.Labels[restart.LogURILabel] != "" {
		return strings.TrimPrefix(c.Labels[restart.LogURILabel], "file://")
	}
	if c.Labels[restart.LogPathLabel] != "" {
		return c.Labels[restart.LogPathLabel]
	}
	return ""
}

func withRlimits(rlimits []specs.POSIXRlimit) oci.SpecOpts {
	return func(_ context.Context, _ oci.Client, _ *containers.Container, spec *oci.Spec) error {
		if spec.Process == nil {
			spec.Process = &specs.Process{}
		}
		rlimitsMap := make(map[string]specs.POSIXRlimit)
		for _, rlimit := range append(spec.Process.Rlimits, rlimits...) {
			rlimitsMap[rlimit.Type] = rlimit
		}
		spec.Process.Rlimits = make([]specs.POSIXRlimit, 0, len(rlimitsMap))
		for _, rlimit := range rlimitsMap {
			spec.Process.Rlimits = append(spec.Process.Rlimits, rlimit)
		}
		return nil
	}
}

func withSpecPatches(specPatches []json.RawMessage) oci.SpecOpts {
	opts := make([]oci.SpecOpts, 0, len(specPatches))
	for _, specPatch := range specPatches {
		opts = append(opts, withSpecPatch(specPatch))
	}
	return oci.Compose(opts...)
}

func withSpecPatch(specPatch json.RawMessage) oci.SpecOpts {
	return func(_ context.Context, _ oci.Client, _ *containers.Container, spec *oci.Spec) error {
		if len(specPatch) == 0 {
			return nil
		}
		patch, err := jsonpatch.DecodePatch(specPatch)
		if err != nil {
			return fmt.Errorf("invalid spec-patch(%s): %w", string(specPatch), err)
		}
		rawSpec, err := json.Marshal(spec)
		if err != nil {
			return fmt.Errorf("marshal spec as json: %w", err)
		}
		patSpec, err := patch.Apply(rawSpec)
		if err != nil {
			return fmt.Errorf("patch container spec: %w", err)
		}
		return json.Unmarshal(patSpec, spec)
	}
}

func withRuntime(defaultRuncPath string, container *model.Container) containerd.NewContainerOpts {
	binaryName := container.Runtime.BinaryName
	if binaryName == "" {
		binaryName = defaultRuncPath
	}
	return containerd.WithRuntime(plugin.RuntimeRuncV2, &options.Options{
		NoPivotRoot:   container.Runtime.NoPivotRoot,
		BinaryName:    binaryName,
		SystemdCgroup: container.Runtime.SystemdCgroup,
	})
}

func withCgroupParent(namespace string, container *model.Container) oci.SpecOpts {
	if container.Runtime.SystemdCgroup {
		return oci.WithCgroup(container.CgroupParent)
	}
	return oci.WithCgroup(path.Join(container.CgroupParent, namespace, container.Name))
}

func ignoreNotFoundError(err error) error {
	if errdefs.IsNotFound(err) {
		return nil
	}
	return err
}

func withRuntimeENV(namespace string, container *model.Container) oci.SpecOpts {
	return oci.WithEnv([]string{
		fmt.Sprintf("%s=%s", ENVRuntimeContainerNamespace, namespace),
		fmt.Sprintf("%s=%s", ENVRuntimeContainerName, container.Name),
		fmt.Sprintf("%s=%s", ENVRuntimeContainerImage, container.Image),
	})
}