Create server package for containerd daemon

Signed-off-by: Michael Crosby <crosbymichael@gmail.com>

cmd/containerd/main.go

@@ -1,10 +1,9 @@
 package main
 
 import (
-	_ "expvar"
+	"context"
 	"fmt"
-	"net/http"
-	_ "net/http/pprof"
+	"net"
 	"os"
 	"os/signal"
 	"path/filepath"
@@ -12,27 +11,15 @@ import (
 	"time"
 
 	"github.com/boltdb/bolt"
-	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 	gocontext "golang.org/x/net/context"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/health/grpc_health_v1"
 
 	"github.com/Sirupsen/logrus"
-	containersapi "github.com/containerd/containerd/api/services/containers/v1"
-	contentapi "github.com/containerd/containerd/api/services/content/v1"
-	diffapi "github.com/containerd/containerd/api/services/diff/v1"
-	imagesapi "github.com/containerd/containerd/api/services/images/v1"
-	namespacesapi "github.com/containerd/containerd/api/services/namespaces/v1"
-	snapshotapi "github.com/containerd/containerd/api/services/snapshot/v1"
-	api "github.com/containerd/containerd/api/services/tasks/v1"
-	versionapi "github.com/containerd/containerd/api/services/version/v1"
 	"github.com/containerd/containerd/content"
-	"github.com/containerd/containerd/events"
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/plugin"
+	"github.com/containerd/containerd/server"
 	"github.com/containerd/containerd/sys"
 	"github.com/containerd/containerd/version"
-	metrics "github.com/docker/go-metrics"
 	"github.com/pkg/errors"
 	"github.com/urfave/cli"
 )
@@ -47,11 +34,6 @@ const usage = `
 high performance container runtime
 `
 
-var (
-	conf   = defaultConfig()
-	global = log.WithModule(gocontext.Background(), "containerd")
-)
-
 func init() {
 	cli.VersionPrinter = func(c *cli.Context) {
 		fmt.Println(c.App.Name, version.Package, c.App.Version)
@@ -73,10 +55,6 @@ func main() {
 			Name:  "log-level,l",
 			Usage: "set the logging level [debug, info, warn, error, fatal, panic]",
 		},
-		cli.StringFlag{
-			Name:  "state",
-			Usage: "containerd state directory",
-		},
 		cli.StringFlag{
 			Name:  "address,a",
 			Usage: "address for containerd's GRPC server",
@@ -89,81 +67,66 @@ func main() {
 	app.Commands = []cli.Command{
 		configCommand,
 	}
-	app.Before = before
 	app.Action = func(context *cli.Context) error {
-		start := time.Now()
+		var (
+			start   = time.Now()
+			signals = make(chan os.Signal, 2048)
+			ctx     = log.WithModule(gocontext.Background(), "containerd")
+			config  = defaultConfig()
+		)
 		// start the signal handler as soon as we can to make sure that
 		// we don't miss any signals during boot
-		signals := make(chan os.Signal, 2048)
 		signal.Notify(signals, handledSignals...)
-		if err := platformInit(context); err != nil {
-			return err
-		}
-		log.G(global).Info("starting containerd boot...")
 
-		// load all plugins into containerd
-		if err := plugin.Load(filepath.Join(conf.Root, "plugins")); err != nil {
+		if err := server.LoadConfig(context.GlobalString("config"), config); err != nil && !os.IsNotExist(err) {
 			return err
 		}
-		registerContentStore()
-		registerMetaDB()
-		// start debug and metrics APIs
-		if err := serveDebugAPI(); err != nil {
+		// apply flags to the config
+		if err := applyFlags(context, config); err != nil {
+			return err
+		}
+		address := config.GRPC.Address
+		if address == "" {
+			return errors.New("grpc address cannot be empty")
+		}
+
+		if err := platformInit(ctx, config); err != nil {
+			return err
+		}
+		log.G(ctx).Info("starting containerd boot...")
+
+		plugins, err := loadPlugins(config)
+		if err != nil {
 			return err
 		}
 
-		var (
-			services     []plugin.Service
-			plugins      = make(map[plugin.PluginType][]interface{})
-			eventEmitter = events.NewEmitter()
-		)
-		for _, init := range plugin.Graph() {
-			id := init.URI()
-			if !shouldLoad(init) {
-				log.G(global).WithField("type", init.Type).Infof("skipping plugin %q...", id)
-				continue
-			}
-			log.G(global).WithField("type", init.Type).Infof("loading plugin %q...", id)
-			ic := plugin.NewContext(plugins)
-			ic.Root = filepath.Join(conf.Root, id)
-
-			rCtx := events.WithPoster(global, eventEmitter)
-			ic.Context = log.WithModule(rCtx, id)
-			if init.Config != nil {
-				if err := loadPluginConfig(init.ID, init.Config, ic); err != nil {
-					return err
-				}
-			}
-			ic.Emitter = eventEmitter
-
-			p, err := init.Init(ic)
+		server, err := server.New(ctx, config, plugins)
+		if err != nil {
+			return err
+		}
+		if config.Debug.Address != "" {
+			l, err := sys.GetLocalListener(config.Debug.Address, config.Debug.Uid, config.Debug.Gid)
 			if err != nil {
 				return err
 			}
-			plugins[init.Type] = append(plugins[init.Type], p)
-			if s, ok := p.(plugin.Service); ok {
-				services = append(services, s)
-			}
+			serve(log.WithModule(ctx, "debug"), l, server.ServeDebug)
 		}
-
-		// start the GRPC api with the execution service registered
-		server := newGRPCServer()
-		for _, service := range services {
-			if err := service.Register(server); err != nil {
+		if config.Metrics.Address != "" {
+			l, err := net.Listen("tcp", config.Metrics.Address)
+			if err != nil {
 				return err
 			}
+			serve(log.WithModule(ctx, "metrics"), l, server.ServeMetrics)
 		}
-		// register metrics last after all other services
-		grpc_prometheus.Register(server)
-		log.G(global).Info("starting GRPC API server...")
-		if err := serveGRPC(server); err != nil {
+
+		l, err := sys.GetLocalListener(address, config.GRPC.Uid, config.GRPC.Gid)
+		if err != nil {
 			return err
 		}
-		// start the prometheus metrics API for containerd
-		serveMetricsAPI()
+		serve(log.WithModule(ctx, "grpc"), l, server.ServeGRPC)
 
-		log.G(global).Infof("containerd successfully booted in %fs", time.Since(start).Seconds())
-		return handleSignals(signals, server)
+		log.G(ctx).Infof("containerd successfully booted in %fs", time.Since(start).Seconds())
+		return handleSignals(ctx, signals, server)
 	}
 	if err := app.Run(os.Args); err != nil {
 		fmt.Fprintf(os.Stderr, "containerd: %s\n", err)
@@ -171,14 +134,21 @@ func main() {
 	}
 }
 
-func before(context *cli.Context) error {
-	err := loadConfig(context.GlobalString("config"))
-	if err != nil && !os.IsNotExist(err) {
-		return err
-	}
+func serve(ctx context.Context, l net.Listener, serveFunc func(net.Listener) error) {
+	path := l.Addr().String()
+	log.G(ctx).WithField("address", path).Info("serving...")
+	go func() {
+		defer l.Close()
+		if err := serveFunc(l); err != nil {
+			log.G(ctx).WithError(err).WithField("address", path).Fatal("serve failure")
+		}
+	}()
+}
+
+func applyFlags(context *cli.Context, config *server.Config) error {
 	// the order for config vs flag values is that flags will always override
 	// the config values if they are set
-	if err := setLevel(context); err != nil {
+	if err := setLevel(context, config); err != nil {
 		return err
 	}
 	for _, v := range []struct {
@@ -187,11 +157,11 @@ func before(context *cli.Context) error {
 	}{
 		{
 			name: "root",
-			d:    &conf.Root,
+			d:    &config.Root,
 		},
 		{
 			name: "address",
-			d:    &conf.GRPC.Address,
+			d:    &config.GRPC.Address,
 		},
 	} {
 		if s := context.GlobalString(v.name); s != "" {
@@ -201,10 +171,10 @@ func before(context *cli.Context) error {
 	return nil
 }
 
-func setLevel(context *cli.Context) error {
+func setLevel(context *cli.Context, config *server.Config) error {
 	l := context.GlobalString("log-level")
 	if l == "" {
-		l = conf.Debug.Level
+		l = config.Debug.Level
 	}
 	if l != "" {
 		lvl, err := logrus.ParseLevel(l)
@@ -216,43 +186,16 @@ func setLevel(context *cli.Context) error {
 	return nil
 }
 
-func serveMetricsAPI() {
-	if conf.Metrics.Address != "" {
-		log.G(global).WithField("metrics", conf.Metrics.Address).Info("starting metrics API...")
-		h := newMetricsHandler()
-		go func() {
-			if err := http.ListenAndServe(conf.Metrics.Address, h); err != nil {
-				log.G(global).WithError(err).Fatal("serve metrics API")
-			}
-		}()
+func loadPlugins(config *server.Config) ([]*plugin.Registration, error) {
+	// load all plugins into containerd
+	if err := plugin.Load(filepath.Join(config.Root, "plugins")); err != nil {
+		return nil, err
 	}
-}
-
-func newMetricsHandler() http.Handler {
-	m := http.NewServeMux()
-	m.Handle("/metrics", metrics.Handler())
-	return m
-}
-
-func serveDebugAPI() error {
-	path := conf.Debug.Address
-	if path == "" {
-		return errors.New("debug socket path cannot be empty")
-	}
-	l, err := sys.GetLocalListener(path, conf.GRPC.Uid, conf.GRPC.Gid)
-	if err != nil {
-		return err
-	}
-	log.G(global).WithField("debug", path).Info("starting debug API...")
-	go func() {
-		defer l.Close()
-		// pprof and expvars are imported and automatically register their endpoints
-		// under /debug
-		if err := http.Serve(l, nil); err != nil {
-			log.G(global).WithError(err).Fatal("serve debug API")
-		}
-	}()
-	return nil
+	// load additional plugins that don't automatically register themselves
+	registerContentStore()
+	registerMetaDB()
+	// return the ordered graph for plugins
+	return plugin.Graph(), nil
 }
 
 func registerContentStore() {
@@ -278,63 +221,6 @@ func registerMetaDB() {
 	})
 }
 
-func newGRPCServer() *grpc.Server {
-	s := grpc.NewServer(
-		grpc.UnaryInterceptor(interceptor),
-		grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor),
-	)
-	return s
-}
-
-func serveGRPC(server *grpc.Server) error {
-	path := conf.GRPC.Address
-	if path == "" {
-		return errors.New("--socket path cannot be empty")
-	}
-	l, err := sys.GetLocalListener(path, conf.GRPC.Uid, conf.GRPC.Gid)
-	if err != nil {
-		return err
-	}
-	go func() {
-		defer l.Close()
-		if err := server.Serve(l); err != nil {
-			log.G(global).WithError(err).Fatal("serve GRPC")
-		}
-	}()
-	return nil
-}
-
-func interceptor(ctx gocontext.Context,
-	req interface{},
-	info *grpc.UnaryServerInfo,
-	handler grpc.UnaryHandler,
-) (interface{}, error) {
-	ctx = log.WithModule(ctx, "containerd")
-	switch info.Server.(type) {
-	case api.TasksServer:
-		ctx = log.WithModule(ctx, "execution")
-	case containersapi.ContainersServer:
-		ctx = log.WithModule(ctx, "containers")
-	case contentapi.ContentServer:
-		ctx = log.WithModule(ctx, "content")
-	case imagesapi.ImagesServer:
-		ctx = log.WithModule(ctx, "images")
-	case grpc_health_v1.HealthServer:
-		// No need to change the context
-	case versionapi.VersionServer:
-		ctx = log.WithModule(ctx, "version")
-	case snapshotapi.SnapshotsServer:
-		ctx = log.WithModule(ctx, "snapshot")
-	case diffapi.DiffServer:
-		ctx = log.WithModule(ctx, "diff")
-	case namespacesapi.NamespacesServer:
-		ctx = log.WithModule(ctx, "namespaces")
-	default:
-		log.G(ctx).Warnf("unknown GRPC server type: %#v\n", info.Server)
-	}
-	return grpc_prometheus.UnaryServerInterceptor(ctx, req, info, handler)
-}
-
 func dumpStacks() {
 	var (
 		buf       []byte
@@ -349,23 +235,3 @@ func dumpStacks() {
 	buf = buf[:stackSize]
 	logrus.Infof("=== BEGIN goroutine stack dump ===\n%s\n=== END goroutine stack dump ===", buf)
 }
-
-func loadPluginConfig(name string, c interface{}, ic *plugin.InitContext) error {
-	if err := conf.decodePlugin(name, c); err != nil {
-		return err
-	}
-	ic.Config = c
-	return nil
-}
-
-func shouldLoad(r *plugin.Registration) bool {
-	// only load certain plugins based on the config values
-	switch r.Type {
-	case plugin.SnapshotPlugin:
-		return r.URI() == conf.Snapshotter
-	case plugin.DiffPlugin:
-		return r.URI() == conf.Differ
-	default:
-		return true
-	}
-}