Merge pull request #5835 from dmcgowan/plugin-events-cleanup

Move plugin context events into separate plugin

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

@@ -154,9 +154,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{})
 	)
@@ -178,7 +179,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
 
@@ -254,7 +255,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

@@ -22,6 +22,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

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