Refactor Reacter plugin (#509)

Author: mthenw

cmd/event-gateway/main.go

@@ -97,8 +97,7 @@ func main() {
 	}
 
 	// Plugin manager
-	pluginManager := plugin.NewManager(plugins, log)
-	err = pluginManager.Connect()
+	pluginManager, err := plugin.NewManager(plugins, log)
 	if err != nil {
 		log.Fatal("Loading plugins failed.", zap.Error(err))
 	}

docs/system-events-and-plugin-system.md

@@ -47,4 +47,4 @@ further processed by the Event Gateway.
 
 `React` method is called for every system event that plugin subscribed to.
 
-For more details, see [the example plugin](../plugin/example).
+For more details, see [the example plugin](../plugin/example/reacter).

plugin/example/main.go

@@ -0,0 +1,18 @@
+package main
+
+import (
+	goplugin "github.com/hashicorp/go-plugin"
+	"github.com/serverless/event-gateway/plugin"
+	"github.com/serverless/event-gateway/plugin/shared"
+)
+
+func main() {
+	pluginMap := map[string]goplugin.Plugin{
+		"reacter": &plugin.ReacterRPCPlugin{Reacter: &Simple{}},
+	}
+
+	goplugin.Serve(&goplugin.ServeConfig{
+		HandshakeConfig: shared.Handshake,
+		Plugins:         pluginMap,
+	})
+}

plugin/example/reacter/main.go

@@ -32,7 +32,7 @@ func (s *Simple) React(instance event.Event) error {
 	switch instance.EventType {
 	case event.SystemEventReceivedType:
 		received := instance.Data.(event.SystemEventReceivedData)
-		log.Printf("received gateway.received.event for event: %q", received.Event.EventType)
+		log.Printf("received %s for event: %q", event.SystemEventReceivedType, received.Event.EventType)
 	}
 
 	return nil

plugin/example/plugin.go renamed to plugin/example/reacter/plugin.go

@@ -8,6 +8,7 @@ import (
 
 	goplugin "github.com/hashicorp/go-plugin"
 	"github.com/serverless/event-gateway/event"
+	"github.com/serverless/event-gateway/plugin/shared"
 )
 
 func init() {
@@ -18,84 +19,85 @@ func init() {
 	gob.Register(event.SystemFunctionInvocationFailedData{})
 }
 
+// Type of a subscription.
+type Type int
+
+const (
+	// Async subscription type. Plugin host will not block on it.
+	Async Type = iota
+	// Sync subscription type. Plugin host will use the response from the plugin before proceeding.
+	Sync
+)
+
 // Plugin is a generic struct for storing info about a plugin.
 type Plugin struct {
-	Path          string
-	Client        *goplugin.Client
-	Reacter       Reacter
-	Subscriptions []Subscription
+	Path    string
+	Reacter Reacter
 }
 
 // Manager handles lifecycle of plugin management.
 type Manager struct {
-	Plugins []*Plugin
-	Log     *zap.Logger
+	Reacters []*Plugin
+	Log      *zap.Logger
 }
 
 // NewManager creates new Manager.
-func NewManager(paths []string, log *zap.Logger) *Manager {
-	plugins := []*Plugin{}
+func NewManager(paths []string, log *zap.Logger) (*Manager, error) {
+	reacters := []*Plugin{}
 	logger := Hclog2ZapLogger{Zap: log}
+
 	for _, path := range paths {
 		client := goplugin.NewClient(&goplugin.ClientConfig{
-			HandshakeConfig: handshakeConfig,
+			HandshakeConfig: shared.Handshake,
 			Plugins:         pluginMap,
 			Cmd:             exec.Command(path),
 			Logger:          logger.Named("PluginManager"),
+			Managed:         true,
 		})
 
-		plugins = append(plugins, &Plugin{
-			Client: client,
-			Path:   path,
-		})
-	}
-
-	return &Manager{
-		Plugins: plugins,
-		Log:     log,
-	}
-}
-
-// Connect connects to plugins.
-func (m *Manager) Connect() error {
-	for _, plugin := range m.Plugins {
-		rpcClient, err := plugin.Client.Client()
+		rpcClient, err := client.Client()
 		if err != nil {
-			return err
+			return nil, err
 		}
 
 		// Request the plugin
-		raw, err := rpcClient.Dispense("subscriber")
+		raw, err := rpcClient.Dispense("reacter")
 		if err != nil {
-			return err
+			return nil, err
 		}
 
-		plugin.Reacter = raw.(*Subscriber)
-		plugin.Subscriptions = plugin.Reacter.Subscriptions()
+		switch raw.(type) {
+		case *ReacterClient:
+			reacters = append(reacters, &Plugin{
+				Path:    path,
+				Reacter: raw.(*ReacterClient),
+			})
+		}
 	}
 
-	return nil
+	return &Manager{
+		Reacters: reacters,
+		Log:      log,
+	}, nil
 }
 
 // Kill disconnects plugins and kill subprocesses.
 func (m *Manager) Kill() {
-	for _, plugin := range m.Plugins {
-		plugin.Client.Kill()
-	}
+	goplugin.CleanupClients()
 }
 
 // React call all plugins' React method. It returns when the first error is returned by a plugin.
 func (m *Manager) React(event *event.Event) error {
-	for _, plugin := range m.Plugins {
-		for _, subscription := range plugin.Subscriptions {
-			if subscription.EventType == event.EventType {
+	for _, plugin := range m.Reacters {
+		for _, sub := range plugin.Reacter.Subscriptions() {
+			if sub.EventType == event.EventType {
 				err := plugin.Reacter.React(*event)
 				if err != nil {
 					m.Log.Debug("Plugin returned error.",
 						zap.String("plugin", plugin.Path),
 						zap.Error(err),
-						zap.String("subscriptionType", string(subscription.Type)))
-					if subscription.Type == Sync {
+						zap.String("subscriptionType", string(sub.Type)))
+					if sub.Type == Sync {
 						return err
 					}
 				}
@@ -106,19 +108,7 @@ func (m *Manager) React(event *event.Event) error {
 	return nil
 }
 
-// handshakeConfig is used to just do a basic handshake between a plugin and host. If the handshake fails, a user
-// friendly error is shown. This prevents users from executing bad plugins or executing a plugin directory. It is a UX
-// feature, not a security feature.
-var handshakeConfig = goplugin.HandshakeConfig{
-	// The ProtocolVersion is the version that must match between EG core and EG plugins. This should be bumped whenever
-	// a change happens in one or the other that makes it so that they can't safely communicate.
-	ProtocolVersion: 1,
-	// The magic cookie values should NEVER be changed.
-	MagicCookieKey:   "EVENT_GATEWAY_MAGIC_COOKIE",
-	MagicCookieValue: "0329c93c-a64c-4eb5-bf72-63172430d433",
-}
-
 // pluginMap is the map of plugins we can dispense.
 var pluginMap = map[string]goplugin.Plugin{
-	"subscriber": &SubscriberPlugin{},
+	"reacter": &ReacterRPCPlugin{},
 }

plugin/manager.go

@@ -0,0 +1,99 @@
+package plugin
+
+import (
+	"net/rpc"
+
+	goplugin "github.com/hashicorp/go-plugin"
+	eventpkg "github.com/serverless/event-gateway/event"
+)
+
+// Subscription use by plugin to indicate which event it wants to react to.
+type Subscription struct {
+	EventType eventpkg.TypeName
+	Type
+}
+
+// Reacter allows reacting on subscribed event types.
+type Reacter interface {
+	Subscriptions() []Subscription
+	React(event eventpkg.Event) error
+}
+
+// ReacterRPCPlugin is the go-plugin's Plugin implementation.
+type ReacterRPCPlugin struct {
+	Reacter Reacter
+}
+
+// Server hosts ReacterServer.
+func (r *ReacterRPCPlugin) Server(*goplugin.MuxBroker) (interface{}, error) {
+	return &ReacterServer{Reacter: r.Reacter}, nil
+}
+
+// Client provides ReacterClient client.
+func (r *ReacterRPCPlugin) Client(b *goplugin.MuxBroker, c *rpc.Client) (interface{}, error) {
+	return &ReacterClient{client: c}, nil
+}
+
+// ReacterServer is a net/rpc compatibile structure for serving a Reacter.
+type ReacterServer struct {
+	Reacter Reacter
+}
+
+// Subscriptions server implementation.
+func (r *ReacterServer) Subscriptions(_ interface{}, resp *ReacterSubscriptionsResponse) error {
+	*resp = ReacterSubscriptionsResponse{Subscriptions: r.Reacter.Subscriptions()}
+	return nil
+}
+
+// React server implementation.
+func (r *ReacterServer) React(args *ReacterReactArgs, resp *ReacterReactResponse) error {
+	err := r.Reacter.React(args.Event)
+
+	*resp = ReacterReactResponse{Error: goplugin.NewBasicError(err)}
+	return nil
+}
+
+// ReacterClient is a RPC implementation of Reacter.
+type ReacterClient struct {
+	client *rpc.Client
+}
+
+// Subscriptions call plugin implementation.
+func (r *ReacterClient) Subscriptions() []Subscription {
+	var resp ReacterSubscriptionsResponse
+	err := r.client.Call("Plugin.Subscriptions", new(interface{}), &resp)
+	if err != nil {
+		return []Subscription{}
+	}
+
+	return resp.Subscriptions
+}
+
+// React calls plugin implementation.
+func (r *ReacterClient) React(event eventpkg.Event) error {
+	args := &ReacterReactArgs{Event: event}
+	var resp ReacterReactResponse
+	err := r.client.Call("Plugin.React", args, &resp)
+	if err != nil {
+		return err
+	}
+	if resp.Error != nil {
+		err = resp.Error
+	}
+	return err
+}
+
+// ReacterSubscriptionsResponse RPC response
+type ReacterSubscriptionsResponse struct {
+	Subscriptions []Subscription
+}
+
+// ReacterReactArgs RPC args
+type ReacterReactArgs struct {
+	Event eventpkg.Event
+}
+
+// ReacterReactResponse RPC response
+type ReacterReactResponse struct {
+	Error *goplugin.BasicError
+}

plugin/plugin.go

@@ -0,0 +1,14 @@
+// Package shared contains shared data between the host and plugins.
+package shared
+
+import "github.com/hashicorp/go-plugin"
+
+// Handshake is a common handshake that is shared by plugin and host.
+var Handshake = plugin.HandshakeConfig{
+	// The ProtocolVersion is the version that must match between EG core and EG plugins. This should be bumped whenever
+	// a change happens in one or the other that makes it so that they can't safely communicate.
+	ProtocolVersion: 1,
+	// The magic cookie values should NEVER be changed.
+	MagicCookieKey:   "EVENT_GATEWAY_MAGIC_COOKIE",
+	MagicCookieValue: "0329c93c-a64c-4eb5-bf72-63172430d433",
+}