/
cluster_discovery.go
102 lines (84 loc) · 2.97 KB
/
cluster_discovery.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package app
import (
"time"
"github.com/mattermost/mattermost-server/v6/model"
"github.com/mattermost/mattermost-server/v6/shared/mlog"
)
const (
DiscoveryServiceWritePing = 60 * time.Second
)
type ClusterDiscoveryService struct {
model.ClusterDiscovery
srv *Server
stop chan bool
}
func (s *Server) NewClusterDiscoveryService() *ClusterDiscoveryService {
ds := &ClusterDiscoveryService{
ClusterDiscovery: model.ClusterDiscovery{},
srv: s,
stop: make(chan bool),
}
return ds
}
func (a *App) NewClusterDiscoveryService() *ClusterDiscoveryService {
return a.Srv().NewClusterDiscoveryService()
}
func (cds *ClusterDiscoveryService) Start() {
err := cds.srv.Store.ClusterDiscovery().Cleanup()
if err != nil {
mlog.Warn("ClusterDiscoveryService failed to cleanup the outdated cluster discovery information", mlog.Err(err))
}
exists, err := cds.srv.Store.ClusterDiscovery().Exists(&cds.ClusterDiscovery)
if err != nil {
mlog.Warn("ClusterDiscoveryService failed to check if row exists", mlog.String("ClusterDiscoveryID", cds.ClusterDiscovery.Id), mlog.Err(err))
} else if exists {
if _, err := cds.srv.Store.ClusterDiscovery().Delete(&cds.ClusterDiscovery); err != nil {
mlog.Warn("ClusterDiscoveryService failed to start clean", mlog.String("ClusterDiscoveryID", cds.ClusterDiscovery.Id), mlog.Err(err))
}
}
if err := cds.srv.Store.ClusterDiscovery().Save(&cds.ClusterDiscovery); err != nil {
mlog.Error("ClusterDiscoveryService failed to save", mlog.String("ClusterDiscoveryID", cds.ClusterDiscovery.Id), mlog.Err(err))
return
}
go func() {
mlog.Debug("ClusterDiscoveryService ping writer started", mlog.String("ClusterDiscoveryID", cds.ClusterDiscovery.Id))
ticker := time.NewTicker(DiscoveryServiceWritePing)
defer func() {
ticker.Stop()
if _, err := cds.srv.Store.ClusterDiscovery().Delete(&cds.ClusterDiscovery); err != nil {
mlog.Warn("ClusterDiscoveryService failed to cleanup", mlog.String("ClusterDiscoveryID", cds.ClusterDiscovery.Id), mlog.Err(err))
}
mlog.Debug("ClusterDiscoveryService ping writer stopped", mlog.String("ClusterDiscoveryID", cds.ClusterDiscovery.Id))
}()
for {
select {
case <-ticker.C:
if err := cds.srv.Store.ClusterDiscovery().SetLastPingAt(&cds.ClusterDiscovery); err != nil {
mlog.Error("ClusterDiscoveryService failed to write ping", mlog.String("ClusterDiscoveryID", cds.ClusterDiscovery.Id), mlog.Err(err))
}
case <-cds.stop:
return
}
}
}()
}
func (cds *ClusterDiscoveryService) Stop() {
cds.stop <- true
}
func (s *Server) IsLeader() bool {
if s.License() != nil && *s.Config().ClusterSettings.Enable && s.Cluster != nil {
return s.Cluster.IsLeader()
}
return true
}
func (a *App) IsLeader() bool {
return a.Srv().IsLeader()
}
func (a *App) GetClusterId() string {
if a.Cluster() == nil {
return ""
}
return a.Cluster().GetClusterId()
}