279 lines
8.2 KiB
Go
279 lines
8.2 KiB
Go
package server
|
|
|
|
import (
|
|
"expvar"
|
|
"net"
|
|
"net/http"
|
|
"net/http/pprof"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
|
|
"github.com/boltdb/bolt"
|
|
containers "github.com/containerd/containerd/api/services/containers/v1"
|
|
contentapi "github.com/containerd/containerd/api/services/content/v1"
|
|
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"
|
|
leasesapi "github.com/containerd/containerd/api/services/leases/v1"
|
|
namespaces "github.com/containerd/containerd/api/services/namespaces/v1"
|
|
snapshotsapi "github.com/containerd/containerd/api/services/snapshots/v1"
|
|
tasks "github.com/containerd/containerd/api/services/tasks/v1"
|
|
version "github.com/containerd/containerd/api/services/version/v1"
|
|
"github.com/containerd/containerd/content"
|
|
"github.com/containerd/containerd/content/local"
|
|
"github.com/containerd/containerd/events/exchange"
|
|
"github.com/containerd/containerd/log"
|
|
"github.com/containerd/containerd/metadata"
|
|
"github.com/containerd/containerd/plugin"
|
|
"github.com/containerd/containerd/snapshots"
|
|
metrics "github.com/docker/go-metrics"
|
|
grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
|
|
"github.com/pkg/errors"
|
|
"golang.org/x/net/context"
|
|
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/health/grpc_health_v1"
|
|
)
|
|
|
|
// New creates and initializes a new containerd server
|
|
func New(ctx context.Context, config *Config) (*Server, error) {
|
|
switch {
|
|
case config.Root == "":
|
|
return nil, errors.New("root must be specified")
|
|
case config.State == "":
|
|
return nil, errors.New("state must be specified")
|
|
case config.Root == config.State:
|
|
return nil, errors.New("root and state must be different paths")
|
|
}
|
|
|
|
if err := os.MkdirAll(config.Root, 0711); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := os.MkdirAll(config.State, 0711); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := apply(ctx, config); err != nil {
|
|
return nil, err
|
|
}
|
|
plugins, err := loadPlugins(config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rpc := grpc.NewServer(
|
|
grpc.UnaryInterceptor(interceptor),
|
|
grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor),
|
|
)
|
|
var (
|
|
services []plugin.Service
|
|
s = &Server{
|
|
rpc: rpc,
|
|
events: exchange.NewExchange(),
|
|
}
|
|
initialized = plugin.NewPluginSet()
|
|
)
|
|
for _, p := range plugins {
|
|
id := p.URI()
|
|
log.G(ctx).WithField("type", p.Type).Infof("loading plugin %q...", id)
|
|
|
|
initContext := plugin.NewContext(
|
|
ctx,
|
|
p,
|
|
initialized,
|
|
config.Root,
|
|
config.State,
|
|
)
|
|
initContext.Events = s.events
|
|
initContext.Address = config.GRPC.Address
|
|
|
|
// load the plugin specific configuration if it is provided
|
|
if p.Config != nil {
|
|
pluginConfig, err := config.Decode(p.ID, p.Config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
initContext.Config = pluginConfig
|
|
}
|
|
result := p.Init(initContext)
|
|
if err := initialized.Add(result); err != nil {
|
|
return nil, errors.Wrapf(err, "could not add plugin result to plugin set")
|
|
}
|
|
|
|
instance, err := result.Instance()
|
|
if err != nil {
|
|
if plugin.IsSkipPlugin(err) {
|
|
log.G(ctx).WithField("type", p.Type).Infof("skip loading plugin %q...", id)
|
|
} else {
|
|
log.G(ctx).WithError(err).Warnf("failed to load plugin %s", id)
|
|
}
|
|
continue
|
|
}
|
|
// check for grpc services that should be registered with the server
|
|
if service, ok := instance.(plugin.Service); ok {
|
|
services = append(services, service)
|
|
}
|
|
}
|
|
// register services after all plugins have been initialized
|
|
for _, service := range services {
|
|
if err := service.Register(rpc); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
// Server is the containerd main daemon
|
|
type Server struct {
|
|
rpc *grpc.Server
|
|
events *exchange.Exchange
|
|
}
|
|
|
|
// ServeGRPC provides the containerd grpc APIs on the provided listener
|
|
func (s *Server) ServeGRPC(l net.Listener) error {
|
|
// before we start serving the grpc API regster the grpc_prometheus metrics
|
|
// handler. This needs to be the last service registered so that it can collect
|
|
// metrics for every other service
|
|
grpc_prometheus.Register(s.rpc)
|
|
return trapClosedConnErr(s.rpc.Serve(l))
|
|
}
|
|
|
|
// ServeMetrics provides a prometheus endpoint for exposing metrics
|
|
func (s *Server) ServeMetrics(l net.Listener) error {
|
|
m := http.NewServeMux()
|
|
m.Handle("/v1/metrics", metrics.Handler())
|
|
return trapClosedConnErr(http.Serve(l, m))
|
|
}
|
|
|
|
// ServeDebug provides a debug endpoint
|
|
func (s *Server) ServeDebug(l net.Listener) error {
|
|
// don't use the default http server mux to make sure nothing gets registered
|
|
// that we don't want to expose via containerd
|
|
m := http.NewServeMux()
|
|
m.Handle("/debug/vars", expvar.Handler())
|
|
m.Handle("/debug/pprof/", http.HandlerFunc(pprof.Index))
|
|
m.Handle("/debug/pprof/cmdline", http.HandlerFunc(pprof.Cmdline))
|
|
m.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
|
|
m.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
|
|
m.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
|
|
return trapClosedConnErr(http.Serve(l, m))
|
|
}
|
|
|
|
// Stop the containerd server canceling any open connections
|
|
func (s *Server) Stop() {
|
|
s.rpc.Stop()
|
|
}
|
|
|
|
func loadPlugins(config *Config) ([]*plugin.Registration, error) {
|
|
// load all plugins into containerd
|
|
if err := plugin.Load(filepath.Join(config.Root, "plugins")); err != nil {
|
|
return nil, err
|
|
}
|
|
// load additional plugins that don't automatically register themselves
|
|
plugin.Register(&plugin.Registration{
|
|
Type: plugin.ContentPlugin,
|
|
ID: "content",
|
|
InitFn: func(ic *plugin.InitContext) (interface{}, error) {
|
|
ic.Meta.Exports["root"] = ic.Root
|
|
return local.NewStore(ic.Root)
|
|
},
|
|
})
|
|
plugin.Register(&plugin.Registration{
|
|
Type: plugin.MetadataPlugin,
|
|
ID: "bolt",
|
|
Requires: []plugin.Type{
|
|
plugin.ContentPlugin,
|
|
plugin.SnapshotPlugin,
|
|
},
|
|
InitFn: func(ic *plugin.InitContext) (interface{}, error) {
|
|
if err := os.MkdirAll(ic.Root, 0711); err != nil {
|
|
return nil, err
|
|
}
|
|
cs, err := ic.Get(plugin.ContentPlugin)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
snapshottersRaw, err := ic.GetByType(plugin.SnapshotPlugin)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
snapshotters := make(map[string]snapshots.Snapshotter)
|
|
for name, sn := range snapshottersRaw {
|
|
sn, err := sn.Instance()
|
|
if err != nil {
|
|
log.G(ic.Context).WithError(err).
|
|
Warnf("could not use snapshotter %v in metadata plugin", name)
|
|
continue
|
|
}
|
|
snapshotters[name] = sn.(snapshots.Snapshotter)
|
|
}
|
|
|
|
path := filepath.Join(ic.Root, "meta.db")
|
|
ic.Meta.Exports["path"] = path
|
|
|
|
db, err := bolt.Open(path, 0644, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
mdb := metadata.NewDB(db, cs.(content.Store), snapshotters)
|
|
if err := mdb.Init(ic.Context); err != nil {
|
|
return nil, err
|
|
}
|
|
return mdb, nil
|
|
},
|
|
})
|
|
|
|
// return the ordered graph for plugins
|
|
return plugin.Graph(), nil
|
|
}
|
|
|
|
func interceptor(
|
|
ctx context.Context,
|
|
req interface{},
|
|
info *grpc.UnaryServerInfo,
|
|
handler grpc.UnaryHandler,
|
|
) (interface{}, error) {
|
|
ctx = log.WithModule(ctx, "containerd")
|
|
switch info.Server.(type) {
|
|
case tasks.TasksServer:
|
|
ctx = log.WithModule(ctx, "tasks")
|
|
case containers.ContainersServer:
|
|
ctx = log.WithModule(ctx, "containers")
|
|
case contentapi.ContentServer:
|
|
ctx = log.WithModule(ctx, "content")
|
|
case images.ImagesServer:
|
|
ctx = log.WithModule(ctx, "images")
|
|
case grpc_health_v1.HealthServer:
|
|
// No need to change the context
|
|
case version.VersionServer:
|
|
ctx = log.WithModule(ctx, "version")
|
|
case snapshotsapi.SnapshotsServer:
|
|
ctx = log.WithModule(ctx, "snapshot")
|
|
case diff.DiffServer:
|
|
ctx = log.WithModule(ctx, "diff")
|
|
case namespaces.NamespacesServer:
|
|
ctx = log.WithModule(ctx, "namespaces")
|
|
case eventsapi.EventsServer:
|
|
ctx = log.WithModule(ctx, "events")
|
|
case introspection.IntrospectionServer:
|
|
ctx = log.WithModule(ctx, "introspection")
|
|
case leasesapi.LeasesServer:
|
|
ctx = log.WithModule(ctx, "leases")
|
|
default:
|
|
log.G(ctx).Warnf("unknown GRPC server type: %#v\n", info.Server)
|
|
}
|
|
return grpc_prometheus.UnaryServerInterceptor(ctx, req, info, handler)
|
|
}
|
|
|
|
func trapClosedConnErr(err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
if strings.Contains(err.Error(), "use of closed network connection") {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|