services/introspection: implement plugin reporting

With this change, we integrate all the plugin changes into the
introspection service.

All plugins can be listed with the following command:

```console
$ ctr plugins
TYPE                            ID             PLATFORM       STATUS
io.containerd.content.v1        content        -              ok
io.containerd.metadata.v1       bolt           -              ok
io.containerd.differ.v1         walking        linux/amd64    ok
io.containerd.grpc.v1           containers     -              ok
io.containerd.grpc.v1           content        -              ok
io.containerd.grpc.v1           diff           -              ok
io.containerd.grpc.v1           events         -              ok
io.containerd.grpc.v1           healthcheck    -              ok
io.containerd.grpc.v1           images         -              ok
io.containerd.grpc.v1           namespaces     -              ok
io.containerd.snapshotter.v1    btrfs          linux/amd64    error
io.containerd.snapshotter.v1    overlayfs      linux/amd64    ok
io.containerd.grpc.v1           snapshots      -              ok
io.containerd.monitor.v1        cgroups        linux/amd64    ok
io.containerd.runtime.v1        linux          linux/amd64    ok
io.containerd.grpc.v1           tasks          -              ok
io.containerd.grpc.v1           version        -              ok
```

There are few things to note about this output. The first is that it is
printed in the order in which plugins are initialized. This useful for
debugging plugin initialization problems. Also note that even though the
introspection GPRC api is a itself a plugin, it is not listed. This is
because the plugin takes a snapshot of the initialization state at the
end of the plugin init process. This allows us to see errors from each
plugin, as they happen. If it is required to introspect the existence of
the introspection service, we can make modifications to include it in
the future.

The last thing to note is that the btrfs plugin is in an error state.
This is a common state for containerd because even though we load the
plugin, most installations aren't on top of btrfs and the plugin cannot
be used. We can actually view this error using the detailed view with a
filter:

```console
$ ctr plugins --detailed id==btrfs
Type:          io.containerd.snapshotter.v1
ID:            btrfs
Platforms:     linux/amd64
Exports:
               root      /var/lib/containerd/io.containerd.snapshotter.v1.btrfs
Error:
               Code:        Unknown
               Message:     path /var/lib/containerd/io.containerd.snapshotter.v1.btrfs must be a btrfs filesystem to be used with the btrfs snapshotter
```

Along with several other values, this is a valuable tool for evaluating the
state of components in containerd.

Signed-off-by: Stephen J Day <stephen.day@docker.com>

client.go

@@ -15,6 +15,7 @@ import (
 	diffapi "github.com/containerd/containerd/api/services/diff/v1"
 	eventsapi "github.com/containerd/containerd/api/services/events/v1"
 	imagesapi "github.com/containerd/containerd/api/services/images/v1"
+	introspectionapi "github.com/containerd/containerd/api/services/introspection/v1"
 	namespacesapi "github.com/containerd/containerd/api/services/namespaces/v1"
 	snapshotapi "github.com/containerd/containerd/api/services/snapshot/v1"
 	"github.com/containerd/containerd/api/services/tasks/v1"
@@ -443,6 +444,10 @@ func (c *Client) DiffService() diff.DiffService {
 	return diffservice.NewDiffServiceFromClient(diffapi.NewDiffClient(c.conn))
 }
 
+func (c *Client) IntrospectionService() introspectionapi.IntrospectionClient {
+	return introspectionapi.NewIntrospectionClient(c.conn)
+}
+
 // HealthService returns the underlying GRPC HealthClient
 func (c *Client) HealthService() grpc_health_v1.HealthClient {
 	return grpc_health_v1.NewHealthClient(c.conn)

cmd/containerd/builtins.go

@@ -9,6 +9,7 @@ import (
 	_ "github.com/containerd/containerd/services/events"
 	_ "github.com/containerd/containerd/services/healthcheck"
 	_ "github.com/containerd/containerd/services/images"
+	_ "github.com/containerd/containerd/services/introspection"
 	_ "github.com/containerd/containerd/services/namespaces"
 	_ "github.com/containerd/containerd/services/snapshot"
 	_ "github.com/containerd/containerd/services/tasks"

cmd/containerd/main.go

@@ -82,6 +82,7 @@ func main() {
 			ctx     = log.WithModule(gocontext.Background(), "containerd")
 			config  = defaultConfig()
 		)
+
 		done := handleSignals(ctx, signals, serverC)
 		// start the signal handler as soon as we can to make sure that
 		// we don't miss any signals during boot

cmd/ctr/main.go

@@ -80,6 +80,7 @@ containerd CLI
 		runCommand,
 		snapshotCommand,
 		tasksCommand,
+		pluginsCommand,
 		versionCommand,
 	}, extraCmds...)
 	app.Before = func(context *cli.Context) error {

cmd/ctr/plugins.go

@@ -0,0 +1,139 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"sort"
+	"strings"
+	"text/tabwriter"
+
+	introspection "github.com/containerd/containerd/api/services/introspection/v1"
+	"github.com/containerd/containerd/api/types"
+	"github.com/containerd/containerd/platforms"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/urfave/cli"
+	"google.golang.org/grpc/codes"
+)
+
+var pluginsCommand = cli.Command{
+	Name:  "plugins",
+	Usage: "Provides information about containerd plugins",
+	Flags: []cli.Flag{
+		cli.BoolFlag{
+			Name:  "quiet, q",
+			Usage: "print only the plugin ids",
+		},
+		cli.BoolFlag{
+			Name:  "detailed, d",
+			Usage: "print detailed information about each plugin",
+		},
+	},
+	Action: func(context *cli.Context) error {
+		var (
+			quiet       = context.Bool("quiet")
+			detailed    = context.Bool("detailed")
+			ctx, cancel = appContext(context)
+		)
+		defer cancel()
+
+		client, err := newClient(context)
+		if err != nil {
+			return err
+		}
+		ps := client.IntrospectionService()
+		response, err := ps.Plugins(ctx, &introspection.PluginsRequest{
+			Filters: context.Args(),
+		})
+		if err != nil {
+			return err
+		}
+		if quiet {
+			for _, plugin := range response.Plugins {
+				fmt.Println(plugin.ID)
+			}
+			return nil
+		}
+
+		w := tabwriter.NewWriter(os.Stdout, 4, 8, 4, ' ', 0)
+		if detailed {
+			first := true
+			for _, plugin := range response.Plugins {
+				if !first {
+					fmt.Fprintln(w, "\t\t\t")
+				}
+				first = false
+				fmt.Fprintln(w, "Type:\t", plugin.Type)
+				fmt.Fprintln(w, "ID:\t", plugin.ID)
+				if len(plugin.Requires) > 0 {
+					fmt.Fprintln(w, "Requires:\t")
+					for _, r := range plugin.Requires {
+						fmt.Fprintln(w, "\t", r)
+					}
+				}
+				if len(plugin.Platforms) > 0 {
+					fmt.Fprintln(w, "Platforms:\t", prettyPlatforms(plugin.Platforms))
+				}
+
+				if len(plugin.Exports) > 0 {
+					fmt.Fprintln(w, "Exports:\t")
+					for k, v := range plugin.Exports {
+						fmt.Fprintln(w, "\t", k, "\t", v)
+					}
+				}
+
+				if len(plugin.Capabilities) > 0 {
+					fmt.Fprintln(w, "Capabilities:\t", strings.Join(plugin.Capabilities, ","))
+				}
+
+				if plugin.InitErr != nil {
+					fmt.Fprintln(w, "Error:\t")
+					fmt.Fprintln(w, "\t Code:\t", codes.Code(plugin.InitErr.Code))
+					fmt.Fprintln(w, "\t Message:\t", plugin.InitErr.Message)
+				}
+			}
+			return w.Flush()
+		}
+
+		fmt.Fprintln(w, "TYPE\tID\tPLATFORM\tSTATUS\t")
+		for _, plugin := range response.Plugins {
+			status := "ok"
+
+			if plugin.InitErr != nil {
+				status = "error"
+			}
+
+			var platformColumn = "-"
+			if len(plugin.Platforms) > 0 {
+				platformColumn = prettyPlatforms(plugin.Platforms)
+			}
+
+			if _, err := fmt.Fprintf(w, "%s\t%s\t%s\t%s\t\n",
+				plugin.Type,
+				plugin.ID,
+				platformColumn,
+				status,
+			); err != nil {
+				return err
+			}
+		}
+		return w.Flush()
+	},
+}
+
+func prettyPlatforms(pspb []types.Platform) string {
+	psm := map[string]struct{}{}
+	for _, p := range pspb {
+		psm[platforms.Format(ocispec.Platform{
+			OS:           p.OS,
+			Architecture: p.Architecture,
+			Variant:      p.Variant,
+		})] = struct{}{}
+	}
+	var ps []string
+	for p := range psm {
+		ps = append(ps, p)
+	}
+	sort.Stable(sort.StringSlice(ps))
+
+	return strings.Join(ps, ",")
+}

server/server.go

@@ -15,6 +15,7 @@ import (
 	diff "github.com/containerd/containerd/api/services/diff/v1"
 	eventsapi "github.com/containerd/containerd/api/services/events/v1"
 	images "github.com/containerd/containerd/api/services/images/v1"
+	introspection "github.com/containerd/containerd/api/services/introspection/v1"
 	namespaces "github.com/containerd/containerd/api/services/namespaces/v1"
 	snapshotapi "github.com/containerd/containerd/api/services/snapshot/v1"
 	tasks "github.com/containerd/containerd/api/services/tasks/v1"
@@ -252,6 +253,8 @@ func interceptor(
 		ctx = log.WithModule(ctx, "namespaces")
 	case eventsapi.EventsServer:
 		ctx = log.WithModule(ctx, "events")
+	case introspection.IntrospectionServer:
+		ctx = log.WithModule(ctx, "introspection")
 	default:
 		log.G(ctx).Warnf("unknown GRPC server type: %#v\n", info.Server)
 	}

services/introspection/service.go

@@ -0,0 +1,146 @@
+package introspection
+
+import (
+	api "github.com/containerd/containerd/api/services/introspection/v1"
+	"github.com/containerd/containerd/api/types"
+	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/filters"
+	"github.com/containerd/containerd/plugin"
+	"github.com/containerd/containerd/protobuf/google/rpc"
+	ptypes "github.com/gogo/protobuf/types"
+	context "golang.org/x/net/context"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/status"
+)
+
+func init() {
+	plugin.Register(&plugin.Registration{
+		Type:     plugin.GRPCPlugin,
+		ID:       "introspection",
+		Requires: []plugin.Type{"*"},
+		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
+			// this service works by using the plugin context up till the point
+			// this service is initialized. Since we require this service last,
+			// it should provide the full set of plugins.
+			pluginsPB := pluginsToPB(ic.GetAll())
+			return NewService(pluginsPB), nil
+		},
+	})
+}
+
+type Service struct {
+	plugins []api.Plugin
+}
+
+func NewService(plugins []api.Plugin) api.IntrospectionServer {
+	return &Service{
+		plugins: plugins,
+	}
+}
+
+func (s *Service) Register(server *grpc.Server) error {
+	api.RegisterIntrospectionServer(server, s)
+	return nil
+}
+
+func (s *Service) Plugins(ctx context.Context, req *api.PluginsRequest) (*api.PluginsResponse, error) {
+	filter, err := filters.ParseAll(req.Filters...)
+	if err != nil {
+		return nil, errdefs.ToGRPCf(errdefs.ErrInvalidArgument, err.Error())
+	}
+
+	var plugins []api.Plugin
+	for _, p := range s.plugins {
+		if !filter.Match(adaptPlugin(p)) {
+			continue
+		}
+
+		plugins = append(plugins, p)
+	}
+
+	return &api.PluginsResponse{
+		Plugins: plugins,
+	}, nil
+}
+
+func adaptPlugin(o interface{}) filters.Adaptor {
+	obj := o.(api.Plugin)
+	return filters.AdapterFunc(func(fieldpath []string) (string, bool) {
+		if len(fieldpath) == 0 {
+			return "", false
+		}
+
+		switch fieldpath[0] {
+		case "type":
+			return string(obj.Type), len(obj.Type) > 0
+		case "id":
+			return string(obj.ID), len(obj.ID) > 0
+		case "platforms":
+			// TODO(stevvooe): Another case here where have multiple values.
+			// May need to refactor the filter system to allow filtering by
+			// platform, if this is required.
+		case "capabilities":
+			// TODO(stevvooe): Need a better way to match against
+			// collections. We can only return "the value" but really it
+			// would be best if we could return a set of values for the
+			// path, any of which could match.
+		}
+
+		return "", false
+	})
+}
+
+func pluginsToPB(plugins []*plugin.Plugin) []api.Plugin {
+	var pluginsPB []api.Plugin
+	for _, p := range plugins {
+		var platforms []types.Platform
+		for _, p := range p.Meta.Platforms {
+			platforms = append(platforms, types.Platform{
+				OS:           p.OS,
+				Architecture: p.Architecture,
+				Variant:      p.Variant,
+			})
+		}
+
+		var requires []string
+		for _, r := range p.Registration.Requires {
+			requires = append(requires, r.String())
+		}
+
+		var initErr *rpc.Status
+		if err := p.Err(); err != nil {
+			st, ok := status.FromError(errdefs.ToGRPC(err))
+			if ok {
+				var details []*ptypes.Any
+				for _, d := range st.Proto().Details {
+					details = append(details, &ptypes.Any{
+						TypeUrl: d.TypeUrl,
+						Value:   d.Value,
+					})
+				}
+				initErr = &rpc.Status{
+					Code:    int32(st.Code()),
+					Message: st.Message(),
+					Details: details,
+				}
+			} else {
+				initErr = &rpc.Status{
+					Code:    int32(rpc.Code_UNKNOWN),
+					Message: err.Error(),
+				}
+			}
+		}
+
+		pluginsPB = append(pluginsPB, api.Plugin{
+			Type:         p.Registration.Type.String(),
+			ID:           p.Registration.ID,
+			Requires:     requires,
+			Platforms:    platforms,
+			Capabilities: p.Meta.Capabilities,
+			Exports:      p.Meta.Exports,
+			InitErr:      initErr,
+		})
+	}
+
+	return pluginsPB
+}

services/tasks/service.go

@@ -2,7 +2,6 @@ package tasks
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"