From 6af032d06a1753d818b51ad657618d8ad6c825c8 Mon Sep 17 00:00:00 2001 From: Claudio Costa Date: Wed, 28 Apr 2021 19:59:32 +0200 Subject: [PATCH] [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 | 4 ++ app/cluster_handlers.go | 30 +++++++++++++ app/plugin_api.go | 31 +++++++++++++ einterfaces/cluster.go | 3 +- einterfaces/mocks/ClusterInterface.go | 20 +++++++-- model/cluster_message.go | 1 + model/plugin_cluster_event.go | 28 ++++++++++++ plugin/api.go | 10 +++++ plugin/api_timer_layer_generated.go | 7 +++ plugin/client_rpc_generated.go | 64 +++++++++++++++++++++++++++ plugin/hooks.go | 10 +++++ plugin/hooks_timer_layer_generated.go | 6 +++ plugin/plugintest/api.go | 14 ++++++ plugin/plugintest/hooks.go | 5 +++ testlib/cluster.go | 7 +++ 15 files changed, 236 insertions(+), 4 deletions(-) create mode 100644 model/plugin_cluster_event.go diff --git a/app/busy_test.go b/app/busy_test.go index c810b3f6ae..947cba1f90 100644 --- a/app/busy_test.go +++ b/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) { diff --git a/app/cluster_handlers.go b/app/cluster_handlers.go index 9adf95b4e4..a6ac6067e8 100644 --- a/app/cluster_handlers.go +++ b/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))) } diff --git a/app/plugin_api.go b/app/plugin_api.go index 4cfae495ed..3fed6d442f 100644 --- a/app/plugin_api.go +++ b/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 +} diff --git a/einterfaces/cluster.go b/einterfaces/cluster.go index e2b4bb10d7..bf70c0b85c 100644 --- a/einterfaces/cluster.go +++ b/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) diff --git a/einterfaces/mocks/ClusterInterface.go b/einterfaces/mocks/ClusterInterface.go index 8e18152968..626aa64f75 100644 --- a/einterfaces/mocks/ClusterInterface.go +++ b/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 -func (_m *ClusterInterface) SendClusterMessage(cluster *model.ClusterMessage) { - _m.Called(cluster) +// SendClusterMessage provides a mock function with given fields: msg +func (_m *ClusterInterface) SendClusterMessage(msg *model.ClusterMessage) { + _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: diff --git a/model/cluster_message.go b/model/cluster_message.go index 529f4a93a2..bd73ba8ddd 100644 --- a/model/cluster_message.go +++ b/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" diff --git a/model/plugin_cluster_event.go b/model/plugin_cluster_event.go new file mode 100644 index 0000000000..ba5c805278 --- /dev/null +++ b/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 +} diff --git a/plugin/api.go b/plugin/api.go index 43c4f0404e..f5210f7df4 100644 --- a/plugin/api.go +++ b/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{ diff --git a/plugin/api_timer_layer_generated.go b/plugin/api_timer_layer_generated.go index 433ce3fb42..cd14e1b7e0 100644 --- a/plugin/api_timer_layer_generated.go +++ b/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 +} diff --git a/plugin/client_rpc_generated.go b/plugin/client_rpc_generated.go index 61c8cd4ce0..e7c744f7b6 100644 --- a/plugin/client_rpc_generated.go +++ b/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 +} diff --git a/plugin/hooks.go b/plugin/hooks.go index fe3ed36f7d..b002b6a53e 100644 --- a/plugin/hooks.go +++ b/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) } diff --git a/plugin/hooks_timer_layer_generated.go b/plugin/hooks_timer_layer_generated.go index 05a0a4bcaf..81635d7a13 100644 --- a/plugin/hooks_timer_layer_generated.go +++ b/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) +} diff --git a/plugin/plugintest/api.go b/plugin/plugintest/api.go index 8c569705d3..32f1a1d9bb 100644 --- a/plugin/plugintest/api.go +++ b/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) diff --git a/plugin/plugintest/hooks.go b/plugin/plugintest/hooks.go index 56842adf99..c62e4f6447 100644 --- a/plugin/plugintest/hooks.go +++ b/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) diff --git a/testlib/cluster.go b/testlib/cluster.go index cd17aa7bc6..c40c30b07d 100644 --- a/testlib/cluster.go +++ b/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) {