Fixing race conditions in the code base (#5966)

* Adding initial race detector

* Remove setting of config twice

* Fixing config file watch and config reload on license save

* Fixing config file watch and config reload on license save

* Fixing build error

* Fixing locking issue

* Fixing makefile

* Fixing race in config

* Fixing race in status unit test

* Adding EE race tests

* Fixing race in cluster info

* Removing code that's isn't needed

* Fixing some more races

* Fixing govet issue

model/cluster_info.go

@@ -6,19 +6,25 @@ package model
 import (
 	"encoding/json"
 	"io"
+	"strings"
+	"sync"
+	"sync/atomic"
 )
 
 type ClusterInfo struct {
-	Id                 string `json:"id"`
-	Version            string `json:"version"`
-	ConfigHash         string `json:"config_hash"`
-	InterNodeUrl       string `json:"internode_url"`
-	Hostname           string `json:"hostname"`
-	LastSuccessfulPing int64  `json:"last_ping"`
-	IsAlive            bool   `json:"is_alive"`
+	Id                 string       `json:"id"`
+	Version            string       `json:"version"`
+	ConfigHash         string       `json:"config_hash"`
+	InterNodeUrl       string       `json:"internode_url"`
+	Hostname           string       `json:"hostname"`
+	LastSuccessfulPing int64        `json:"last_ping"`
+	Alive              int32        `json:"is_alive"`
+	Mutex              sync.RWMutex `json:"-"`
 }
 
 func (me *ClusterInfo) ToJson() string {
+	me.Mutex.RLock()
+	defer me.Mutex.RUnlock()
 	b, err := json.Marshal(me)
 	if err != nil {
 		return ""
@@ -27,9 +33,15 @@ func (me *ClusterInfo) ToJson() string {
 	}
 }
 
+func (me *ClusterInfo) Copy() *ClusterInfo {
+	json := me.ToJson()
+	return ClusterInfoFromJson(strings.NewReader(json))
+}
+
 func ClusterInfoFromJson(data io.Reader) *ClusterInfo {
 	decoder := json.NewDecoder(data)
 	var me ClusterInfo
+	me.Mutex = sync.RWMutex{}
 	err := decoder.Decode(&me)
 	if err == nil {
 		return &me
@@ -38,7 +50,21 @@ func ClusterInfoFromJson(data io.Reader) *ClusterInfo {
 	}
 }
 
+func (me *ClusterInfo) SetAlive(alive bool) {
+	if alive {
+		atomic.StoreInt32(&me.Alive, 1)
+	} else {
+		atomic.StoreInt32(&me.Alive, 0)
+	}
+}
+
+func (me *ClusterInfo) IsAlive() bool {
+	return atomic.LoadInt32(&me.Alive) == 1
+}
+
 func (me *ClusterInfo) HaveEstablishedInitialContact() bool {
+	me.Mutex.RLock()
+	defer me.Mutex.RUnlock()
 	if me.Id != "" {
 		return true
 	}
@@ -46,6 +72,16 @@ func (me *ClusterInfo) HaveEstablishedInitialContact() bool {
 	return false
 }
 
+func (me *ClusterInfo) IdEqualTo(in string) bool {
+	me.Mutex.RLock()
+	defer me.Mutex.RUnlock()
+	if me.Id == in {
+		return true
+	}
+
+	return false
+}
+
 func ClusterInfosToJson(objmap []*ClusterInfo) string {
 	if b, err := json.Marshal(objmap); err != nil {
 		return ""

model/cluster_info_test.go

@@ -16,6 +16,16 @@ func TestClusterInfoJson(t *testing.T) {
 	if cluster.Id != result.Id {
 		t.Fatal("Ids do not match")
 	}
+
+	cluster.SetAlive(true)
+	if !cluster.IsAlive() {
+		t.Fatal("should be live")
+	}
+
+	cluster.SetAlive(false)
+	if cluster.IsAlive() {
+		t.Fatal("should be not live")
+	}
 }
 
 func TestClusterInfosJson(t *testing.T) {

model/job.go

@@ -5,6 +5,7 @@ package model
 
 import (
 	"fmt"
+	"sync"
 	"time"
 )
 
@@ -18,17 +19,24 @@ type ScheduledTask struct {
 	timer     *time.Timer
 }
 
+var taskMutex = sync.Mutex{}
 var tasks = make(map[string]*ScheduledTask)
 
 func addTask(task *ScheduledTask) {
+	taskMutex.Lock()
+	defer taskMutex.Unlock()
 	tasks[task.Name] = task
 }
 
 func removeTaskByName(name string) {
+	taskMutex.Lock()
+	defer taskMutex.Unlock()
 	delete(tasks, name)
 }
 
 func GetTaskByName(name string) *ScheduledTask {
+	taskMutex.Lock()
+	defer taskMutex.Unlock()
 	if task, ok := tasks[name]; ok {
 		return task
 	}
@@ -36,6 +44,8 @@ func GetTaskByName(name string) *ScheduledTask {
 }
 
 func GetAllTasks() *map[string]*ScheduledTask {
+	taskMutex.Lock()
+	defer taskMutex.Unlock()
 	return &tasks
 }