[MM-35077] Add basic support for plugin intra-cluster communication (#17495)

* Add basic support for plugin intra-cluster communication

* Some renaming for added clarity

* Allow sending cluster event to specific nodes

* Improve naming and documentation

* Improve logging

app/busy_test.go

@@ -120,6 +120,10 @@ func (c *ClusterMock) SendClusterMessage(msg *model.ClusterMessage) {
 	c.Busy.ClusterEventChanged(sbs)
 }
 
+func (c *ClusterMock) SendClusterMessageToNode(nodeID string, msg *model.ClusterMessage) error {
+	return nil
+}
+
 func (c *ClusterMock) StartInterNodeCommunication() {}
 func (c *ClusterMock) StopInterNodeCommunication()  {}
 func (c *ClusterMock) RegisterClusterMessageHandler(event string, crm einterfaces.ClusterMessageHandler) {

app/cluster_handlers.go

@@ -20,6 +20,7 @@ func (a *App) registerAppClusterMessageHandlers() {
 	a.Cluster().RegisterClusterMessageHandler(model.CLUSTER_EVENT_CLEAR_SESSION_CACHE_FOR_ALL_USERS, a.clusterClearSessionCacheForAllUsersHandler)
 	a.Cluster().RegisterClusterMessageHandler(model.CLUSTER_EVENT_INSTALL_PLUGIN, a.clusterInstallPluginHandler)
 	a.Cluster().RegisterClusterMessageHandler(model.CLUSTER_EVENT_REMOVE_PLUGIN, a.clusterRemovePluginHandler)
+	a.Cluster().RegisterClusterMessageHandler(model.CLUSTER_EVENT_PLUGIN_EVENT, a.clusterPluginEventHandler)
 }
 
 func (a *App) clusterClearSessionCacheForUserHandler(msg *model.ClusterMessage) {
@@ -34,6 +35,35 @@ func (a *App) clusterInstallPluginHandler(msg *model.ClusterMessage) {
 	a.InstallPluginFromData(model.PluginEventDataFromJson(strings.NewReader(msg.Data)))
 }
 
+func (a *App) clusterPluginEventHandler(msg *model.ClusterMessage) {
+	env := a.GetPluginsEnvironment()
+	if env == nil {
+		return
+	}
+	if msg.Props == nil {
+		mlog.Warn("ClusterMessage.Props for plugin event should not be nil")
+		return
+	}
+	pluginID := msg.Props["PluginID"]
+	eventID := msg.Props["EventID"]
+	if pluginID == "" || eventID == "" {
+		mlog.Warn("Invalid ClusterMessage.Props values for plugin event",
+			mlog.String("plugin_id", pluginID), mlog.String("event_id", eventID))
+		return
+	}
+
+	hooks, err := env.HooksForPlugin(pluginID)
+	if err != nil {
+		mlog.Warn("Getting hooks for plugin failed", mlog.String("plugin_id", pluginID), mlog.Err(err))
+		return
+	}
+
+	hooks.OnPluginClusterEvent(a.PluginContext(), model.PluginClusterEvent{
+		Id:   eventID,
+		Data: []byte(msg.Data),
+	})
+}
+
 func (a *App) clusterRemovePluginHandler(msg *model.ClusterMessage) {
 	a.RemovePluginFromData(model.PluginEventDataFromJson(strings.NewReader(msg.Data)))
 }

app/plugin_api.go

@@ -1063,3 +1063,34 @@ func (api *PluginAPI) DeleteCommand(commandID string) error {
 
 	return nil
 }
+
+// PublishPluginClusterEvent broadcasts a plugin event to all other running instances of
+// the calling plugin.
+func (api *PluginAPI) PublishPluginClusterEvent(ev model.PluginClusterEvent,
+	opts model.PluginClusterEventSendOptions) error {
+	if api.app.Cluster() == nil {
+		return nil
+	}
+
+	msg := &model.ClusterMessage{
+		Event:            model.CLUSTER_EVENT_PLUGIN_EVENT,
+		SendType:         opts.SendType,
+		WaitForAllToSend: false,
+		Props: map[string]string{
+			"PluginID": api.id,
+			"EventID":  ev.Id,
+		},
+		Data: string(ev.Data),
+	}
+
+	// If TargetId is empty we broadcast to all other cluster nodes.
+	if opts.TargetId == "" {
+		api.app.Cluster().SendClusterMessage(msg)
+	} else {
+		if err := api.app.Cluster().SendClusterMessageToNode(opts.TargetId, msg); err != nil {
+			return fmt.Errorf("failed to send message to cluster node %q: %w", opts.TargetId, err)
+		}
+	}
+
+	return nil
+}

einterfaces/cluster.go

@@ -21,7 +21,8 @@ type ClusterInterface interface {
 	HealthScore() int
 	GetMyClusterInfo() *model.ClusterInfo
 	GetClusterInfos() []*model.ClusterInfo
-	SendClusterMessage(cluster *model.ClusterMessage)
+	SendClusterMessage(msg *model.ClusterMessage)
+	SendClusterMessageToNode(nodeID string, msg *model.ClusterMessage) error
 	NotifyMsg(buf []byte)
 	GetClusterStats() ([]*model.ClusterStats, *model.AppError)
 	GetLogs(page, perPage int) ([]string, *model.AppError)

einterfaces/mocks/ClusterInterface.go

@@ -191,9 +191,23 @@ func (_m *ClusterInterface) RegisterClusterMessageHandler(event string, crm eint
 	_m.Called(event, crm)
 }
 
-// SendClusterMessage provides a mock function with given fields: cluster
+// SendClusterMessage provides a mock function with given fields: msg
-func (_m *ClusterInterface) SendClusterMessage(cluster *model.ClusterMessage) {
+func (_m *ClusterInterface) SendClusterMessage(msg *model.ClusterMessage) {
-	_m.Called(cluster)
+	_m.Called(msg)
+}
+
+// SendClusterMessageToNode provides a mock function with given fields: nodeID, msg
+func (_m *ClusterInterface) SendClusterMessageToNode(nodeID string, msg *model.ClusterMessage) error {
+	ret := _m.Called(nodeID, msg)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, *model.ClusterMessage) error); ok {
+		r0 = rf(nodeID, msg)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
 }
 
 // StartInterNodeCommunication provides a mock function with given fields:

model/cluster_message.go

@@ -40,6 +40,7 @@ const (
 	CLUSTER_EVENT_CLEAR_SESSION_CACHE_FOR_ALL_USERS                 = "inv_all_user_sessions"
 	CLUSTER_EVENT_INSTALL_PLUGIN                                    = "install_plugin"
 	CLUSTER_EVENT_REMOVE_PLUGIN                                     = "remove_plugin"
+	CLUSTER_EVENT_PLUGIN_EVENT                                      = "plugin_event"
 	CLUSTER_EVENT_INVALIDATE_CACHE_FOR_TERMS_OF_SERVICE             = "inv_terms_of_service"
 	CLUSTER_EVENT_BUSY_STATE_CHANGED                                = "busy_state_change"
 

model/plugin_cluster_event.go

@@ -0,0 +1,28 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package model
+
+const (
+	PluginClusterEventSendTypeReliable   = CLUSTER_SEND_RELIABLE
+	PluginClusterEventSendTypeBestEffort = CLUSTER_SEND_BEST_EFFORT
+)
+
+// PluginClusterEvent is used to allow intra-cluster plugin communication.
+type PluginClusterEvent struct {
+	// Id is the unique identifier for the event.
+	Id string
+	// Data is the event payload.
+	Data []byte
+}
+
+// PluginClusterEventSendOptions defines some properties that apply when sending
+// plugin events across a cluster.
+type PluginClusterEventSendOptions struct {
+	// SendType defines the type of communication channel used to send the event.
+	SendType string
+	// TargetId identifies the cluster node to which the event should be sent.
+	// It should match the cluster id of the receiving instance.
+	// If empty, the event gets broadcasted to all other nodes.
+	TargetId string
+}

plugin/api.go

@@ -1050,6 +1050,16 @@ type API interface {
 	// @tag SlashCommand
 	// Minimum server version: 5.28
 	DeleteCommand(commandID string) error
+
+	// PublishPluginClusterEvent broadcasts a plugin event to all other running instances of
+	// the calling plugin that are present in the cluster.
+	//
+	// This method is used to allow plugin communication in a High-Availability cluster.
+	// The receiving side should implement the OnPluginClusterEvent hook
+	// to receive events sent through this method.
+	//
+	// Minimum server version: 5.36
+	PublishPluginClusterEvent(ev model.PluginClusterEvent, opts model.PluginClusterEventSendOptions) error
 }
 
 var handshake = plugin.HandshakeConfig{

plugin/api_timer_layer_generated.go

@@ -1113,3 +1113,10 @@ func (api *apiTimerLayer) DeleteCommand(commandID string) error {
 	api.recordTime(startTime, "DeleteCommand", _returnsA == nil)
 	return _returnsA
 }
+
+func (api *apiTimerLayer) PublishPluginClusterEvent(ev model.PluginClusterEvent, opts model.PluginClusterEventSendOptions) error {
+	startTime := timePkg.Now()
+	_returnsA := api.apiImpl.PublishPluginClusterEvent(ev, opts)
+	api.recordTime(startTime, "PublishPluginClusterEvent", _returnsA == nil)
+	return _returnsA
+}

plugin/client_rpc_generated.go

@@ -532,6 +532,40 @@ func (s *hooksRPCServer) ReactionHasBeenRemoved(args *Z_ReactionHasBeenRemovedAr
 	return nil
 }
 
+func init() {
+	hookNameToId["OnPluginClusterEvent"] = OnPluginClusterEventID
+}
+
+type Z_OnPluginClusterEventArgs struct {
+	A *Context
+	B model.PluginClusterEvent
+}
+
+type Z_OnPluginClusterEventReturns struct {
+}
+
+func (g *hooksRPCClient) OnPluginClusterEvent(c *Context, ev model.PluginClusterEvent) {
+	_args := &Z_OnPluginClusterEventArgs{c, ev}
+	_returns := &Z_OnPluginClusterEventReturns{}
+	if g.implemented[OnPluginClusterEventID] {
+		if err := g.client.Call("Plugin.OnPluginClusterEvent", _args, _returns); err != nil {
+			g.log.Error("RPC call OnPluginClusterEvent to plugin failed.", mlog.Err(err))
+		}
+	}
+
+}
+
+func (s *hooksRPCServer) OnPluginClusterEvent(args *Z_OnPluginClusterEventArgs, returns *Z_OnPluginClusterEventReturns) error {
+	if hook, ok := s.impl.(interface {
+		OnPluginClusterEvent(c *Context, ev model.PluginClusterEvent)
+	}); ok {
+		hook.OnPluginClusterEvent(args.A, args.B)
+	} else {
+		return encodableError(fmt.Errorf("Hook OnPluginClusterEvent called but not implemented."))
+	}
+	return nil
+}
+
 type Z_RegisterCommandArgs struct {
 	A *model.Command
 }
@@ -4906,3 +4940,33 @@ func (s *apiRPCServer) DeleteCommand(args *Z_DeleteCommandArgs, returns *Z_Delet
 	}
 	return nil
 }
+
+type Z_PublishPluginClusterEventArgs struct {
+	A model.PluginClusterEvent
+	B model.PluginClusterEventSendOptions
+}
+
+type Z_PublishPluginClusterEventReturns struct {
+	A error
+}
+
+func (g *apiRPCClient) PublishPluginClusterEvent(ev model.PluginClusterEvent, opts model.PluginClusterEventSendOptions) error {
+	_args := &Z_PublishPluginClusterEventArgs{ev, opts}
+	_returns := &Z_PublishPluginClusterEventReturns{}
+	if err := g.client.Call("Plugin.PublishPluginClusterEvent", _args, _returns); err != nil {
+		log.Printf("RPC call to PublishPluginClusterEvent API failed: %s", err.Error())
+	}
+	return _returns.A
+}
+
+func (s *apiRPCServer) PublishPluginClusterEvent(args *Z_PublishPluginClusterEventArgs, returns *Z_PublishPluginClusterEventReturns) error {
+	if hook, ok := s.impl.(interface {
+		PublishPluginClusterEvent(ev model.PluginClusterEvent, opts model.PluginClusterEventSendOptions) error
+	}); ok {
+		returns.A = hook.PublishPluginClusterEvent(args.A, args.B)
+		returns.A = encodableError(returns.A)
+	} else {
+		return encodableError(fmt.Errorf("API PublishPluginClusterEvent called but not implemented."))
+	}
+	return nil
+}

plugin/hooks.go

@@ -35,6 +35,7 @@ const (
 	UserHasBeenCreatedID     = 17
 	ReactionHasBeenAddedID   = 18
 	ReactionHasBeenRemovedID = 19
+	OnPluginClusterEventID   = 20
 	TotalHooksID             = iota
 )
 
@@ -209,4 +210,13 @@ type Hooks interface {
 	//
 	// Minimum server version: 5.30
 	ReactionHasBeenRemoved(c *Context, reaction *model.Reaction)
+
+	// OnPluginClusterEvent is invoked when an intra-cluster plugin event is received.
+	//
+	// This is used to allow communication between multiple instances of the same plugin
+	// that are running on separate nodes of the same High-Availability cluster.
+	// This hook receives events sent by a call to PublishPluginClusterEvent.
+	//
+	// Minimum server version: 5.36
+	OnPluginClusterEvent(c *Context, ev model.PluginClusterEvent)
 }

plugin/hooks_timer_layer_generated.go

@@ -162,3 +162,9 @@ func (hooks *hooksTimerLayer) ReactionHasBeenRemoved(c *Context, reaction *model
 	hooks.hooksImpl.ReactionHasBeenRemoved(c, reaction)
 	hooks.recordTime(startTime, "ReactionHasBeenRemoved", true)
 }
+
+func (hooks *hooksTimerLayer) OnPluginClusterEvent(c *Context, ev model.PluginClusterEvent) {
+	startTime := timePkg.Now()
+	hooks.hooksImpl.OnPluginClusterEvent(c, ev)
+	hooks.recordTime(startTime, "OnPluginClusterEvent", true)
+}

plugin/plugintest/api.go

@@ -2682,6 +2682,20 @@ func (_m *API) PluginHTTP(request *http.Request) *http.Response {
 	return r0
 }
 
+// PublishPluginClusterEvent provides a mock function with given fields: ev, opts
+func (_m *API) PublishPluginClusterEvent(ev model.PluginClusterEvent, opts model.PluginClusterEventSendOptions) error {
+	ret := _m.Called(ev, opts)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(model.PluginClusterEvent, model.PluginClusterEventSendOptions) error); ok {
+		r0 = rf(ev, opts)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // PublishUserTyping provides a mock function with given fields: userID, channelId, parentId
 func (_m *API) PublishUserTyping(userID string, channelId string, parentId string) *model.AppError {
 	ret := _m.Called(userID, channelId, parentId)

plugin/plugintest/hooks.go

@@ -194,6 +194,11 @@ func (_m *Hooks) OnDeactivate() error {
 	return r0
 }
 
+// OnPluginClusterEvent provides a mock function with given fields: c, ev
+func (_m *Hooks) OnPluginClusterEvent(c *plugin.Context, ev model.PluginClusterEvent) {
+	_m.Called(c, ev)
+}
+
 // ReactionHasBeenAdded provides a mock function with given fields: c, reaction
 func (_m *Hooks) ReactionHasBeenAdded(c *plugin.Context, reaction *model.Reaction) {
 	_m.Called(c, reaction)

testlib/cluster.go

@@ -42,6 +42,13 @@ func (c *FakeClusterInterface) SendClusterMessage(message *model.ClusterMessage)
 	c.messages = append(c.messages, message)
 }
 
+func (c *FakeClusterInterface) SendClusterMessageToNode(nodeID string, message *model.ClusterMessage) error {
+	c.mut.Lock()
+	defer c.mut.Unlock()
+	c.messages = append(c.messages, message)
+	return nil
+}
+
 func (c *FakeClusterInterface) NotifyMsg(buf []byte) {}
 
 func (c *FakeClusterInterface) GetClusterStats() ([]*model.ClusterStats, *model.AppError) {