Test program for simple containerd gRPC use via client API

go-containerd.go

package main

import (
	"bytes"
	"context"
	"fmt"
	"os"
	"strconv"
	"strings"
	"sync"
	"syscall"

	"github.com/containerd/containerd"
	"github.com/containerd/containerd/namespaces"
	"github.com/opencontainers/runtime-spec/specs-go"
)

const (
	defaultContainerdPath = "/run/containerd/containerd.sock"
	defaultImage          = "docker.io/library/alpine:latest"
	defaultNamespace      = "test-pfe"
	defaultName           = "test-ctrd"
	iterations            = 10
)

var (
	containerCmd = "date"
	stdouterr    = bytes.NewBuffer(nil)
	wg           sync.WaitGroup
)

func main() {
	// connect to containerd
	client, err := containerd.New(defaultContainerdPath)
	if err != nil {
		fmt.Printf("can't connect to containerd gRPC (%q): %v\n", defaultContainerdPath, err)
		os.Exit(-1)
	}
	// set up namespace
	ctx := namespaces.WithNamespace(context.Background(), defaultNamespace)

	if err := setup(ctx, client); err != nil {
		fmt.Printf("can't pull image (%q): %v", defaultImage, err)
		client.Close()
		os.Exit(-1)
	}
	if len(os.Args) != 2 {
		fmt.Printf("Please provide an integer argument for the # of routines to start\n")
		os.Exit(-1)
	}
	routines, err := strconv.Atoi(os.Args[1])
	if err != nil {
		fmt.Printf("Please provide an integer argument; could not convert %q to int: %v\n", os.Args[1], err)
		os.Exit(-1)
	}

	for i := 0; i < routines; i++ {
		wg.Add(1)
		go runContainers(ctx, i, iterations)
	}
	wg.Wait()

	if err := client.Close(); err != nil {
		fmt.Printf("error closing gRPC client connection: %v\n", err)
		os.Exit(-1)
	}
}

func setup(ctx context.Context, client *containerd.Client) error {
	// pre pull image
	_, err := client.GetImage(ctx, defaultImage)
	if err != nil {
		// if the image isn't already in our namespaced context, then pull it
		// using the reference and default resolver (most likely DockerHub)
		if _, err = client.Pull(ctx, defaultImage, containerd.WithPullUnpack); err != nil {
			// error pulling the image
			return err
		}
	}
	return nil
}

// go routine in which containers are executed
func runContainers(ctx context.Context, routineNum, iterations int) {

	client, err := containerd.New(defaultContainerdPath)
	if err != nil {
		fmt.Printf("%d: error connecting to client: %v\n", routineNum, err)
		return
	}
	// simple iteration to test creating/running/deleting more than one container per run
	for i := 0; i < iterations; i++ {
		fmt.Printf("%d: Running container; iteration %d\n", routineNum, i)
		runOne(ctx, client, fmt.Sprintf("%s-%d-%d", defaultName, routineNum, i))
	}
	if err := client.Close(); err != nil {
		fmt.Printf("%d: error closing client connection: %v", routineNum, err)
	}
	wg.Done()
}

func runOne(ctx context.Context, client *containerd.Client, name string) {
	image, err := client.GetImage(ctx, defaultImage)
	// create and start container
	var spec *specs.Spec
	if containerCmd != "" {
		// the command needs to be overridden in the generated spec
		spec, err = containerd.GenerateSpec(containerd.WithImageConfig(ctx, image),
			containerd.WithProcessArgs(strings.Split(containerCmd, " ")...))
	} else {
		spec, err = containerd.GenerateSpec(containerd.WithImageConfig(ctx, image))
	}
	if err != nil {
		fmt.Printf("error generating container spec: %v", err)
		os.Exit(-1)
	}
	container, err := client.NewContainer(ctx, name,
		containerd.WithSpec(spec),
		containerd.WithImage(image),
		containerd.WithNewSnapshot(name, image))
	if err != nil {
		fmt.Printf("error creating container: %v", err)
		os.Exit(-1)
	}

	// start task and capture stdout and stderr
	task, err := container.NewTask(ctx, containerd.NewIO(bytes.NewBuffer(nil), stdouterr, stdouterr))
	if err != nil {
		fmt.Printf("error creating task: %v", err)
		os.Exit(-1)
	}
	if err := task.Start(ctx); err != nil {
		task.Delete(ctx)
		fmt.Printf("error starting task: %v", err)
		os.Exit(-1)
	}

	// now call stop and delete container methods to mimic
	// bucketbench loading of these objects via API
	if err := stopContainer(ctx, client, name); err != nil {
		// only error out if the issue is not that the process is already done
		if !strings.Contains(err.Error(), "process already finished") {
			fmt.Printf("error stopping container: %v", err)
			deleteContainer(ctx, client, name)
			os.Exit(-1)
		}
	}
	if err := deleteContainer(ctx, client, name); err != nil {
		fmt.Printf("error deleting container: %v", err)
		os.Exit(-1)
	}
}

func stopContainer(ctx context.Context, client *containerd.Client, name string) error {
	container, err := client.LoadContainer(ctx, name)
	if err != nil {
		return err
	}
	if err = stopTask(ctx, container); err != nil {
		return err
	}
	return nil
}

func deleteContainer(ctx context.Context, client *containerd.Client, name string) error {
	container, err := client.LoadContainer(ctx, name)
	if err != nil {
		return err
	}
	err = container.Delete(ctx, containerd.WithSnapshotCleanup)
	if err != nil {
		return err
	}
	return nil
}

// common code for task stop/kill using the containerd gRPC API
func stopTask(ctx context.Context, ctr containerd.Container) error {
	task, err := ctr.Task(ctx, nil)
	if err != nil {
		if !strings.Contains(err.Error(), "no running task") {
			return err
		}
		//nothing to do; no task running
		return nil
	}
	status, err := task.Status(ctx)
	switch status {
	case containerd.Stopped:
		_, err := task.Delete(ctx)
		if err != nil {
			return err
		}
	case containerd.Running:
		statusC := make(chan uint32, 1)
		go func() {
			status, err := task.Wait(ctx)
			if err != nil {
				fmt.Printf("container %q: error during wait: %v", ctr.ID(), err)
			}
			statusC <- status
		}()
		err := task.Kill(ctx, syscall.SIGKILL)
		if err != nil {
			task.Delete(ctx)
			return err
		}
		status := <-statusC
		if status != 0 {
			fmt.Printf("non-zero container exit code: %d", status)
		}
		_, err = task.Delete(ctx)
		if err != nil {
			return err
		}
	case containerd.Paused:
		return fmt.Errorf("Can't stop a paused container; unpause first")
	}
	return nil
}

Update 7/27/2017 to work with containerd 1.0.0-alpha2