From 420c4115952eb04141f8282ccbf37cbce3270e1a Mon Sep 17 00:00:00 2001 From: Doug Lauder Date: Tue, 3 Dec 2019 12:56:40 -0500 Subject: [PATCH] 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 | 9 +--- app/admin.go | 10 ++++ app/busy.go | 107 ++++++++++++++++++++++++++++++++++----- app/busy_test.go | 66 +++++++++++++++++++++--- app/cluster_handlers.go | 5 ++ app/server.go | 4 +- model/cluster_message.go | 1 + model/system.go | 1 + 8 files changed, 173 insertions(+), 30 deletions(-) diff --git a/api4/system.go b/api4/system.go index c5a1ea5a48..a8fb5db077 100644 --- a/api4/system.go +++ b/api4/system.go @@ -492,12 +492,5 @@ func getServerBusyExpires(c *Context, w http.ResponseWriter, r *http.Request) { c.SetPermissionError(model.PERMISSION_MANAGE_SYSTEM) return } - - 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())) + w.Write([]byte(c.App.Srv.Busy.ToJson())) } diff --git a/app/admin.go b/app/admin.go index 3a69995da5..fa71a7180b 100644 --- a/app/admin.go +++ b/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") + } +} diff --git a/app/busy.go b/app/busy.go index a6dde358db..bb1c0ceee9 100644 --- a/app/busy.go +++ b/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" ) -type Busy struct { - busy int32 // protected via atomic for fast IsBusy calls +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() - atomic.StoreInt32(&b.busy, 1) + // minimum 1 second + if dur < (time.Second * 1) { + dur = time.Second * 1 + } - b.timer = time.AfterFunc(dur, b.Clear) - b.expires = time.Now().Add(dur) -} + b.setWithoutNotify(dur) -// ClearBusy marks the server as not busy. -func (b *Busy) Clear() { - b.mux.Lock() - defer b.mux.Unlock() - b.clear() + if b.cluster != nil { + sbs := &model.ServerBusyState{Busy: true, Expires: b.expires.Unix(), Expires_ts: b.expires.UTC().Format(TIMESTAMP_FORMAT)} + b.notifyServerBusyChange(sbs) + } } // 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() +} diff --git a/app/busy_test.go b/app/busy_test.go index b7f8a3aa79..fbed8cde1a 100644 --- a/app/busy_test.go +++ b/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.Eventually(t, isNotBusy, time.Second*5, time.Millisecond*20) + require.True(t, compareBusyState(t, busy, cluster.Busy)) + // 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 } diff --git a/app/cluster_handlers.go b/app/cluster_handlers.go index 71c2417afa..6b4acdc874 100644 --- a/app/cluster_handlers.go +++ b/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))) +} diff --git a/app/server.go b/app/server.go index f5185b5974..5fe54e217f 100644 --- a/app/server.go +++ b/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")) diff --git a/model/cluster_message.go b/model/cluster_message.go index 8432264e3f..d090284774 100644 --- a/model/cluster_message.go +++ b/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" diff --git a/model/system.go b/model/system.go index 412bc2be65..473292a876 100644 --- a/model/system.go +++ b/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"`