Move plugin context events into separate plugin

Signed-off-by: Derek McGowan <derek@mcg.dev>

cmd/containerd/builtins.go

@@ -19,6 +19,7 @@ package main
 // register containerd builtins here
 import (
 	_ "github.com/containerd/containerd/diff/walking/plugin"
+	_ "github.com/containerd/containerd/events/plugin"
 	_ "github.com/containerd/containerd/gc/scheduler"
 	_ "github.com/containerd/containerd/runtime/restart/monitor"
 	_ "github.com/containerd/containerd/services/containers"

events/plugin/plugin.go

@@ -0,0 +1,32 @@
+/*
+   Copyright The containerd 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 plugin
+
+import (
+	"github.com/containerd/containerd/plugin"
+)
+
+func init() {
+	plugin.Register(&plugin.Registration{
+		Type: plugin.EventPlugin,
+		ID:   "exchange",
+		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
+			// TODO: In 2.0, create exchange since ic.Events will be removed
+			return ic.Events, nil
+		},
+	})
+}

metrics/cgroups/cgroups.go

@@ -20,6 +20,7 @@ package cgroups
 
 import (
 	"github.com/containerd/cgroups"
+	"github.com/containerd/containerd/events"
 	v1 "github.com/containerd/containerd/metrics/cgroups/v1"
 	v2 "github.com/containerd/containerd/metrics/cgroups/v2"
 	"github.com/containerd/containerd/platforms"
@@ -38,6 +39,9 @@ func init() {
 		Type:   plugin.TaskMonitorPlugin,
 		ID:     "cgroups",
 		InitFn: New,
+		Requires: []plugin.Type{
+			plugin.EventPlugin,
+		},
 		Config: &Config{},
 	})
 }
@@ -53,10 +57,16 @@ func New(ic *plugin.InitContext) (interface{}, error) {
 		tm  runtime.TaskMonitor
 		err error
 	)
+
+	ep, err := ic.Get(plugin.EventPlugin)
+	if err != nil {
+		return nil, err
+	}
+
 	if cgroups.Mode() == cgroups.Unified {
-		tm, err = v2.NewTaskMonitor(ic.Context, ic.Events, ns)
+		tm, err = v2.NewTaskMonitor(ic.Context, ep.(events.Publisher), ns)
 	} else {
-		tm, err = v1.NewTaskMonitor(ic.Context, ic.Events, ns)
+		tm, err = v1.NewTaskMonitor(ic.Context, ep.(events.Publisher), ns)
 	}
 	if err != nil {
 		return nil, err

pkg/cri/cri.go

@@ -53,6 +53,7 @@ func init() {
 		ID:     "cri",
 		Config: &config,
 		Requires: []plugin.Type{
+			plugin.EventPlugin,
 			plugin.ServicePlugin,
 		},
 		InitFn: initCRIService,
@@ -118,8 +119,13 @@ func getServicesOpts(ic *plugin.InitContext) ([]containerd.ServicesOpt, error) {
 		return nil, errors.Wrap(err, "failed to get service plugin")
 	}
 
+	ep, err := ic.Get(plugin.EventPlugin)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get event plugin")
+	}
+
 	opts := []containerd.ServicesOpt{
-		containerd.WithEventService(ic.Events),
+		containerd.WithEventService(ep.(containerd.EventService)),
 	}
 	for s, fn := range map[string]func(interface{}) containerd.ServicesOpt{
 		services.ContentService: func(s interface{}) containerd.ServicesOpt {

plugin/context.go

@@ -34,7 +34,9 @@ type InitContext struct {
 	Config       interface{}
 	Address      string
 	TTRPCAddress string
-	Events       *exchange.Exchange
+
+	// deprecated: will be removed in 2.0, use plugin.EventType
+	Events *exchange.Exchange
 
 	Meta *Meta // plugins can fill in metadata at init.
 
@@ -135,6 +137,19 @@ func (i *InitContext) GetAll() []*Plugin {
 	return i.plugins.ordered
 }
 
+// GetByID returns the plugin of the given type and ID
+func (i *InitContext) GetByID(t Type, id string) (interface{}, error) {
+	ps, err := i.GetByType(t)
+	if err != nil {
+		return nil, err
+	}
+	p, ok := ps[id]
+	if !ok {
+		return nil, errors.Wrapf(errdefs.ErrNotFound, "no %s plugins with id %s", t, id)
+	}
+	return p.Instance()
+}
+
 // GetByType returns all plugins with the specific type.
 func (i *InitContext) GetByType(t Type) (map[string]*Plugin, error) {
 	p, ok := i.plugins.byTypeAndID[t]

plugin/plugin.go

@@ -75,6 +75,8 @@ const (
 	ContentPlugin Type = "io.containerd.content.v1"
 	// GCPlugin implements garbage collection policy
 	GCPlugin Type = "io.containerd.gc.v1"
+	// EventPlugin implements event handling
+	EventPlugin Type = "io.containerd.event.v1"
 )
 
 const (

runtime/restart/monitor/monitor.go

@@ -63,6 +63,7 @@ func init() {
 	plugin.Register(&plugin.Registration{
 		Type: plugin.InternalPlugin,
 		Requires: []plugin.Type{
+			plugin.EventPlugin,
 			plugin.ServicePlugin,
 		},
 		ID: "restart",
@@ -95,8 +96,14 @@ func getServicesOpts(ic *plugin.InitContext) ([]containerd.ServicesOpt, error) {
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get service plugin")
 	}
+
+	ep, err := ic.Get(plugin.EventPlugin)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get event plugin")
+	}
+
 	opts := []containerd.ServicesOpt{
-		containerd.WithEventService(ic.Events),
+		containerd.WithEventService(ep.(containerd.EventService)),
 	}
 	for s, fn := range map[string]func(interface{}) containerd.ServicesOpt{
 		services.ContentService: func(s interface{}) containerd.ServicesOpt {

runtime/v1/linux/runtime.go

@@ -73,6 +73,7 @@ func init() {
 		ID:     "linux",
 		InitFn: New,
 		Requires: []plugin.Type{
+			plugin.EventPlugin,
 			plugin.MetadataPlugin,
 		},
 		Config: &Config{
@@ -112,6 +113,12 @@ func New(ic *plugin.InitContext) (interface{}, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	ep, err := ic.GetByID(plugin.EventPlugin, "exchange")
+	if err != nil {
+		return nil, err
+	}
+
 	cfg := ic.Config.(*Config)
 	r := &Runtime{
 		root:       ic.Root,
@@ -119,7 +126,7 @@ func New(ic *plugin.InitContext) (interface{}, error) {
 		tasks:      runtime.NewTaskList(),
 		containers: metadata.NewContainerStore(m.(*metadata.DB)),
 		address:    ic.Address,
-		events:     ic.Events,
+		events:     ep.(*exchange.Exchange),
 		config:     cfg,
 	}
 	tasks, err := r.restoreTasks(ic.Context)

runtime/v2/manager.go

@@ -49,6 +49,7 @@ func init() {
 		Type: plugin.RuntimePluginV2,
 		ID:   "task",
 		Requires: []plugin.Type{
+			plugin.EventPlugin,
 			plugin.MetadataPlugin,
 		},
 		Config: &Config{
@@ -71,9 +72,14 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
+			ep, err := ic.GetByID(plugin.EventPlugin, "exchange")
+			if err != nil {
+				return nil, err
+			}
 			cs := metadata.NewContainerStore(m.(*metadata.DB))
+			events := ep.(*exchange.Exchange)
 
-			return New(ic.Context, ic.Root, ic.State, ic.Address, ic.TTRPCAddress, ic.Events, cs)
+			return New(ic.Context, ic.Root, ic.State, ic.Address, ic.TTRPCAddress, events, cs)
 		},
 	})
 }

services/containers/local.go

@@ -41,6 +41,7 @@ func init() {
 		Type: plugin.ServicePlugin,
 		ID:   services.ContainersService,
 		Requires: []plugin.Type{
+			plugin.EventPlugin,
 			plugin.MetadataPlugin,
 		},
 		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
@@ -48,12 +49,16 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
+			ep, err := ic.Get(plugin.EventPlugin)
+			if err != nil {
+				return nil, err
+			}
 
 			db := m.(*metadata.DB)
 			return &local{
 				Store:     metadata.NewContainerStore(db),
 				db:        db,
-				publisher: ic.Events,
+				publisher: ep.(events.Publisher),
 			}, nil
 		},
 	})

services/content/store.go

@@ -39,6 +39,7 @@ func init() {
 		Type: plugin.ServicePlugin,
 		ID:   services.ContentService,
 		Requires: []plugin.Type{
+			plugin.EventPlugin,
 			plugin.MetadataPlugin,
 		},
 		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
@@ -46,8 +47,12 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
+			ep, err := ic.Get(plugin.EventPlugin)
+			if err != nil {
+				return nil, err
+			}
 
-			s, err := newContentStore(m.(*metadata.DB).ContentStore(), ic.Events)
+			s, err := newContentStore(m.(*metadata.DB).ContentStore(), ep.(events.Publisher))
 			return s, err
 		},
 	})

services/events/service.go

@@ -35,8 +35,15 @@ func init() {
 	plugin.Register(&plugin.Registration{
 		Type: plugin.GRPCPlugin,
 		ID:   "events",
+		Requires: []plugin.Type{
+			plugin.EventPlugin,
+		},
 		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
-			return NewService(ic.Events), nil
+			ep, err := ic.GetByID(plugin.EventPlugin, "exchange")
+			if err != nil {
+				return nil, err
+			}
+			return NewService(ep.(*exchange.Exchange)), nil
 		},
 	})
 }

services/namespaces/local.go

@@ -40,6 +40,7 @@ func init() {
 		Type: plugin.ServicePlugin,
 		ID:   services.NamespacesService,
 		Requires: []plugin.Type{
+			plugin.EventPlugin,
 			plugin.MetadataPlugin,
 		},
 		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
@@ -47,9 +48,13 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
+			ep, err := ic.Get(plugin.EventPlugin)
+			if err != nil {
+				return nil, err
+			}
 			return &local{
 				db:        m.(*metadata.DB),
-				publisher: ic.Events,
+				publisher: ep.(events.Publisher),
 			}, nil
 		},
 	})

services/server/server.go

@@ -146,9 +146,10 @@ func New(ctx context.Context, config *srvconfig.Config) (*Server, error) {
 			grpcServer:  grpcServer,
 			tcpServer:   tcpServer,
 			ttrpcServer: ttrpcServer,
-			events:      exchange.NewExchange(),
 			config:      config,
 		}
+		// TODO: Remove this in 2.0 and let event plugin crease it
+		events      = exchange.NewExchange()
 		initialized = plugin.NewPluginSet()
 		required    = make(map[string]struct{})
 	)
@@ -170,7 +171,7 @@ func New(ctx context.Context, config *srvconfig.Config) (*Server, error) {
 			config.Root,
 			config.State,
 		)
-		initContext.Events = s.events
+		initContext.Events = events
 		initContext.Address = config.GRPC.Address
 		initContext.TTRPCAddress = config.TTRPC.Address
 
@@ -246,7 +247,6 @@ type Server struct {
 	grpcServer  *grpc.Server
 	ttrpcServer *ttrpc.Server
 	tcpServer   *grpc.Server
-	events      *exchange.Exchange
 	config      *srvconfig.Config
 	plugins     []*plugin.Plugin
 }

services/snapshots/snapshotters.go

@@ -39,6 +39,7 @@ func init() {
 		Type: plugin.ServicePlugin,
 		ID:   services.SnapshotsService,
 		Requires: []plugin.Type{
+			plugin.EventPlugin,
 			plugin.MetadataPlugin,
 		},
 		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
@@ -46,11 +47,15 @@ func init() {
 			if err != nil {
 				return nil, err
 			}
+			ep, err := ic.Get(plugin.EventPlugin)
+			if err != nil {
+				return nil, err
+			}
 
 			db := m.(*metadata.DB)
 			ss := make(map[string]snapshots.Snapshotter)
 			for n, sn := range db.Snapshotters() {
-				ss[n] = newSnapshotter(sn, ic.Events)
+				ss[n] = newSnapshotter(sn, ep.(events.Publisher))
 			}
 			return ss, nil
 		},

services/tasks/local.go

@@ -92,6 +92,11 @@ func initFunc(ic *plugin.InitContext) (interface{}, error) {
 		return nil, err
 	}
 
+	ep, err := ic.Get(plugin.EventPlugin)
+	if err != nil {
+		return nil, err
+	}
+
 	monitor, err := ic.Get(plugin.TaskMonitorPlugin)
 	if err != nil {
 		if !errdefs.IsNotFound(err) {
@@ -105,7 +110,7 @@ func initFunc(ic *plugin.InitContext) (interface{}, error) {
 		runtimes:   runtimes,
 		containers: metadata.NewContainerStore(db),
 		store:      db.ContentStore(),
-		publisher:  ic.Events,
+		publisher:  ep.(events.Publisher),
 		monitor:    monitor.(runtime.TaskMonitor),
 		v2Runtime:  v2r.(*v2.TaskManager),
 	}

services/tasks/local_freebsd.go

@@ -24,6 +24,7 @@ import (
 )
 
 var tasksServiceRequires = []plugin.Type{
+	plugin.EventPlugin,
 	plugin.RuntimePluginV2,
 	plugin.MetadataPlugin,
 	plugin.TaskMonitorPlugin,

services/tasks/local_unix.go

@@ -26,6 +26,7 @@ import (
 )
 
 var tasksServiceRequires = []plugin.Type{
+	plugin.EventPlugin,
 	plugin.RuntimePlugin,
 	plugin.RuntimePluginV2,
 	plugin.MetadataPlugin,

services/tasks/local_windows.go

@@ -24,6 +24,7 @@ import (
 )
 
 var tasksServiceRequires = []plugin.Type{
+	plugin.EventPlugin,
 	plugin.RuntimePluginV2,
 	plugin.MetadataPlugin,
 	plugin.TaskMonitorPlugin,