Add more mutex code

Author: svaroqui

cluster/cluster.go

@@ -546,6 +546,8 @@ func (cluster *Cluster) Run() {
 		go cluster.createKeys()
 	}
 
+	cluster.Topology = cluster.GetTopologyFromConf()
+
 	for cluster.exit == false {
 		if !cluster.Conf.MonitorPause {
 			cluster.ServerIdList = cluster.GetDBServerIdList()
@@ -653,7 +655,6 @@ func (cluster *Cluster) Run() {
 				// CheckFailed trigger failover code if passing all false positiv and constraints
 				cluster.CheckFailed()
 
-				cluster.Topology = cluster.GetTopology()
 				cluster.SetStatus()
 				cluster.StateProcessing()
 			}
@@ -813,9 +814,8 @@ func (cluster *Cluster) Save() error {
 	if err != nil {
 		return err
 	}
-	cluster.Lock()
+
 	saveAgents, _ := json.MarshalIndent(cluster.Agents, "", "\t")
-	cluster.Unlock()
 
 	err = ioutil.WriteFile(cluster.Conf.WorkingDir+"/"+cluster.Name+"/agents.json", saveAgents, 0644)
 	if err != nil {
@@ -1428,6 +1428,7 @@ func (cluster *Cluster) MonitorSchema() {
 			t.TableSync = oldtable.TableSync
 		}
 		// lookup other clusters
+
 		for _, cl := range cluster.clusterList {
 			if cl.GetName() != cluster.GetName() {
 

cluster/cluster_get.go

@@ -593,7 +593,8 @@ func (cluster *Cluster) GetProxyServerIdList() []string {
 	return ret
 }
 
-func (cluster *Cluster) GetTopology() string {
+func (cluster *Cluster) GetTopologyFromConf() string {
+
 	cluster.Conf.Topology = topoUnknown
 	if cluster.Conf.MultiMaster {
 		cluster.Conf.Topology = topoMultiMaster
@@ -626,6 +627,10 @@ func (cluster *Cluster) GetTopology() string {
 	return cluster.Conf.Topology
 }
 
+func (cluster *Cluster) GetTopology() string {
+	return cluster.Topology
+}
+
 /*
 	func (cluster *Cluster) GetDatabaseTags() []string {
 		return strings.Split(cluster.Conf.ProvTags, ",")

cluster/cluster_topo.go

@@ -75,7 +75,8 @@ func (cluster *Cluster) newServerList() error {
 func (cluster *Cluster) AddChildServers() error {
 
 	mychilds := cluster.GetChildClusters()
-
+	cluster.Lock()
+	defer cluster.Unlock()
 	for _, c := range mychilds {
 		for _, sv := range c.Servers {
 			cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModTopology, LvlDbg, "AddChildServers checking %s of %s ", sv.URL, c.Name)
@@ -126,6 +127,9 @@ func (cluster *Cluster) AddChildServers() error {
 // Start of topology detection
 // Create a connection to each host and build list of slaves.
 func (cluster *Cluster) TopologyDiscover(wcg *sync.WaitGroup) error {
+	cluster.Lock()
+	defer cluster.Unlock()
+
 	defer wcg.Done()
 	//monitor ignored server fist so that their replication position get oldest
 	wg := new(sync.WaitGroup)

cluster/prx.go

@@ -65,6 +65,7 @@ type Proxy struct {
 	ServiceName     string               `json:"serviceName"`
 	Agent           string               `json:"agent"`
 	Weight          string               `json:"weight"`
+	Lock            sync.Mutex
 }
 
 type DatabaseProxy interface {
@@ -83,7 +84,8 @@ type DatabaseProxy interface {
 
 	GetFailCount() int
 	SetFailCount(c int)
-
+	DelLock()
+	SetLock()
 	GetAgent() string
 	GetName() string
 	GetHost() string
@@ -319,7 +321,9 @@ func (cluster *Cluster) SetProxyServerMaintenance(serverid uint64) {
 // called  by server monitor if state change
 func (cluster *Cluster) backendStateChangeProxies() {
 	for _, pr := range cluster.Proxies {
+		pr.SetLock()
 		pr.BackendsStateChange()
+		pr.DelLock()
 	}
 }
 
@@ -332,6 +336,8 @@ func (cluster *Cluster) refreshProxies(wcg *sync.WaitGroup) {
 	for _, pr := range cluster.Proxies {
 		if pr != nil {
 			var err error
+			pr.SetLock()
+
 			err = pr.Refresh()
 			if err == nil {
 				pr.SetFailCount(0)
@@ -363,6 +369,7 @@ func (cluster *Cluster) refreshProxies(wcg *sync.WaitGroup) {
 			if cluster.Conf.GraphiteMetrics {
 				pr.SendStats()
 			}
+			pr.DelLock()
 		}
 	}
 	// if cluster.Conf.LogLevel > 2 {

cluster/prx_del.go

@@ -16,6 +16,10 @@ import (
 	"github.com/signal18/replication-manager/config"
 )
 
+func (proxy *Proxy) DelLock() {
+	proxy.Lock.Unlock()
+}
+
 func (proxy *Proxy) delCookie(key string) error {
 	err := os.Remove(proxy.Datadir + "/@" + key)
 	cluster := proxy.ClusterGroup

cluster/prx_proxysql.go

@@ -253,7 +253,6 @@ func (proxy *ProxySQLProxy) Failover() {
 }
 
 func (proxy *ProxySQLProxy) Refresh() error {
-
 	cluster := proxy.ClusterGroup
 	// if cluster.Conf.LogLevel > 9 {
 	cluster.LogModulePrintf(cluster.Conf.Verbose, config.ConstLogModProxySQL, LvlDbg, "ProxySQL port : %s, user %s, pass %s\n", proxy.Port, proxy.User, proxy.Pass)
@@ -298,6 +297,7 @@ func (proxy *ProxySQLProxy) Refresh() error {
 		if err != nil {
 			isFoundBackendWrite = false
 		} else {
+
 			proxy.BackendsWrite = append(proxy.BackendsWrite, bke)
 		}
 		isFoundBackendRead := true

cluster/prx_set.go

@@ -27,6 +27,10 @@ func (p *Proxy) SetID() {
 		10)
 }
 
+func (p *Proxy) SetLock() {
+	p.Lock.Lock()
+}
+
 // TODO: clarify where this is used, can maybe be replaced with a Getter
 func (proxy *Proxy) SetServiceName(namespace string) {
 	proxy.ServiceName = namespace + "/svc/" + proxy.Name

cluster/srv.go

@@ -184,6 +184,7 @@ type ServerMonitor struct {
 	DelayStat                   *ServerDelayStat             `json:"delayStat"`
 	IsInSlowQueryCapture        bool
 	IsInPFSQueryCapture         bool
+	sync.Mutex
 }
 
 type serverList []*ServerMonitor
@@ -595,6 +596,8 @@ var start_time time.Time
 
 // Refresh a server object
 func (server *ServerMonitor) Refresh() error {
+	server.Lock()
+	defer server.Unlock()
 	cluster := server.ClusterGroup
 	var err error
 

etc/opensvc/cluster-api/cluster-demo/stephane.toml

@@ -67,5 +67,5 @@ backup-streaming-endpoint= "https://s3.signal18.io/"
 backup-streaming-region= "fr-1"
 backup-streaming-bucket= "repman"
 backup-mysqldump-path = "/usr/local/bin/mysqldump"
-graphite-embedded     = true
+graphite-embedded     = false
 graphite-metrics = true

go.mod

@@ -119,6 +119,7 @@ require (
 	github.com/go-git/go-git v4.7.0+incompatible // indirect
 	github.com/go-git/go-git/v5 v5.6.1 // indirect
 	github.com/go-ole/go-ole v1.2.5 // indirect
+	github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
 	github.com/gonum/blas v0.0.0-20180125090452-e7c5890b24cf // indirect
 	github.com/gonum/floats v0.0.0-20180125090339-7de1f4ea7ab5 // indirect
 	github.com/gonum/internal v0.0.0-20180125090855-fda53f8d2571 // indirect

go.sum

@@ -287,6 +287,8 @@ github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zV
 github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/golang-jwt/jwt v3.2.2+incompatible h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keLg81eXfW3O+oY=
+github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/glog v0.0.0-20210429001901-424d2337a529 h1:2voWjNECnrZRbfwXxHB1/j8wa6xdKn85B5NzgVL/pTU=
 github.com/golang/glog v0.0.0-20210429001901-424d2337a529/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=

server/api.go

@@ -28,8 +28,8 @@ import (
 	"golang.org/x/oauth2"
 
 	"github.com/codegangsta/negroni"
-	jwt "github.com/dgrijalva/jwt-go"
-	"github.com/dgrijalva/jwt-go/request"
+	jwt "github.com/golang-jwt/jwt"
+	"github.com/golang-jwt/jwt/request"
 	"github.com/gorilla/mux"
 	"github.com/signal18/replication-manager/cert"
 	"github.com/signal18/replication-manager/cluster"
@@ -45,6 +45,8 @@ var apiPass string
 var apiUser string
 
 func (repman *ReplicationManager) initKeys() {
+	repman.Lock()
+	defer repman.Unlock()
 	var (
 		err         error
 		privKey     *rsa.PrivateKey

server/repmanv3.go

@@ -11,7 +11,7 @@ import (
 	"os"
 	"strings"
 
-	jwt "github.com/dgrijalva/jwt-go"
+	jwt "github.com/golang-jwt/jwt"
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
 	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"