mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

app: add plugin-api/cluster interface provider (#19720)

Просмотр исходного кода
Этот коммит содержится в:
Ibrahim Serdar Acikgoz
2022-03-21 21:38:48 +03:00
коммит произвёл GitHub
родитель dae769221d
Коммит 42125bff4a
3 изменённых файлов: 91 добавлений и 11 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

56
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.

38
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)
}

8
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 {