From ae76d27b7dbb8f20b35f17153fe77beb8c0d2297 Mon Sep 17 00:00:00 2001 From: Christopher Speller Date: Tue, 15 Jan 2019 09:09:25 -0800 Subject: [PATCH] Migrate cluster to use Server struct directly. (#10101) --- app/cluster.go | 17 ++++++++--------- app/enterprise.go | 6 +++--- app/options.go | 12 ++++++++++++ app/server.go | 10 +++++++--- app/server_app_adapters.go | 4 ++-- cmd/mattermost/commands/server.go | 1 + 6 files changed, 33 insertions(+), 17 deletions(-) diff --git a/app/cluster.go b/app/cluster.go index e166521f16..54556f3c77 100644 --- a/app/cluster.go +++ b/app/cluster.go @@ -4,28 +4,27 @@ package app import ( - "github.com/mattermost/mattermost-server/mlog" "github.com/mattermost/mattermost-server/model" ) // Registers a given function to be called when the cluster leader may have changed. Returns a unique ID for the // listener which can later be used to remove it. If clustering is not enabled in this build, the callback will never // be called. -func (a *App) AddClusterLeaderChangedListener(listener func()) string { +func (s *Server) AddClusterLeaderChangedListener(listener func()) string { id := model.NewId() - a.Srv.clusterLeaderListeners.Store(id, listener) + s.clusterLeaderListeners.Store(id, listener) return id } // Removes a listener function by the unique ID returned when AddConfigListener was called -func (a *App) RemoveClusterLeaderChangedListener(id string) { - a.Srv.clusterLeaderListeners.Delete(id) +func (s *Server) RemoveClusterLeaderChangedListener(id string) { + s.clusterLeaderListeners.Delete(id) } -func (a *App) InvokeClusterLeaderChangedListeners() { - mlog.Info("Cluster leader changed. Invoking ClusterLeaderChanged listeners.") - a.Srv.Go(func() { - a.Srv.clusterLeaderListeners.Range(func(_, listener interface{}) bool { +func (s *Server) InvokeClusterLeaderChangedListeners() { + s.Log.Info("Cluster leader changed. Invoking ClusterLeaderChanged listeners.") + s.Go(func() { + s.clusterLeaderListeners.Range(func(_, listener interface{}) bool { listener.(func())() return true }) diff --git a/app/enterprise.go b/app/enterprise.go index 4b18e22780..7f3ab6382b 100644 --- a/app/enterprise.go +++ b/app/enterprise.go @@ -17,9 +17,9 @@ func RegisterAccountMigrationInterface(f func(*App) einterfaces.AccountMigration accountMigrationInterface = f } -var clusterInterface func(*App) einterfaces.ClusterInterface +var clusterInterface func(*Server) einterfaces.ClusterInterface -func RegisterClusterInterface(f func(*App) einterfaces.ClusterInterface) { +func RegisterClusterInterface(f func(*Server) einterfaces.ClusterInterface) { clusterInterface = f } @@ -135,6 +135,6 @@ func (s *Server) initEnterprise() { s.DataRetention = dataRetentionInterface(s.FakeApp()) } if clusterInterface != nil { - s.Cluster = clusterInterface(s.FakeApp()) + s.Cluster = clusterInterface(s) } } diff --git a/app/options.go b/app/options.go index 77fe99d640..4ba8b45377 100644 --- a/app/options.go +++ b/app/options.go @@ -40,6 +40,18 @@ func RunJobs(s *Server) { s.runjobs = true } +func JoinCluster(s *Server) { + s.joinCluster = true +} + +func StartMetrics(s *Server) { + s.startMetrics = true +} + +func StartElasticsearch(s *Server) { + s.startElasticsearch = true +} + func DisableConfigWatch(s *Server) { s.disableConfigWatch = true } diff --git a/app/server.go b/app/server.go index 4263e441f4..b0324fc349 100644 --- a/app/server.go +++ b/app/server.go @@ -112,6 +112,10 @@ type Server struct { Log *mlog.Logger + joinCluster bool + startMetrics bool + startElasticsearch bool + AccountMigration einterfaces.AccountMigrationInterface Cluster einterfaces.ClusterInterface Compliance einterfaces.ComplianceInterface @@ -214,16 +218,16 @@ func NewServer(options ...Option) (*Server, error) { mlog.Error(fmt.Sprint("Error to reset the server status.", result.Err.Error())) } - if s.Cluster != nil { + if s.joinCluster && s.Cluster != nil { s.FakeApp().RegisterAllClusterMessageHandlers() s.Cluster.StartInterNodeCommunication() } - if s.Metrics != nil { + if s.startMetrics && s.Metrics != nil { s.Metrics.StartServer() } - if s.Elasticsearch != nil { + if s.startElasticsearch && s.Elasticsearch != nil { s.StartElasticsearch() } diff --git a/app/server_app_adapters.go b/app/server_app_adapters.go index 86e93317e1..741895f7ac 100644 --- a/app/server_app_adapters.go +++ b/app/server_app_adapters.go @@ -86,7 +86,7 @@ func (s *Server) RunOldAppInitalization() error { a.EnsureDiagnosticId() a.regenerateClientConfig() - a.Srv.clusterLeaderListenerId = a.AddClusterLeaderChangedListener(func() { + a.Srv.clusterLeaderListenerId = a.Srv.AddClusterLeaderChangedListener(func() { mlog.Info("Cluster leader changed. Determining if job schedulers should be running:", mlog.Bool("isLeader", a.IsLeader())) if a.Srv.Jobs != nil { a.Srv.Jobs.Schedulers.HandleClusterLeaderChange(a.IsLeader()) @@ -156,7 +156,7 @@ func (s *Server) RunOldAppShutdown() { a.StopPushNotificationsHubWorkers() a.ShutDownPlugins() a.RemoveLicenseListener(s.licenseListenerId) - a.RemoveClusterLeaderChangedListener(s.clusterLeaderListenerId) + s.RemoveClusterLeaderChangedListener(s.clusterLeaderListenerId) } // A temporary bridge to deal with cases where the code is so tighly coupled that diff --git a/cmd/mattermost/commands/server.go b/cmd/mattermost/commands/server.go index 90246756d1..e4c123b5c8 100644 --- a/cmd/mattermost/commands/server.go +++ b/cmd/mattermost/commands/server.go @@ -47,6 +47,7 @@ func runServer(configFileLocation string, disableConfigWatch bool, usedPlatform options := []app.Option{ app.ConfigFile(configFileLocation), app.RunJobs, + app.JoinCluster, } if disableConfigWatch { options = append(options, app.DisableConfigWatch)