From 42125bff4a4fccd440bf84647034ca91c6253705 Mon Sep 17 00:00:00 2001 From: Ibrahim Serdar Acikgoz Date: Mon, 21 Mar 2022 21:38:48 +0300 Subject: [PATCH] app: add plugin-api/cluster interface provider (#19720) --- app/cluster.go | 56 +++++++++++++++++++++++++++++++++++ app/plugin_key_value_store.go | 38 +++++++++++++++++------- app/server.go | 8 +++++ 3 files changed, 91 insertions(+), 11 deletions(-) diff --git a/app/cluster.go b/app/cluster.go index db1f926747..4d0850f980 100644 --- a/app/cluster.go +++ b/app/cluster.go @@ -4,9 +4,65 @@ package app import ( + "fmt" + "github.com/mattermost/mattermost-server/v6/model" + "github.com/mattermost/mattermost-server/v6/shared/mlog" ) +type clusterWrapper struct { + srv *Server +} + +func (s *clusterWrapper) PublishPluginClusterEvent(productID string, ev model.PluginClusterEvent, + opts model.PluginClusterEventSendOptions) error { + if s.srv.Cluster == nil { + return nil + } + + msg := &model.ClusterMessage{ + Event: model.ClusterEventPluginEvent, + SendType: opts.SendType, + WaitForAllToSend: false, + Props: map[string]string{ + "ProductID": productID, + "EventID": ev.Id, + }, + Data: ev.Data, + } + + // If TargetId is empty we broadcast to all other cluster nodes. + if opts.TargetId == "" { + s.srv.Cluster.SendClusterMessage(msg) + } else { + if err := s.srv.Cluster.SendClusterMessageToNode(opts.TargetId, msg); err != nil { + return fmt.Errorf("failed to send message to cluster node %q: %w", opts.TargetId, err) + } + } + + return nil +} + +func (s *clusterWrapper) SetPluginKeyWithOptions(productID string, key string, value []byte, options model.PluginKVSetOptions) (bool, *model.AppError) { + return s.srv.setPluginKeyWithOptions(productID, key, value, options) +} + +func (s *clusterWrapper) LogError(productID, msg string, keyValuePairs ...interface{}) { + s.srv.Log.Error(msg, mlog.String("product_id", productID), mlog.Map("key-value pairs", keyValuePairs)) +} + +func (s *clusterWrapper) KVGet(productID, key string) ([]byte, *model.AppError) { + return s.srv.getPluginKey(productID, key) +} + +func (s *clusterWrapper) KVDelete(productID, key string) *model.AppError { + return s.srv.deletePluginKey(productID, key) +} + +func (s *clusterWrapper) KVList(productID string, page, perPage int) ([]string, *model.AppError) { + return s.srv.listPluginKeys(productID, page, perPage) +} + // Registers a given function to be called when the cluster leader may have changed. Returns a unique ID for the // listener which can later be used to remove it. If clustering is not enabled in this build, the callback will never // be called. diff --git a/app/plugin_key_value_store.go b/app/plugin_key_value_store.go index 12dcc6a60e..c86b3493cc 100644 --- a/app/plugin_key_value_store.go +++ b/app/plugin_key_value_store.go @@ -40,13 +40,13 @@ func (a *App) CompareAndSetPluginKey(pluginID string, key string, oldValue, newV return a.SetPluginKeyWithOptions(pluginID, key, newValue, options) } -func (a *App) SetPluginKeyWithOptions(pluginID string, key string, value []byte, options model.PluginKVSetOptions) (bool, *model.AppError) { +func (s *Server) setPluginKeyWithOptions(pluginID string, key string, value []byte, options model.PluginKVSetOptions) (bool, *model.AppError) { if err := options.IsValid(); err != nil { mlog.Debug("Failed to set plugin key value with options", mlog.String("plugin_id", pluginID), mlog.String("key", key), mlog.Err(err)) return false, err } - updated, err := a.Srv().Store.Plugin().SetWithOptions(pluginID, key, value, options) + updated, err := s.Store.Plugin().SetWithOptions(pluginID, key, value, options) if err != nil { mlog.Error("Failed to set plugin key value with options", mlog.String("plugin_id", pluginID), mlog.String("key", key), mlog.Err(err)) var appErr *model.AppError @@ -59,13 +59,17 @@ func (a *App) SetPluginKeyWithOptions(pluginID string, key string, value []byte, } // Clean up a previous entry using the hashed key, if it exists. - if err := a.Srv().Store.Plugin().Delete(pluginID, getKeyHash(key)); err != nil { + if err := s.Store.Plugin().Delete(pluginID, getKeyHash(key)); err != nil { mlog.Warn("Failed to clean up previously hashed plugin key value", mlog.String("plugin_id", pluginID), mlog.String("key", key), mlog.Err(err)) } return updated, nil } +func (a *App) SetPluginKeyWithOptions(pluginID string, key string, value []byte, options model.PluginKVSetOptions) (bool, *model.AppError) { + return a.Srv().setPluginKeyWithOptions(pluginID, key, value, options) +} + func (a *App) CompareAndDeletePluginKey(pluginID string, key string, oldValue []byte) (bool, *model.AppError) { kv := &model.PluginKeyValue{ PluginId: pluginID, @@ -92,8 +96,8 @@ func (a *App) CompareAndDeletePluginKey(pluginID string, key string, oldValue [] return deleted, nil } -func (a *App) GetPluginKey(pluginID string, key string) ([]byte, *model.AppError) { - if kv, err := a.Srv().Store.Plugin().Get(pluginID, key); err == nil { +func (s *Server) getPluginKey(pluginID string, key string) ([]byte, *model.AppError) { + if kv, err := s.Store.Plugin().Get(pluginID, key); err == nil { return kv.Value, nil } else if nfErr := new(store.ErrNotFound); !errors.As(err, &nfErr) { mlog.Error("Failed to query plugin key value", mlog.String("plugin_id", pluginID), mlog.String("key", key), mlog.Err(err)) @@ -101,7 +105,7 @@ func (a *App) GetPluginKey(pluginID string, key string) ([]byte, *model.AppError } // Lookup using the hashed version of the key for keys written prior to v5.6. - if kv, err := a.Srv().Store.Plugin().Get(pluginID, getKeyHash(key)); err == nil { + if kv, err := s.Store.Plugin().Get(pluginID, getKeyHash(key)); err == nil { return kv.Value, nil } else if nfErr := new(store.ErrNotFound); !errors.As(err, &nfErr) { mlog.Error("Failed to query plugin key value using hashed key", mlog.String("plugin_id", pluginID), mlog.String("key", key), mlog.Err(err)) @@ -111,14 +115,18 @@ func (a *App) GetPluginKey(pluginID string, key string) ([]byte, *model.AppError return nil, nil } -func (a *App) DeletePluginKey(pluginID string, key string) *model.AppError { - if err := a.Srv().Store.Plugin().Delete(pluginID, getKeyHash(key)); err != nil { +func (a *App) GetPluginKey(pluginID string, key string) ([]byte, *model.AppError) { + return a.Srv().getPluginKey(pluginID, key) +} + +func (s *Server) deletePluginKey(pluginID string, key string) *model.AppError { + if err := s.Store.Plugin().Delete(pluginID, getKeyHash(key)); err != nil { mlog.Error("Failed to delete plugin key value", mlog.String("plugin_id", pluginID), mlog.String("key", key), mlog.Err(err)) return model.NewAppError("DeletePluginKey", "app.plugin_store.delete.app_error", nil, err.Error(), http.StatusInternalServerError) } // Also delete the key without hashing - if err := a.Srv().Store.Plugin().Delete(pluginID, key); err != nil { + if err := s.Store.Plugin().Delete(pluginID, key); err != nil { mlog.Error("Failed to delete plugin key value using hashed key", mlog.String("plugin_id", pluginID), mlog.String("key", key), mlog.Err(err)) return model.NewAppError("DeletePluginKey", "app.plugin_store.delete.app_error", nil, err.Error(), http.StatusInternalServerError) } @@ -126,6 +134,10 @@ func (a *App) DeletePluginKey(pluginID string, key string) *model.AppError { return nil } +func (a *App) DeletePluginKey(pluginID string, key string) *model.AppError { + return a.Srv().deletePluginKey(pluginID, key) +} + func (a *App) DeleteAllKeysForPlugin(pluginID string) *model.AppError { if err := a.Srv().Store.Plugin().DeleteAllForPlugin(pluginID); err != nil { mlog.Error("Failed to delete all plugin key values", mlog.String("plugin_id", pluginID), mlog.Err(err)) @@ -148,8 +160,8 @@ func (a *App) DeleteAllExpiredPluginKeys() *model.AppError { return nil } -func (a *App) ListPluginKeys(pluginID string, page, perPage int) ([]string, *model.AppError) { - data, err := a.Srv().Store.Plugin().List(pluginID, page*perPage, perPage) +func (s *Server) listPluginKeys(pluginID string, page, perPage int) ([]string, *model.AppError) { + data, err := s.Store.Plugin().List(pluginID, page*perPage, perPage) if err != nil { mlog.Error("Failed to list plugin key values", mlog.Int("page", page), mlog.Int("perPage", perPage), mlog.Err(err)) @@ -158,3 +170,7 @@ func (a *App) ListPluginKeys(pluginID string, page, perPage int) ([]string, *mod return data, nil } + +func (a *App) ListPluginKeys(pluginID string, page, perPage int) ([]string, *model.AppError) { + return a.Srv().listPluginKeys(pluginID, page, perPage) +} diff --git a/app/server.go b/app/server.go index 63be3d67af..4f473d2e97 100644 --- a/app/server.go +++ b/app/server.go @@ -93,6 +93,7 @@ const ( ConfigKey ServiceKey = "config" LicenseKey ServiceKey = "license" FilestoreKey ServiceKey = "filestore" + ClusterKey ServiceKey = "cluster" ) type Server struct { @@ -140,6 +141,7 @@ type Server struct { Jobs *jobs.JobServer clusterLeaderListeners sync.Map + clusterWrapper *clusterWrapper licenseValue atomic.Value clientLicenseValue atomic.Value @@ -363,11 +365,17 @@ func NewServer(options ...Option) (*Server, error) { srv: s, } + s.clusterWrapper = &clusterWrapper{ + srv: s, + } + serviceMap := map[ServiceKey]interface{}{ ConfigKey: s.configStore, LicenseKey: s.licenseWrapper, FilestoreKey: s.filestore, + ClusterKey: s.clusterWrapper, } + // Step 8: Initialize products. // Depends on s.httpService. for name, initializer := range products {