MM-8607: Add ability to turn off non-critical services when under load; cluster support (#13267)

* MM-8607: Add cluster support for server busy status

MM-8607: Busy.Cluster should be non-exported

MM-8607: rename logging field seconds -> expires_sec

MM-8607: each node clears its own busy flag

MM-8607: ensure unit tests are not sensitive to test machine speed

* MM-8607: chg comment to force CI rebuild

api4/system.go

@@ -492,12 +492,5 @@ func getServerBusyExpires(c *Context, w http.ResponseWriter, r *http.Request) {
 		c.SetPermissionError(model.PERMISSION_MANAGE_SYSTEM)
 		return
 	}
-
+	w.Write([]byte(c.App.Srv.Busy.ToJson()))
-	busy := c.App.Srv.Busy
-	sbs := &model.ServerBusyState{
-		Busy:       busy.IsBusy(),
-		Expires:    busy.Expires().Unix(),
-		Expires_ts: busy.Expires().UTC().Format("Mon Jan 2 15:04:05 -0700 MST 2006"),
-	}
-	w.Write([]byte(sbs.ToJson()))
 }

app/admin.go

@@ -218,3 +218,13 @@ func (a *App) TestEmail(userId string, cfg *model.Config) *model.AppError {
 
 	return nil
 }
+
+// ServerBusyStateChanged is called when a CLUSTER_EVENT_BUSY_STATE_CHANGED is received.
+func (a *App) ServerBusyStateChanged(sbs *model.ServerBusyState) {
+	a.Srv.Busy.ClusterEventChanged(sbs)
+	if sbs.Busy {
+		mlog.Warn("server busy state activitated via cluster event - non-critical services disabled", mlog.Int64("expires_sec", sbs.Expires))
+	} else {
+		mlog.Info("server busy state cleared via cluster event - non-critical services enabled")
+	}
+}

app/busy.go

@@ -7,14 +7,31 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/mattermost/mattermost-server/v5/einterfaces"
+	"github.com/mattermost/mattermost-server/v5/model"
 )
 
+const (
+	TIMESTAMP_FORMAT = "Mon Jan 2 15:04:05 -0700 MST 2006"
+)
+
+// Busy represents the busy state of the server. A server marked busy
+// will have non-critical services disabled. If a Cluster is provided
+// any changes will be propagated to each node.
 type Busy struct {
 	busy    int32 // protected via atomic for fast IsBusy calls
-
 	mux     sync.RWMutex
 	timer   *time.Timer
 	expires time.Time
+
+	cluster einterfaces.ClusterInterface
+}
+
+// NewBusy creates a new Busy instance with optional cluster which will
+// be notified of busy state changes.
+func NewBusy(cluster einterfaces.ClusterInterface) *Busy {
+	return &Busy{cluster: cluster}
 }
 
 // IsBusy returns true if the server has been marked as busy.
@@ -25,27 +42,47 @@ func (b *Busy) IsBusy() bool {
 	return atomic.LoadInt32(&b.busy) != 0
 }
 
-// Set marks the server as busy for dur duration.
+// Set marks the server as busy for dur duration and notifies cluster nodes.
 func (b *Busy) Set(dur time.Duration) {
 	b.mux.Lock()
 	defer b.mux.Unlock()
 
-	b.clear()
+	// minimum 1 second
-	atomic.StoreInt32(&b.busy, 1)
+	if dur < (time.Second * 1) {
+		dur = time.Second * 1
+	}
 
-	b.timer = time.AfterFunc(dur, b.Clear)
+	b.setWithoutNotify(dur)
-	b.expires = time.Now().Add(dur)
-}
 
-// ClearBusy marks the server as not busy.
+	if b.cluster != nil {
-func (b *Busy) Clear() {
+		sbs := &model.ServerBusyState{Busy: true, Expires: b.expires.Unix(), Expires_ts: b.expires.UTC().Format(TIMESTAMP_FORMAT)}
-	b.mux.Lock()
+		b.notifyServerBusyChange(sbs)
-	defer b.mux.Unlock()
+	}
-	b.clear()
 }
 
 // must hold mutex
-func (b *Busy) clear() {
+func (b *Busy) setWithoutNotify(dur time.Duration) {
+	b.clearWithoutNotify()
+	atomic.StoreInt32(&b.busy, 1)
+	b.expires = time.Now().Add(dur)
+	b.timer = time.AfterFunc(dur, b.clearWithoutNotify)
+}
+
+// ClearBusy marks the server as not busy and notifies cluster nodes.
+func (b *Busy) Clear() {
+	b.mux.Lock()
+	defer b.mux.Unlock()
+
+	b.clearWithoutNotify()
+
+	if b.cluster != nil {
+		sbs := &model.ServerBusyState{Busy: false, Expires: time.Time{}.Unix(), Expires_ts: ""}
+		b.notifyServerBusyChange(sbs)
+	}
+}
+
+// must hold mutex
+func (b *Busy) clearWithoutNotify() {
 	if b.timer != nil {
 		b.timer.Stop() // don't drain timer.C channel for AfterFunc timers.
 	}
@@ -62,3 +99,45 @@ func (b *Busy) Expires() time.Time {
 	defer b.mux.RUnlock()
 	return b.expires
 }
+
+// notifyServerBusyChange informs all cluster members of a server busy state change.
+func (b *Busy) notifyServerBusyChange(sbs *model.ServerBusyState) {
+	if b.cluster == nil {
+		return
+	}
+	msg := &model.ClusterMessage{
+		Event:            model.CLUSTER_EVENT_BUSY_STATE_CHANGED,
+		SendType:         model.CLUSTER_SEND_RELIABLE,
+		WaitForAllToSend: true,
+		Data:             sbs.ToJson(),
+	}
+	b.cluster.SendClusterMessage(msg)
+}
+
+// ClusterEventChanged is called when a CLUSTER_EVENT_BUSY_STATE_CHANGED is received.
+func (b *Busy) ClusterEventChanged(sbs *model.ServerBusyState) {
+	b.mux.Lock()
+	defer b.mux.Unlock()
+
+	if sbs.Busy {
+		expires := time.Unix(sbs.Expires, 0)
+		dur := time.Until(expires)
+		if dur > 0 {
+			b.setWithoutNotify(dur)
+		}
+	} else {
+		b.clearWithoutNotify()
+	}
+}
+
+func (b *Busy) ToJson() string {
+	b.mux.RLock()
+	defer b.mux.RUnlock()
+
+	sbs := &model.ServerBusyState{
+		Busy:       atomic.LoadInt32(&b.busy) != 0,
+		Expires:    b.expires.Unix(),
+		Expires_ts: b.expires.UTC().Format(TIMESTAMP_FORMAT),
+	}
+	return sbs.ToJson()
+}

app/busy_test.go

@@ -4,14 +4,18 @@
 package app
 
 import (
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/mattermost/mattermost-server/v5/einterfaces"
+	"github.com/mattermost/mattermost-server/v5/model"
 	"github.com/stretchr/testify/require"
 )
 
 func TestBusySet(t *testing.T) {
-	busy := &Busy{}
+	cluster := &ClusterMock{Busy: &Busy{}}
+	busy := NewBusy(cluster)
 
 	isNotBusy := func() bool {
 		return !busy.IsBusy()
@@ -19,29 +23,36 @@ func TestBusySet(t *testing.T) {
 
 	require.False(t, busy.IsBusy())
 
-	busy.Set(time.Millisecond * 100)
+	busy.Set(time.Second * 3)
 	require.True(t, busy.IsBusy())
-	// should automatically expire after 100ms
+	require.True(t, compareBusyState(t, busy, cluster.Busy))
-	require.Eventually(t, isNotBusy, time.Second*5, time.Millisecond*20)
+	// should automatically expire after 3s.
+	require.Eventually(t, isNotBusy, time.Second*15, time.Millisecond*20)
+	// allow a moment for cluster to sync.
+	require.Eventually(t, func() bool { return compareBusyState(t, busy, cluster.Busy) }, time.Second*15, time.Millisecond*20)
 
-	// test set after auto expiry
+	// test set after auto expiry.
 	busy.Set(time.Second * 30)
 	require.True(t, busy.IsBusy())
+	require.True(t, compareBusyState(t, busy, cluster.Busy))
 	expire := busy.Expires()
 	require.Greater(t, expire.Unix(), time.Now().Add(time.Second*10).Unix())
 
 	// test extending existing expiry
 	busy.Set(time.Minute * 5)
 	require.True(t, busy.IsBusy())
+	require.True(t, compareBusyState(t, busy, cluster.Busy))
 	expire = busy.Expires()
 	require.Greater(t, expire.Unix(), time.Now().Add(time.Minute*2).Unix())
 
 	busy.Clear()
 	require.False(t, busy.IsBusy())
+	require.True(t, compareBusyState(t, busy, cluster.Busy))
 }
 
 func TestBusyExpires(t *testing.T) {
-	busy := &Busy{}
+	cluster := &ClusterMock{Busy: &Busy{}}
+	busy := NewBusy(cluster)
 
 	isNotBusy := func() bool {
 		return !busy.IsBusy()
@@ -56,12 +67,14 @@ func TestBusyExpires(t *testing.T) {
 	busy.Set(time.Minute * 5)
 	expire = busy.Expires()
 	require.Greater(t, expire.Unix(), time.Now().Add(time.Minute*2).Unix())
+	require.True(t, compareBusyState(t, busy, cluster.Busy))
 
 	// get expiry after clear
 	busy.Clear()
 	expire = busy.Expires()
 	// should be time.Time zero value
 	require.Equal(t, time.Time{}.Unix(), expire.Unix())
+	require.True(t, compareBusyState(t, busy, cluster.Busy))
 
 	// get expiry after auto-expire
 	busy.Set(time.Millisecond * 100)
@@ -69,4 +82,45 @@ func TestBusyExpires(t *testing.T) {
 	expire = busy.Expires()
 	// should be time.Time zero value
 	require.Equal(t, time.Time{}.Unix(), expire.Unix())
+	// allow a moment for cluster to sync
+	require.Eventually(t, func() bool { return compareBusyState(t, busy, cluster.Busy) }, time.Second*15, time.Millisecond*20)
+}
+
+func compareBusyState(t *testing.T, busy1 *Busy, busy2 *Busy) bool {
+	t.Helper()
+	if busy1.IsBusy() != busy2.IsBusy() {
+		t.Logf("busy1:%s;  busy2:%s\n", busy1.ToJson(), busy2.ToJson())
+		return false
+	}
+	if busy1.Expires().Unix() != busy2.Expires().Unix() {
+		t.Logf("busy1:%s;  busy2:%s\n", busy1.ToJson(), busy2.ToJson())
+		return false
+	}
+	return true
+}
+
+// ClusterMock simulates the busy state of a cluster.
+type ClusterMock struct {
+	Busy *Busy
+}
+
+func (c *ClusterMock) SendClusterMessage(msg *model.ClusterMessage) {
+	sbs := model.ServerBusyStateFromJson(strings.NewReader(msg.Data))
+	c.Busy.ClusterEventChanged(sbs)
+}
+
+func (c *ClusterMock) StartInterNodeCommunication() {}
+func (c *ClusterMock) StopInterNodeCommunication()  {}
+func (c *ClusterMock) RegisterClusterMessageHandler(event string, crm einterfaces.ClusterMessageHandler) {
+}
+func (c *ClusterMock) GetClusterId() string                                       { return "cluster_mock" }
+func (c *ClusterMock) IsLeader() bool                                             { return false }
+func (c *ClusterMock) GetMyClusterInfo() *model.ClusterInfo                       { return nil }
+func (c *ClusterMock) GetClusterInfos() []*model.ClusterInfo                      { return nil }
+func (c *ClusterMock) NotifyMsg(buf []byte)                                       {}
+func (c *ClusterMock) GetClusterStats() ([]*model.ClusterStats, *model.AppError)  { return nil, nil }
+func (c *ClusterMock) GetLogs(page, perPage int) ([]string, *model.AppError)      { return nil, nil }
+func (c *ClusterMock) GetPluginStatuses() (model.PluginStatuses, *model.AppError) { return nil, nil }
+func (c *ClusterMock) ConfigChanged(previousConfig *model.Config, newConfig *model.Config, sendToOtherServer bool) *model.AppError {
+	return nil
 }

app/cluster_handlers.go

@@ -25,6 +25,7 @@ func (a *App) RegisterAllClusterMessageHandlers() {
 	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)
+	a.Cluster.RegisterClusterMessageHandler(model.CLUSTER_EVENT_BUSY_STATE_CHANGED, a.ClusterBusyStateChgHandler)
 
 }
 
@@ -89,3 +90,7 @@ func (a *App) ClusterInstallPluginHandler(msg *model.ClusterMessage) {
 func (a *App) ClusterRemovePluginHandler(msg *model.ClusterMessage) {
 	a.RemovePluginFromData(model.PluginEventDataFromJson(strings.NewReader(msg.Data)))
 }
+
+func (a *App) ClusterBusyStateChgHandler(msg *model.ClusterMessage) {
+	a.ServerBusyStateChanged(model.ServerBusyStateFromJson(strings.NewReader(msg.Data)))
+}

app/server.go

@@ -312,7 +312,7 @@ func NewServer(options ...Option) (*Server, error) {
 	return s, nil
 }
 
-// Global app opptions that should be applied to apps created by this server
+// Global app options that should be applied to apps created by this server
 func (s *Server) AppOptions() []AppOption {
 	return []AppOption{
 		ServerConnector(s),
@@ -474,7 +474,7 @@ func (s *Server) Start() error {
 		s.RateLimiter = rateLimiter
 		handler = rateLimiter.RateLimitHandler(handler)
 	}
-	s.Busy = &Busy{}
+	s.Busy = NewBusy(s.Cluster)
 
 	// Creating a logger for logging errors from http.Server at error level
 	errStdLog, err := s.Log.StdLogAt(mlog.LevelError, mlog.String("source", "httpserver"))

model/cluster_message.go

@@ -36,6 +36,7 @@ const (
 	CLUSTER_EVENT_CLEAR_SESSION_CACHE_FOR_ALL_USERS                 = "inv_all_user_sessions"
 	CLUSTER_EVENT_INSTALL_PLUGIN                                    = "install_plugin"
 	CLUSTER_EVENT_REMOVE_PLUGIN                                     = "remove_plugin"
+	CLUSTER_EVENT_BUSY_STATE_CHANGED                                = "busy_state_change"
 
 	// SendTypes for ClusterMessage.
 	CLUSTER_SEND_BEST_EFFORT = "best_effort"

model/system.go

@@ -51,6 +51,7 @@ type SystemECDSAKey struct {
 	D     *big.Int `json:"d,omitempty"`
 }
 
+// ServerBusyState provides serialization for app.Busy.
 type ServerBusyState struct {
 	Busy       bool   `json:"busy"`
 	Expires    int64  `json:"expires"`