MM-27570,MM-27757 - Activate/Deactivate plugins concurrently (#15244)

* Start plugins concurrently

* Determine which plugins need to be activated/deactivated before changing their state

* Test activation/deactivation of plugins in TestSyncPluginsActiveState

* Change test comments

* remove unneeded temp var

app/plugin.go

@@ -86,46 +86,52 @@ func (a *App) SyncPluginsActiveState() {
 			return
 		}
 
-		// Deactivate any plugins that have been disabled.
+		// Determine which plugins need to be activated or deactivated.
+		disabledPlugins := []*model.BundleInfo{}
+		enabledPlugins := []*model.BundleInfo{}
 		for _, plugin := range availablePlugins {
-			// Determine if plugin is enabled
 			pluginId := plugin.Manifest.Id
 			pluginEnabled := false
 			if state, ok := config.PluginStates[pluginId]; ok {
 				pluginEnabled = state.Enable
 			}
 
-			// If it's not enabled we need to deactivate it
-			if !pluginEnabled {
-				deactivated := pluginsEnvironment.Deactivate(pluginId)
+			if pluginEnabled {
+				enabledPlugins = append(enabledPlugins, plugin)
+			} else {
+				disabledPlugins = append(disabledPlugins, plugin)
+			}
+		}
+
+		// Concurrently activate/deactivate each plugin appropriately.
+		var wg sync.WaitGroup
+
+		// Deactivate any plugins that have been disabled.
+		for _, plugin := range disabledPlugins {
+			wg.Add(1)
+			go func(plugin *model.BundleInfo) {
+				defer wg.Done()
+
+				deactivated := pluginsEnvironment.Deactivate(plugin.Manifest.Id)
 				if deactivated && plugin.Manifest.HasClient() {
 					message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_PLUGIN_DISABLED, "", "", "", nil)
 					message.Add("manifest", plugin.Manifest.ClientManifest())
 					a.Publish(message)
 				}
-			}
+			}(plugin)
 		}
 
 		// Activate any plugins that have been enabled
-		for _, plugin := range availablePlugins {
-			if plugin.Manifest == nil {
-				plugin.WrapLogger(a.Log()).Error("Plugin manifest could not be loaded", mlog.Err(plugin.ManifestError))
-				continue
-			}
+		for _, plugin := range enabledPlugins {
+			wg.Add(1)
+			go func(plugin *model.BundleInfo) {
+				defer wg.Done()
 
-			// Determine if plugin is enabled
-			pluginId := plugin.Manifest.Id
-			pluginEnabled := false
-			if state, ok := config.PluginStates[pluginId]; ok {
-				pluginEnabled = state.Enable
-			}
-
-			// Activate plugin if enabled
-			if pluginEnabled {
+				pluginId := plugin.Manifest.Id
 				updatedManifest, activated, err := pluginsEnvironment.Activate(pluginId)
 				if err != nil {
 					plugin.WrapLogger(a.Log()).Error("Unable to activate plugin", mlog.Err(err))
-					continue
+					return
 				}
 
 				if activated {
@@ -134,8 +140,9 @@ func (a *App) SyncPluginsActiveState() {
 						a.Log().Error("Failed to notify cluster on plugin enable", mlog.Err(err))
 					}
 				}
-			}
+			}(plugin)
 		}
+		wg.Wait()
 	} else { // If plugins are disabled, shutdown plugins.
 		pluginsEnvironment.Shutdown()
 	}

app/plugin_test.go

@@ -629,6 +629,65 @@ func TestPluginSync(t *testing.T) {
 	}
 }
 
+func TestSyncPluginsActiveState(t *testing.T) {
+	th := Setup(t)
+	defer th.TearDown()
+
+	th.App.UpdateConfig(func(cfg *model.Config) {
+		*cfg.PluginSettings.Enable = true
+	})
+
+	env := th.App.GetPluginsEnvironment()
+	require.NotNil(t, env)
+
+	th.App.UpdateConfig(func(cfg *model.Config) {
+		*cfg.PluginSettings.RequirePluginSignature = false
+	})
+
+	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)
+
+	// Sync with file store so the plugin environment has access to this plugin.
+	appErr = th.App.SyncPlugins()
+	checkNoError(t, appErr)
+
+	// Verify the plugin was installed and set to deactivated.
+	pluginStatus, err := env.Statuses()
+	require.Nil(t, err)
+	require.Len(t, pluginStatus, 1)
+	require.Equal(t, pluginStatus[0].PluginId, "testplugin")
+	require.Equal(t, pluginStatus[0].State, model.PluginStateNotRunning)
+
+	// Enable plugin by setting setting config. This implicitly calls SyncPluginsActiveState through a config listener.
+	th.App.UpdateConfig(func(cfg *model.Config) {
+		cfg.PluginSettings.PluginStates["testplugin"] = &model.PluginState{Enable: true}
+	})
+
+	// Verify the plugin was activated due to config change.
+	pluginStatus, err = env.Statuses()
+	require.Nil(t, err)
+	require.Len(t, pluginStatus, 1)
+	require.Equal(t, pluginStatus[0].PluginId, "testplugin")
+	require.Equal(t, pluginStatus[0].State, model.PluginStateRunning)
+
+	// Disable plugin by setting config. This implicitly calls SyncPluginsActiveState through a config listener.
+	th.App.UpdateConfig(func(cfg *model.Config) {
+		cfg.PluginSettings.PluginStates["testplugin"] = &model.PluginState{Enable: false}
+	})
+
+	// Verify the plugin was deactivated due to config change.
+	pluginStatus, err = env.Statuses()
+	require.Nil(t, err)
+	require.Len(t, pluginStatus, 1)
+	require.Equal(t, pluginStatus[0].PluginId, "testplugin")
+	require.Equal(t, pluginStatus[0].State, model.PluginStateNotRunning)
+}
+
 func TestPluginPanicLogs(t *testing.T) {
 	t.Run("should panic", func(t *testing.T) {
 		th := Setup(t).InitBasic()