[MM-15831] Improve system for storing status of available plug… (#11185)

* Move State property from activePlugin to PluginHealthStatus. env.activePlugins is now reserved for healthy running plugins.

* Add comments for function declarations

* Combine activePlugins and pluginHealthStatuses into a common structure, registeredPlugins

* Add check to see if plugin is active before deactivating it

* Make `Deactivate` set plugin status

* Add comment explaining the `registeredPlugins` map
* Give responsibility to set plugin disabled status upon deactivation back to `env.Deactivate`

* check if plugin needs to be deactivated before setting status
Этот коммит содержится в:
Michael Kochell
2019-06-25 17:44:08 -04:00
коммит произвёл GitHub
родитель 332b53a30d
Коммит b68194e035
5 изменённых файлов: 124 добавлений и 115 удалений

Просмотреть файл

@@ -130,6 +130,7 @@ func (a *App) removePlugin(id string) *model.AppError {
}
pluginsEnvironment.Deactivate(id)
pluginsEnvironment.RemovePlugin(id)
a.UnregisterPluginCommands(id)
err = os.RemoveAll(pluginPath)

Просмотреть файл

@@ -13,7 +13,7 @@ const (
PluginStateStarting = 1 // unused by server
PluginStateRunning = 2
PluginStateFailedToStart = 3
PluginStateFailedToStayRunning = 4 // unused by server
PluginStateFailedToStayRunning = 4
PluginStateStopping = 5 // unused by server
)

Просмотреть файл

@@ -10,6 +10,7 @@ import (
"os"
"path/filepath"
"sync"
"time"
"github.com/mattermost/mattermost-server/mlog"
"github.com/mattermost/mattermost-server/model"
@@ -19,11 +20,18 @@ import (
type apiImplCreatorFunc func(*model.Manifest) API
type activePlugin struct {
// registeredPlugin stores the state for a given plugin that has been activated
// or attempted to be activated this server run.
//
// If an installed plugin is missing from the env.registeredPlugins map, then the
// plugin is configured as disabled and has not been activated during this server run.
type registeredPlugin struct {
BundleInfo *model.BundleInfo
State int
State *int
supervisor *supervisor
failTimeStamps []time.Time
lastError error
supervisor *supervisor
}
// Environment represents the execution environment of active plugins.
@@ -31,8 +39,7 @@ type activePlugin struct {
// It is meant for use by the Mattermost server to manipulate, interact with and report on the set
// of active plugins.
type Environment struct {
activePlugins sync.Map
pluginHealthStatuses sync.Map
registeredPlugins sync.Map
pluginHealthCheckJob *PluginHealthCheckJob
logger *mlog.Logger
newAPIImpl apiImplCreatorFunc
@@ -81,9 +88,9 @@ func (env *Environment) Available() ([]*model.BundleInfo, error) {
// Returns a list of all currently active plugins within the environment.
func (env *Environment) Active() []*model.BundleInfo {
activePlugins := []*model.BundleInfo{}
env.activePlugins.Range(func(key, value interface{}) bool {
plugin := value.(activePlugin)
if plugin.State == model.PluginStateRunning {
env.registeredPlugins.Range(func(key, value interface{}) bool {
plugin := value.(*registeredPlugin)
if env.IsActive(plugin.BundleInfo.Manifest.Id) {
activePlugins = append(activePlugins, plugin.BundleInfo)
}
@@ -95,14 +102,30 @@ func (env *Environment) Active() []*model.BundleInfo {
// IsActive returns true if the plugin with the given id is active.
func (env *Environment) IsActive(id string) bool {
_, ok := env.activePlugins.Load(id)
return ok
return env.GetPluginState(id) == model.PluginStateRunning
}
// GetPluginState returns the current state of a plugin (disabled, running, or error)
func (env *Environment) GetPluginState(id string) int {
rp, ok := env.registeredPlugins.Load(id)
if !ok {
return model.PluginStateNotRunning
}
return *rp.(*registeredPlugin).State
}
// SetPluginState sets the current state of a plugin (disabled, running, or error)
func (env *Environment) SetPluginState(id string, state int) {
if rp, ok := env.registeredPlugins.Load(id); ok {
*rp.(*registeredPlugin).State = state
}
}
// PublicFilesPath returns a path and true if the plugin with the given id is active.
// It returns an empty string and false if the path is not set or invalid
func (env *Environment) PublicFilesPath(id string) (string, error) {
if _, ok := env.activePlugins.Load(id); !ok {
if _, ok := env.registeredPlugins.Load(id); !ok {
return "", fmt.Errorf("plugin not found: %v", id)
}
return filepath.Join(env.pluginDir, id, "public"), nil
@@ -122,10 +145,7 @@ func (env *Environment) Statuses() (model.PluginStatuses, error) {
continue
}
pluginState := model.PluginStateNotRunning
if plugin, ok := env.activePlugins.Load(plugin.Manifest.Id); ok {
pluginState = plugin.(activePlugin).State
}
pluginState := env.GetPluginState(plugin.Manifest.Id)
status := &model.PluginStatus{
PluginId: plugin.Manifest.Id,
@@ -144,7 +164,7 @@ func (env *Environment) Statuses() (model.PluginStatuses, error) {
func (env *Environment) Activate(id string) (manifest *model.Manifest, activated bool, reterr error) {
// Check if we are already active
if _, ok := env.activePlugins.Load(id); ok {
if env.IsActive(id) {
return nil, false, nil
}
@@ -165,14 +185,23 @@ func (env *Environment) Activate(id string) (manifest *model.Manifest, activated
return nil, false, fmt.Errorf("plugin not found: %v", id)
}
ap := activePlugin{BundleInfo: pluginInfo}
value, ok := env.registeredPlugins.Load(id)
if !ok {
value = newRegisteredPlugin(pluginInfo)
env.registeredPlugins.Store(id, value)
}
rp := value.(*registeredPlugin)
// Store latest BundleInfo in case something has changed since last activation
rp.BundleInfo = pluginInfo
defer func() {
if reterr == nil {
ap.State = model.PluginStateRunning
env.SetPluginState(id, model.PluginStateRunning)
} else {
ap.State = model.PluginStateFailedToStart
env.SetPluginState(id, model.PluginStateFailedToStart)
}
env.activePlugins.Store(pluginInfo.Manifest.Id, ap)
}()
if pluginInfo.Manifest.MinServerVersion != "" {
@@ -229,18 +258,9 @@ func (env *Environment) Activate(id string) (manifest *model.Manifest, activated
if err != nil {
return nil, false, errors.Wrapf(err, "unable to start plugin: %v", id)
}
ap.supervisor = sup
rp.supervisor = sup
componentActivated = true
var h *PluginHealthStatus
if health, ok := env.pluginHealthStatuses.Load(id); ok {
h = health.(*PluginHealthStatus)
} else {
h = newPluginHealthStatus()
env.pluginHealthStatuses.Store(id, h)
}
h.Crashed = false
}
if !componentActivated {
@@ -250,21 +270,33 @@ func (env *Environment) Activate(id string) (manifest *model.Manifest, activated
return pluginInfo.Manifest, true, nil
}
func (env *Environment) RemovePlugin(id string) {
if _, ok := env.registeredPlugins.Load(id); ok {
env.registeredPlugins.Delete(id)
}
}
// Deactivates the plugin with the given id.
func (env *Environment) Deactivate(id string) bool {
p, ok := env.activePlugins.Load(id)
p, ok := env.registeredPlugins.Load(id)
if !ok {
return false
}
env.activePlugins.Delete(id)
isActive := env.IsActive(id)
ap := p.(activePlugin)
if ap.supervisor != nil {
if err := ap.supervisor.Hooks().OnDeactivate(); err != nil {
env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", ap.BundleInfo.Manifest.Id), mlog.Err(err))
env.SetPluginState(id, model.PluginStateNotRunning)
if !isActive {
return false
}
rp := p.(*registeredPlugin)
if rp.supervisor != nil {
if err := rp.supervisor.Hooks().OnDeactivate(); err != nil {
env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", rp.BundleInfo.Manifest.Id), mlog.Err(err))
}
ap.supervisor.Shutdown()
rp.supervisor.Shutdown()
}
return true
@@ -277,36 +309,19 @@ func (env *Environment) RestartPlugin(id string) error {
return err
}
// UpdatePluginHealthStatus accepts a callback to edit the stored health status of the plugin.
func (env *Environment) UpdatePluginHealthStatus(id string, callback func(*PluginHealthStatus)) {
if h, ok := env.pluginHealthStatuses.Load(id); ok {
callback(h.(*PluginHealthStatus))
}
}
// CheckPluginHealthStatus checks if the plugin is in a failed state, based on information gathered from previous health checks.
func (env *Environment) CheckPluginHealthStatus(id string) error {
if h, ok := env.pluginHealthStatuses.Load(id); ok {
if h.(*PluginHealthStatus).Crashed {
return h.(*PluginHealthStatus).lastError
}
}
return nil
}
// Shutdown deactivates all plugins and gracefully shuts down the environment.
func (env *Environment) Shutdown() {
env.activePlugins.Range(func(key, value interface{}) bool {
ap := value.(activePlugin)
env.registeredPlugins.Range(func(key, value interface{}) bool {
rp := value.(*registeredPlugin)
if ap.supervisor != nil {
if err := ap.supervisor.Hooks().OnDeactivate(); err != nil {
env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", ap.BundleInfo.Manifest.Id), mlog.Err(err))
if rp.supervisor != nil {
if err := rp.supervisor.Hooks().OnDeactivate(); err != nil {
env.logger.Error("Plugin OnDeactivate() error", mlog.String("plugin_id", rp.BundleInfo.Manifest.Id), mlog.Err(err))
}
ap.supervisor.Shutdown()
rp.supervisor.Shutdown()
}
env.activePlugins.Delete(key)
env.registeredPlugins.Delete(key)
return true
})
@@ -316,10 +331,10 @@ func (env *Environment) Shutdown() {
//
// Consider using RunMultiPluginHook instead.
func (env *Environment) HooksForPlugin(id string) (Hooks, error) {
if p, ok := env.activePlugins.Load(id); ok {
ap := p.(activePlugin)
if ap.supervisor != nil {
return ap.supervisor.Hooks(), nil
if p, ok := env.registeredPlugins.Load(id); ok {
rp := p.(*registeredPlugin)
if rp.supervisor != nil {
return rp.supervisor.Hooks(), nil
}
}
@@ -331,16 +346,21 @@ func (env *Environment) HooksForPlugin(id string) (Hooks, error) {
// If hookRunnerFunc returns false, iteration will not continue. The iteration order among active
// plugins is not specified.
func (env *Environment) RunMultiPluginHook(hookRunnerFunc func(hooks Hooks) bool, hookId int) {
env.activePlugins.Range(func(key, value interface{}) bool {
ap := value.(activePlugin)
env.registeredPlugins.Range(func(key, value interface{}) bool {
rp := value.(*registeredPlugin)
if ap.supervisor == nil || !ap.supervisor.Implements(hookId) {
if rp.supervisor == nil || !rp.supervisor.Implements(hookId) {
return true
}
if !hookRunnerFunc(ap.supervisor.Hooks()) {
if !hookRunnerFunc(rp.supervisor.Hooks()) {
return false
}
return true
})
}
func newRegisteredPlugin(bundle *model.BundleInfo) *registeredPlugin {
state := model.PluginStateNotRunning
return &registeredPlugin{failTimeStamps: []time.Time{}, State: &state, BundleInfo: bundle}
}

Просмотреть файл

@@ -8,6 +8,7 @@ import (
"time"
"github.com/mattermost/mattermost-server/mlog"
"github.com/mattermost/mattermost-server/model"
)
const (
@@ -23,12 +24,6 @@ type PluginHealthCheckJob struct {
env *Environment
}
type PluginHealthStatus struct {
Crashed bool
failTimeStamps []time.Time
lastError error
}
// InitPluginHealthCheckJob starts a new job for checking all active plugins
func (env *Environment) InitPluginHealthCheckJob() {
job := newPluginHealthCheckJob(env)
@@ -65,17 +60,13 @@ func (job *PluginHealthCheckJob) Start() {
// checkPlugin determines the plugin's health status, then handles the error or success case.
func (job *PluginHealthCheckJob) checkPlugin(id string) {
p, ok := job.env.activePlugins.Load(id)
p, ok := job.env.registeredPlugins.Load(id)
if !ok {
return
}
ap := p.(activePlugin)
rp := p.(*registeredPlugin)
if _, ok := job.env.pluginHealthStatuses.Load(id); !ok {
job.env.pluginHealthStatuses.Store(id, newPluginHealthStatus())
}
sup := ap.supervisor
sup := rp.supervisor
if sup == nil {
return
}
@@ -90,21 +81,21 @@ func (job *PluginHealthCheckJob) checkPlugin(id string) {
// handleHealthCheckFail restarts or deactivates the plugin based on how many times it has failed in a configured amount of time.
func (job *PluginHealthCheckJob) handleHealthCheckFail(id string, err error) {
health, ok := job.env.pluginHealthStatuses.Load(id)
rp, ok := job.env.registeredPlugins.Load(id)
if !ok {
return
}
h := health.(*PluginHealthStatus)
p := rp.(*registeredPlugin)
// Append current failure before checking for deactivate vs restart action
h.failTimeStamps = append(h.failTimeStamps, time.Now())
h.lastError = err
p.failTimeStamps = append(p.failTimeStamps, time.Now())
p.lastError = err
if shouldDeactivatePlugin(h) {
h.failTimeStamps = []time.Time{}
h.Crashed = true
if shouldDeactivatePlugin(p) {
p.failTimeStamps = []time.Time{}
mlog.Debug(fmt.Sprintf("Deactivating plugin due to multiple crashes `%s`", id))
job.env.Deactivate(id)
job.env.SetPluginState(id, model.PluginStateFailedToStayRunning)
} else {
mlog.Debug(fmt.Sprintf("Restarting plugin due to failed health check `%s`", id))
if err := job.env.RestartPlugin(id); err != nil {
@@ -126,17 +117,13 @@ func (job *PluginHealthCheckJob) Cancel() {
<-job.cancelled
}
func newPluginHealthStatus() *PluginHealthStatus {
return &PluginHealthStatus{failTimeStamps: []time.Time{}, Crashed: false}
}
// shouldDeactivatePlugin determines if a plugin needs to be deactivated after certain criteria is met.
//
// The criteria is based on if the plugin has consistently failed during the configured number of restarts, within the configured time window.
func shouldDeactivatePlugin(h *PluginHealthStatus) bool {
if len(h.failTimeStamps) >= HEALTH_CHECK_RESTART_LIMIT {
index := len(h.failTimeStamps) - HEALTH_CHECK_RESTART_LIMIT
t := h.failTimeStamps[index]
func shouldDeactivatePlugin(rp *registeredPlugin) bool {
if len(rp.failTimeStamps) >= HEALTH_CHECK_RESTART_LIMIT {
index := len(rp.failTimeStamps) - HEALTH_CHECK_RESTART_LIMIT
t := rp.failTimeStamps[index]
now := time.Now()
elapsed := now.Sub(t).Minutes()
if elapsed <= HEALTH_CHECK_DISABLE_DURATION {

Просмотреть файл

@@ -118,38 +118,39 @@ func testPluginHealthCheck_Panic(t *testing.T) {
}
func TestShouldDeactivatePlugin(t *testing.T) {
h := newPluginHealthStatus()
require.NotNil(t, h)
bundle := &model.BundleInfo{}
rp := newRegisteredPlugin(bundle)
require.NotNil(t, rp)
// No failures, don't restart
result := shouldDeactivatePlugin(h)
result := shouldDeactivatePlugin(rp)
require.Equal(t, false, result)
now := time.Now()
// Failures are recent enough to restart
h = newPluginHealthStatus()
h.failTimeStamps = append(h.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*0.2*time.Minute))
h.failTimeStamps = append(h.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*0.1*time.Minute))
h.failTimeStamps = append(h.failTimeStamps, now)
rp = newRegisteredPlugin(bundle)
rp.failTimeStamps = append(rp.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*0.2*time.Minute))
rp.failTimeStamps = append(rp.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*0.1*time.Minute))
rp.failTimeStamps = append(rp.failTimeStamps, now)
result = shouldDeactivatePlugin(h)
result = shouldDeactivatePlugin(rp)
require.Equal(t, true, result)
// Failures are too spaced out to warrant a restart
h = newPluginHealthStatus()
h.failTimeStamps = append(h.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*2*time.Minute))
h.failTimeStamps = append(h.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*1*time.Minute))
h.failTimeStamps = append(h.failTimeStamps, now)
rp = newRegisteredPlugin(bundle)
rp.failTimeStamps = append(rp.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*2*time.Minute))
rp.failTimeStamps = append(rp.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*1*time.Minute))
rp.failTimeStamps = append(rp.failTimeStamps, now)
result = shouldDeactivatePlugin(h)
result = shouldDeactivatePlugin(rp)
require.Equal(t, false, result)
// Not enough failures are present to warrant a restart
h = newPluginHealthStatus()
h.failTimeStamps = append(h.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*0.1*time.Minute))
h.failTimeStamps = append(h.failTimeStamps, now)
rp = newRegisteredPlugin(bundle)
rp.failTimeStamps = append(rp.failTimeStamps, now.Add(-HEALTH_CHECK_DISABLE_DURATION*0.1*time.Minute))
rp.failTimeStamps = append(rp.failTimeStamps, now)
result = shouldDeactivatePlugin(h)
result = shouldDeactivatePlugin(rp)
require.Equal(t, false, result)
}