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

@@ -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
+}

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

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)

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

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(),
+		})
+	}
+}

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

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

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

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"
 )

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
+}

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() {

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
 }

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)

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
+}