From 98ff5fab3243ffa3cbd239770e49582c0944dbc5 Mon Sep 17 00:00:00 2001 From: Jesse Hallam Date: Thu, 18 Jul 2019 15:05:53 -0300 Subject: [PATCH] MM-16261: Synchronize plugins in HA (#11657) * MM-16272 - Synchronize plugins across cluster (#11611) * MM-16272 - Synchronize plugins across cluster * Adding a test * MM-16272 - Fixed tests * MM-16272 - PR feedback * MM-16270 - Plugin Sync (#11615) * Initial implementation for plugin synch with file store. WIP * Removed ListAll implementation. Used ListDirectory and change localstore to be consistent and return all items (files and folders) from directory * Refactored plugin filestore operations out of main install/remove plugin * Fixing error handling details * Changes to use structured logging * More logging fixes * Wording and comments improvements * Error handling and control flow improvements * Changed managed flag check to use os.stat * Added file store plugin dir and filename consts * Replaced FileRead to use a the FileReader in PluginSync * Minor styling and PR feedback changes * Minor error handling improvements * Added unit test for SyncPlugins. Changed SyncPlugins to use plugins environment to list available plugins * PR Feedback improvements * Minor err handling fix * Removing FileStorePath from PluginEventData (#11644) * Fix plugin path (#11654) * tweak path, logging Fix an issue not finding the plugins folder in S3. Tweak logging messages to add additional clarity. * Removing FileExists check when Syncing plugins. Updated localstore to not return an error when directory does not exist * PR Feedback * Install prepackaged plugins locally only (#11656) * s/uninstall/remove * Updated ClusterMessage comment * Updated PluginSync to test against s3 + local storage --- api4/plugin_test.go | 74 ++++++++++++++++ app/cluster_handlers.go | 11 +++ app/file.go | 13 +++ app/plugin.go | 72 +++++++++++++++- app/plugin_event.go | 20 +++++ app/plugin_install.go | 112 ++++++++++++++++++++----- app/plugin_test.go | 93 ++++++++++++++++++++ i18n/en.json | 12 +++ model/cluster_message.go | 3 + model/plugin_event_data.go | 25 ++++++ services/filesstore/filesstore_test.go | 21 ++++- services/filesstore/localstore.go | 7 +- services/filesstore/s3store.go | 6 +- testlib/cluster.go | 13 ++- 14 files changed, 453 insertions(+), 29 deletions(-) create mode 100644 app/plugin_event.go create mode 100644 model/plugin_event_data.go diff --git a/api4/plugin_test.go b/api4/plugin_test.go index 88c6301c01..482ad32f6f 100644 --- a/api4/plugin_test.go +++ b/api4/plugin_test.go @@ -14,8 +14,10 @@ import ( "testing" "github.com/mattermost/mattermost-server/model" + "github.com/mattermost/mattermost-server/testlib" "github.com/mattermost/mattermost-server/utils/fileutils" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestPlugin(t *testing.T) { @@ -234,3 +236,75 @@ func TestPlugin(t *testing.T) { _, resp = th.SystemAdminClient.RemovePlugin("bad.id") CheckBadRequestStatus(t, resp) } + +func TestNotifyClusterPluginEvent(t *testing.T) { + th := Setup().InitBasic() + defer th.TearDown() + + testCluster := &testlib.FakeClusterInterface{} + th.Server.Cluster = testCluster + + th.App.UpdateConfig(func(cfg *model.Config) { + *cfg.PluginSettings.Enable = true + *cfg.PluginSettings.EnableUploads = true + }) + + path, _ := fileutils.FindDir("tests") + tarData, err := ioutil.ReadFile(filepath.Join(path, "testplugin.tar.gz")) + if err != nil { + t.Fatal(err) + } + + // Successful upload + manifest, resp := th.SystemAdminClient.UploadPlugin(bytes.NewReader(tarData)) + CheckNoError(t, resp) + require.Equal(t, "testplugin", manifest.Id) + + // Stored in File Store: Upload Plugin case + expectedPath := filepath.Join("./plugins", manifest.Id) + ".tar.gz" + pluginStored, err := th.App.FileExists(expectedPath) + require.Nil(t, err) + require.True(t, pluginStored) + + expectedPluginData := model.PluginEventData{ + Id: manifest.Id, + } + expectedInstallMessage := &model.ClusterMessage{ + Event: model.CLUSTER_EVENT_INSTALL_PLUGIN, + SendType: model.CLUSTER_SEND_RELIABLE, + WaitForAllToSend: true, + Data: expectedPluginData.ToJson(), + } + expectedMessages := findClusterMessages(model.CLUSTER_EVENT_INSTALL_PLUGIN, testCluster.GetMessages()) + require.Equal(t, []*model.ClusterMessage{expectedInstallMessage}, expectedMessages) + + // Successful remove + testCluster.ClearMessages() + + ok, resp := th.SystemAdminClient.RemovePlugin(manifest.Id) + CheckNoError(t, resp) + require.True(t, ok) + + expectedRemoveMessage := &model.ClusterMessage{ + Event: model.CLUSTER_EVENT_REMOVE_PLUGIN, + SendType: model.CLUSTER_SEND_RELIABLE, + WaitForAllToSend: true, + Data: expectedPluginData.ToJson(), + } + expectedMessages = findClusterMessages(model.CLUSTER_EVENT_REMOVE_PLUGIN, testCluster.GetMessages()) + require.Equal(t, []*model.ClusterMessage{expectedRemoveMessage}, expectedMessages) + + pluginStored, err = th.App.FileExists(expectedPath) + require.Nil(t, err) + require.False(t, pluginStored) +} + +func findClusterMessages(event string, msgs []*model.ClusterMessage) []*model.ClusterMessage { + var result []*model.ClusterMessage + for _, msg := range msgs { + if msg.Event == event { + result = append(result, msg) + } + } + return result +} diff --git a/app/cluster_handlers.go b/app/cluster_handlers.go index 71ee028b55..cdca7b2a71 100644 --- a/app/cluster_handlers.go +++ b/app/cluster_handlers.go @@ -23,6 +23,9 @@ func (a *App) RegisterAllClusterMessageHandlers() { a.Cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_INVALIDATE_CACHE_FOR_USER_TEAMS, a.ClusterInvalidateCacheForUserTeamsHandler) a.Cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_CLEAR_SESSION_CACHE_FOR_USER, a.ClusterClearSessionCacheForUserHandler) 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) + } func (a *App) ClusterPublishHandler(msg *model.ClusterMessage) { @@ -78,3 +81,11 @@ func (a *App) ClusterClearSessionCacheForUserHandler(msg *model.ClusterMessage) func (a *App) ClusterClearSessionCacheForAllUsersHandler(msg *model.ClusterMessage) { a.ClearSessionCacheForAllUsersSkipClusterSend() } + +func (a *App) ClusterInstallPluginHandler(msg *model.ClusterMessage) { + a.InstallPluginFromData(model.PluginEventDataFromJson(strings.NewReader(msg.Data))) +} + +func (a *App) ClusterRemovePluginHandler(msg *model.ClusterMessage) { + a.RemovePluginFromData(model.PluginEventDataFromJson(strings.NewReader(msg.Data))) +} diff --git a/app/file.go b/app/file.go index 9d9d961e22..3ae1641560 100644 --- a/app/file.go +++ b/app/file.go @@ -122,6 +122,19 @@ func (a *App) RemoveFile(path string) *model.AppError { return backend.RemoveFile(path) } +func (a *App) ListDirectory(path string) ([]string, *model.AppError) { + backend, err := a.FileBackend() + if err != nil { + return nil, err + } + paths, err := backend.ListDirectory(path) + if err != nil { + return nil, err + } + + return *paths, nil +} + func (a *App) GetInfoForFilename(post *model.Post, teamId string, filename string) *model.FileInfo { // Find the path from the Filename of the form /{channelId}/{userId}/{uid}/{nameWithExtension} split := strings.SplitN(filename, "/", 5) diff --git a/app/plugin.go b/app/plugin.go index a8454132f0..5c60e682c9 100644 --- a/app/plugin.go +++ b/app/plugin.go @@ -12,6 +12,7 @@ import ( "github.com/mattermost/mattermost-server/mlog" "github.com/mattermost/mattermost-server/model" "github.com/mattermost/mattermost-server/plugin" + "github.com/mattermost/mattermost-server/services/filesstore" "github.com/mattermost/mattermost-server/utils/fileutils" ) @@ -146,6 +147,10 @@ func (a *App) InitPlugins(pluginDir, webappPluginDir string) { } a.SetPluginsEnvironment(env) + if err := a.SyncPlugins(); err != nil { + mlog.Error("Failed to sync plugins from the file store", mlog.Err(err)) + } + prepackagedPluginsDir, found := fileutils.FindDir("prepackaged_plugins") if found { if err := filepath.Walk(prepackagedPluginsDir, func(walkPath string, info os.FileInfo, err error) error { @@ -155,7 +160,7 @@ func (a *App) InitPlugins(pluginDir, webappPluginDir string) { if fileReader, err := os.Open(walkPath); err != nil { mlog.Error("Failed to open prepackaged plugin", mlog.Err(err), mlog.String("path", walkPath)) - } else if _, err := a.InstallPlugin(fileReader, true); err != nil { + } else if _, err := a.installPluginLocally(fileReader, true); err != nil { mlog.Error("Failed to unpack prepackaged plugin", mlog.Err(err), mlog.String("path", walkPath)) } @@ -182,6 +187,71 @@ func (a *App) InitPlugins(pluginDir, webappPluginDir string) { a.SyncPluginsActiveState() } +// SyncPlugins synchronizes the plugins installed locally +// with the plugin bundles available in the file store. +func (a *App) SyncPlugins() *model.AppError { + mlog.Info("Syncing plugins from the file store") + + pluginsEnvironment := a.GetPluginsEnvironment() + if pluginsEnvironment == nil { + return model.NewAppError("SyncPlugins", "app.plugin.disabled.app_error", nil, "", http.StatusNotImplemented) + } + + availablePlugins, err := pluginsEnvironment.Available() + if err != nil { + return model.NewAppError("SyncPlugins", "app.plugin.sync.read_local_folder.app_error", nil, err.Error(), http.StatusInternalServerError) + } + + for _, plugin := range availablePlugins { + pluginId := plugin.Manifest.Id + + // Only handle managed plugins with .filestore flag file. + _, err := os.Stat(filepath.Join(*a.Config().PluginSettings.Directory, pluginId, managedPluginFileName)) + if os.IsNotExist(err) { + mlog.Warn("Skipping sync for unmanaged plugin", mlog.String("plugin_id", pluginId)) + } else if err != nil { + mlog.Error("Skipping sync for plugin after failure to check if managed", mlog.String("plugin_id", pluginId), mlog.Err(err)) + } else { + mlog.Debug("Removing local installation of managed plugin before sync", mlog.String("plugin_id", pluginId)) + if err := a.removePluginLocally(pluginId); err != nil { + mlog.Error("Failed to remove local installation of managed plugin before sync", mlog.String("plugin_id", pluginId), mlog.Err(err)) + } + } + } + + // Install plugins from the file store. + fileStorePaths, appErr := a.ListDirectory(fileStorePluginFolder) + if appErr != nil { + return model.NewAppError("SyncPlugins", "app.plugin.sync.list_filestore.app_error", nil, appErr.Error(), http.StatusInternalServerError) + } + if len(fileStorePaths) == 0 { + mlog.Info("Found no files in plugins file store") + return nil + } + + for _, path := range fileStorePaths { + if !strings.HasSuffix(path, ".tar.gz") { + mlog.Warn("Ignoring non-plugin in file store", mlog.String("bundle", path)) + continue + } + + var reader filesstore.ReadCloseSeeker + reader, appErr = a.FileReader(path) + if appErr != nil { + mlog.Error("Failed to open plugin bundle from file store.", mlog.String("bundle", path), mlog.Err(appErr)) + continue + } + defer reader.Close() + + mlog.Info("Syncing plugin from file store", mlog.String("bundle", path)) + if _, err := a.installPluginLocally(reader, true); err != nil { + mlog.Error("Failed to sync plugin from file store", mlog.String("bundle", path), mlog.Err(err)) + } + } + + return nil +} + func (a *App) ShutDownPlugins() { a.Srv.PluginsLock.Lock() pluginsEnvironment := a.Srv.PluginsEnvironment diff --git a/app/plugin_event.go b/app/plugin_event.go new file mode 100644 index 0000000000..d9785b8921 --- /dev/null +++ b/app/plugin_event.go @@ -0,0 +1,20 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package app + +import ( + "github.com/mattermost/mattermost-server/model" +) + +// notifyClusterPluginEvent publishes `event` to other clusters. +func (a *App) notifyClusterPluginEvent(event string, data model.PluginEventData) { + if a.Cluster != nil { + a.Cluster.SendClusterMessage(&model.ClusterMessage{ + Event: event, + SendType: model.CLUSTER_SEND_RELIABLE, + WaitForAllToSend: true, + Data: data.ToJson(), + }) + } +} diff --git a/app/plugin_install.go b/app/plugin_install.go index 367a33656b..d6a8dc9242 100644 --- a/app/plugin_install.go +++ b/app/plugin_install.go @@ -4,6 +4,7 @@ package app import ( + "fmt" "io" "io/ioutil" "net/http" @@ -16,12 +17,65 @@ import ( "github.com/mattermost/mattermost-server/utils" ) +// managedPluginFileName is the file name of the flag file that marks +// a local plugin folder as "managed" by the file store. +const managedPluginFileName = ".filestore" + +// fileStorePluginFolder is the folder name in the file store of the plugin bundles installed. +const fileStorePluginFolder = "plugins" + +func (a *App) InstallPluginFromData(data model.PluginEventData) { + mlog.Debug("Installing plugin as per cluster message", mlog.String("plugin_id", data.Id)) + + fileStorePath := a.getBundleStorePath(data.Id) + reader, appErr := a.FileReader(fileStorePath) + if appErr != nil { + mlog.Error("Failed to open plugin bundle from filestore.", mlog.String("path", fileStorePath), mlog.Err(appErr)) + } + defer reader.Close() + + if _, appErr = a.installPluginLocally(reader, true); appErr != nil { + mlog.Error("Failed to unpack plugin from filestore", mlog.Err(appErr), mlog.String("path", fileStorePath)) + } +} + +func (a *App) RemovePluginFromData(data model.PluginEventData) { + mlog.Debug("Removing plugin as per cluster message", mlog.String("plugin_id", data.Id)) + + if err := a.removePluginLocally(data.Id); err != nil { + mlog.Error("Failed to remove plugin locally", mlog.Err(err), mlog.String("id", data.Id)) + } +} + // InstallPlugin unpacks and installs a plugin but does not enable or activate it. func (a *App) InstallPlugin(pluginFile io.ReadSeeker, replace bool) (*model.Manifest, *model.AppError) { return a.installPlugin(pluginFile, replace) } func (a *App) installPlugin(pluginFile io.ReadSeeker, replace bool) (*model.Manifest, *model.AppError) { + manifest, appErr := a.installPluginLocally(pluginFile, replace) + if appErr != nil { + return nil, appErr + } + + // Store bundle in the file store to allow access from other servers. + pluginFile.Seek(0, 0) + + if _, err := a.WriteFile(pluginFile, a.getBundleStorePath(manifest.Id)); err != nil { + return nil, model.NewAppError("uploadPlugin", "app.plugin.store_bundle.app_error", nil, err.Error(), http.StatusInternalServerError) + } + + a.notifyClusterPluginEvent( + model.CLUSTER_EVENT_INSTALL_PLUGIN, + model.PluginEventData{ + Id: manifest.Id, + }, + ) + + return manifest, nil +} + +func (a *App) installPluginLocally(pluginFile io.ReadSeeker, replace bool) (*model.Manifest, *model.AppError) { pluginsEnvironment := a.GetPluginsEnvironment() if pluginsEnvironment == nil { return nil, model.NewAppError("installPlugin", "app.plugin.disabled.app_error", nil, "", http.StatusNotImplemented) @@ -71,7 +125,7 @@ func (a *App) installPlugin(pluginFile io.ReadSeeker, replace bool) (*model.Mani return nil, model.NewAppError("installPlugin", "app.plugin.install_id.app_error", nil, "", http.StatusBadRequest) } - if err := a.RemovePlugin(manifest.Id); err != nil { + if err := a.removePluginLocally(manifest.Id); err != nil { return nil, model.NewAppError("installPlugin", "app.plugin.install_id_failed_remove.app_error", nil, "", http.StatusBadRequest) } } @@ -83,13 +137,12 @@ func (a *App) installPlugin(pluginFile io.ReadSeeker, replace bool) (*model.Mani return nil, model.NewAppError("installPlugin", "app.plugin.mvdir.app_error", nil, err.Error(), http.StatusInternalServerError) } - // Store bundle in the file store to allow access from other servers. - pluginFile.Seek(0, 0) - - storePluginFileName := filepath.Join("./plugins", manifest.Id) + ".tar.gz" - if _, err := a.WriteFile(pluginFile, storePluginFileName); err != nil { - return nil, model.NewAppError("uploadPlugin", "app.plugin.store_bundle.app_error", nil, err.Error(), http.StatusInternalServerError) + // Flag plugin locally as managed by the filestore. + f, err := os.Create(filepath.Join(pluginPath, managedPluginFileName)) + if err != nil { + return nil, model.NewAppError("uploadPlugin", "app.plugin.flag_managed.app_error", nil, err.Error(), http.StatusInternalServerError) } + f.Close() if stashed != nil && stashed.Enable { a.EnablePlugin(manifest.Id) @@ -107,6 +160,34 @@ func (a *App) RemovePlugin(id string) *model.AppError { } func (a *App) removePlugin(id string) *model.AppError { + if err := a.removePluginLocally(id); err != nil { + return err + } + + // Remove bundle from the file store. + storePluginFileName := a.getBundleStorePath(id) + bundleExist, err := a.FileExists(storePluginFileName) + if err != nil { + return model.NewAppError("removePlugin", "app.plugin.remove_bundle.app_error", nil, err.Error(), http.StatusInternalServerError) + } + if !bundleExist { + return nil + } + if err := a.RemoveFile(storePluginFileName); err != nil { + return model.NewAppError("removePlugin", "app.plugin.remove_bundle.app_error", nil, err.Error(), http.StatusInternalServerError) + } + + a.notifyClusterPluginEvent( + model.CLUSTER_EVENT_REMOVE_PLUGIN, + model.PluginEventData{ + Id: id, + }, + ) + + return nil +} + +func (a *App) removePluginLocally(id string) *model.AppError { pluginsEnvironment := a.GetPluginsEnvironment() if pluginsEnvironment == nil { return model.NewAppError("removePlugin", "app.plugin.disabled.app_error", nil, "", http.StatusNotImplemented) @@ -146,18 +227,9 @@ func (a *App) removePlugin(id string) *model.AppError { return model.NewAppError("removePlugin", "app.plugin.remove.app_error", nil, err.Error(), http.StatusInternalServerError) } - // Remove bundle from the file store. - storePluginFileName := filepath.Join("./plugins", manifest.Id) + ".tar.gz" - bundleExist, fileErr := a.FileExists(storePluginFileName) - if fileErr != nil { - return model.NewAppError("removePlugin", "app.plugin.remove_bundle.app_error", nil, err.Error(), http.StatusInternalServerError) - } - - if bundleExist { - if err := a.RemoveFile(storePluginFileName); err != nil { - return model.NewAppError("removePlugin", "app.plugin.remove_bundle.app_error", nil, err.Error(), http.StatusInternalServerError) - } - } - return nil } + +func (a *App) getBundleStorePath(id string) string { + return filepath.Join(fileStorePluginFolder, fmt.Sprintf("%s.tar.gz", id)) +} diff --git a/app/plugin_test.go b/app/plugin_test.go index 621cc2744d..6d6a9b533f 100644 --- a/app/plugin_test.go +++ b/app/plugin_test.go @@ -7,14 +7,18 @@ import ( "bytes" "crypto/sha256" "encoding/base64" + "fmt" "io/ioutil" "net/http" "net/http/httptest" + "os" + "path/filepath" "testing" "github.com/gorilla/mux" "github.com/mattermost/mattermost-server/model" "github.com/mattermost/mattermost-server/plugin" + "github.com/mattermost/mattermost-server/utils/fileutils" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -312,3 +316,92 @@ func TestGetPluginStatuses(t *testing.T) { require.Nil(t, err) require.NotNil(t, pluginStatuses) } + +func TestPluginSync(t *testing.T) { + th := Setup(t).InitBasic() + defer th.TearDown() + + testCases := []struct { + Description string + ConfigFunc func(cfg *model.Config) + }{ + { + "local", + func(cfg *model.Config) { + cfg.FileSettings.DriverName = model.NewString(model.IMAGE_DRIVER_LOCAL) + }, + }, + { + "s3", + func(cfg *model.Config) { + s3Host := os.Getenv("CI_MINIO_HOST") + if s3Host == "" { + s3Host = "dockerhost" + } + + s3Port := os.Getenv("CI_MINIO_PORT") + if s3Port == "" { + s3Port = "9001" + } + + s3Endpoint := fmt.Sprintf("%s:%s", s3Host, s3Port) + cfg.FileSettings.DriverName = model.NewString(model.IMAGE_DRIVER_S3) + cfg.FileSettings.AmazonS3AccessKeyId = model.NewString(model.MINIO_ACCESS_KEY) + cfg.FileSettings.AmazonS3SecretAccessKey = model.NewString(model.MINIO_SECRET_KEY) + cfg.FileSettings.AmazonS3Bucket = model.NewString(model.MINIO_BUCKET) + cfg.FileSettings.AmazonS3Endpoint = model.NewString(s3Endpoint) + cfg.FileSettings.AmazonS3Region = model.NewString("") + cfg.FileSettings.AmazonS3SSL = model.NewBool(false) + + }, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.Description, func(t *testing.T) { + os.MkdirAll("./test-plugins", os.ModePerm) + defer os.RemoveAll("./test-plugins") + + th.App.UpdateConfig(func(cfg *model.Config) { + *cfg.PluginSettings.Enable = true + *cfg.PluginSettings.Directory = "./test-plugins" + *cfg.PluginSettings.ClientDirectory = "./test-client-plugins" + }) + th.App.UpdateConfig(testCase.ConfigFunc) + + env, err := plugin.NewEnvironment(th.App.NewPluginAPI, "./test-plugins", "./test-client-plugins", th.App.Log) + require.NoError(t, err) + th.App.SetPluginsEnvironment(env) + + // New bundle in the file store case + path, _ := fileutils.FindDir("tests") + fileReader, err := os.Open(filepath.Join(path, "testplugin.tar.gz")) + require.NoError(t, err) + defer fileReader.Close() + + _, appErr := th.App.WriteFile(fileReader, th.App.getBundleStorePath("testplugin")) + checkNoError(t, appErr) + + appErr = th.App.SyncPlugins() + checkNoError(t, appErr) + + // Check if installed + pluginStatus, err := env.Statuses() + require.Nil(t, err) + require.True(t, len(pluginStatus) == 1) + require.Equal(t, pluginStatus[0].PluginId, "testplugin") + + // Bundle removed from the file store case + appErr = th.App.RemoveFile(th.App.getBundleStorePath("testplugin")) + checkNoError(t, appErr) + + appErr = th.App.SyncPlugins() + checkNoError(t, appErr) + + // Check if removed + pluginStatus, err = env.Statuses() + require.Nil(t, err) + require.True(t, len(pluginStatus) == 0) + }) + } +} diff --git a/i18n/en.json b/i18n/en.json index 76ddaba666..f0a638d428 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -3386,6 +3386,10 @@ "id": "app.plugin.filesystem.app_error", "translation": "Encountered filesystem error" }, + { + "id": "app.plugin.flag_managed.app_error", + "translation": "Unable to set plugin as managed by the file store." + }, { "id": "app.plugin.get_cluster_plugin_statuses.app_error", "translation": "Unable to get plugin statuses from the cluster." @@ -3438,6 +3442,14 @@ "id": "app.plugin.store_bundle.app_error", "translation": "Unable to store the plugin to the configured file store." }, + { + "id": "app.plugin.sync.list_filestore.app_error", + "translation": "Error reading files from the plugins folder in the file store." + }, + { + "id": "app.plugin.sync.read_local_folder.app_error", + "translation": "Error reading local plugins folder." + }, { "id": "app.plugin.upload_disabled.app_error", "translation": "Plugins and/or plugin uploads have been disabled." diff --git a/model/cluster_message.go b/model/cluster_message.go index 82e0151ea3..04d8673ceb 100644 --- a/model/cluster_message.go +++ b/model/cluster_message.go @@ -25,7 +25,10 @@ const ( CLUSTER_EVENT_INVALIDATE_CACHE_FOR_ROLES = "inv_roles" CLUSTER_EVENT_INVALIDATE_CACHE_FOR_SCHEMES = "inv_schemes" CLUSTER_EVENT_CLEAR_SESSION_CACHE_FOR_ALL_USERS = "inv_all_user_sessions" + CLUSTER_EVENT_INSTALL_PLUGIN = "install_plugin" + CLUSTER_EVENT_REMOVE_PLUGIN = "remove_plugin" + // SendTypes for ClusterMessage. CLUSTER_SEND_BEST_EFFORT = "best_effort" CLUSTER_SEND_RELIABLE = "reliable" ) diff --git a/model/plugin_event_data.go b/model/plugin_event_data.go new file mode 100644 index 0000000000..c704c99339 --- /dev/null +++ b/model/plugin_event_data.go @@ -0,0 +1,25 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package model + +import ( + "encoding/json" + "io" +) + +// PluginEventData used to notify peers about plugin changes. +type PluginEventData struct { + Id string `json:"id"` +} + +func (p *PluginEventData) ToJson() string { + b, _ := json.Marshal(p) + return string(b) +} + +func PluginEventDataFromJson(data io.Reader) PluginEventData { + var m PluginEventData + json.NewDecoder(data).Decode(&m) + return m +} diff --git a/services/filesstore/filesstore_test.go b/services/filesstore/filesstore_test.go index 57b540c2c6..4fa9c340c1 100644 --- a/services/filesstore/filesstore_test.go +++ b/services/filesstore/filesstore_test.go @@ -237,17 +237,29 @@ func (s *FileBackendTestSuite) TestListDirectory() { path1 := "19700101/" + model.NewId() path2 := "19800101/" + model.NewId() + paths, err := s.backend.ListDirectory("19700101") + s.Nil(err) + s.Len(*paths, 0) + written, err := s.backend.WriteFile(bytes.NewReader(b), path1) s.Nil(err) s.EqualValues(len(b), written, "expected given number of bytes to have been written") - defer s.backend.RemoveFile(path1) written, err = s.backend.WriteFile(bytes.NewReader(b), path2) s.Nil(err) s.EqualValues(len(b), written, "expected given number of bytes to have been written") - defer s.backend.RemoveFile(path2) - paths, err := s.backend.ListDirectory("") + paths, err = s.backend.ListDirectory("19700101") + s.Nil(err) + s.Len(*paths, 1) + s.Equal(path1, (*paths)[0]) + + paths, err = s.backend.ListDirectory("19700101/") + s.Nil(err) + s.Len(*paths, 1) + s.Equal(path1, (*paths)[0]) + + paths, err = s.backend.ListDirectory("") s.Nil(err) found1 := false @@ -261,6 +273,9 @@ func (s *FileBackendTestSuite) TestListDirectory() { } s.True(found1) s.True(found2) + + s.backend.RemoveFile(path1) + s.backend.RemoveFile(path2) } func (s *FileBackendTestSuite) TestRemoveDirectory() { diff --git a/services/filesstore/localstore.go b/services/filesstore/localstore.go index 9cafcaa55c..b52494190f 100644 --- a/services/filesstore/localstore.go +++ b/services/filesstore/localstore.go @@ -114,12 +114,13 @@ func (b *LocalFileBackend) ListDirectory(path string) (*[]string, *model.AppErro var paths []string fileInfos, err := ioutil.ReadDir(filepath.Join(b.directory, path)) if err != nil { + if os.IsNotExist(err) { + return &paths, nil + } return nil, model.NewAppError("ListDirectory", "utils.file.list_directory.local.app_error", nil, err.Error(), http.StatusInternalServerError) } for _, fileInfo := range fileInfos { - if fileInfo.IsDir() { - paths = append(paths, filepath.Join(path, fileInfo.Name())) - } + paths = append(paths, filepath.Join(path, fileInfo.Name())) } return &paths, nil } diff --git a/services/filesstore/s3store.go b/services/filesstore/s3store.go index 1218c33091..42113c1b43 100644 --- a/services/filesstore/s3store.go +++ b/services/filesstore/s3store.go @@ -238,9 +238,13 @@ func (b *S3FileBackend) ListDirectory(path string) (*[]string, *model.AppError) } doneCh := make(chan struct{}) - defer close(doneCh) + if !strings.HasSuffix(path, "/") && len(path) > 0 { + // s3Clnt returns only the path itself when "/" is not present + // appending "/" to make it consistent across all filesstores + path = path + "/" + } for object := range s3Clnt.ListObjects(b.bucket, path, false, doneCh) { if object.Err != nil { return nil, model.NewAppError("ListDirectory", "utils.file.list_directory.s3.app_error", nil, object.Err.Error(), http.StatusInternalServerError) diff --git a/testlib/cluster.go b/testlib/cluster.go index 532672baad..244c8e5b74 100644 --- a/testlib/cluster.go +++ b/testlib/cluster.go @@ -10,6 +10,7 @@ import ( type FakeClusterInterface struct { clusterMessageHandler einterfaces.ClusterMessageHandler + messages []*model.ClusterMessage } func (c *FakeClusterInterface) StartInterNodeCommunication() {} @@ -28,7 +29,9 @@ func (c *FakeClusterInterface) GetMyClusterInfo() *model.ClusterInfo { return ni func (c *FakeClusterInterface) GetClusterInfos() []*model.ClusterInfo { return nil } -func (c *FakeClusterInterface) SendClusterMessage(cluster *model.ClusterMessage) {} +func (c *FakeClusterInterface) SendClusterMessage(message *model.ClusterMessage) { + c.messages = append(c.messages, message) +} func (c *FakeClusterInterface) NotifyMsg(buf []byte) {} @@ -53,3 +56,11 @@ func (c *FakeClusterInterface) SendClearRoleCacheMessage() { func (c *FakeClusterInterface) GetPluginStatuses() (model.PluginStatuses, *model.AppError) { return nil, nil } + +func (c *FakeClusterInterface) GetMessages() []*model.ClusterMessage { + return c.messages +} + +func (c *FakeClusterInterface) ClearMessages() { + c.messages = nil +}