Refactor monitoring package

dashboard/dashboard.go

@@ -2,6 +2,7 @@ package dashboard
 
 import (
 	"sync"
+	"time"
 )
 
 type Probe struct {
@@ -15,14 +16,16 @@ type Data struct {
 
 type Operator struct {
 	sync.Mutex
-	workers map[int]chan *Data
-	In      chan *Data
+	workers     map[int]chan *Data
+	PollingFreq time.Duration
+	In          chan *Data
 }
 
 func Handle() *Operator {
 	operator := &Operator{
-		workers: make(map[int]chan *Data),
-		In:      make(chan *Data),
+		workers:     make(map[int]chan *Data),
+		PollingFreq: 5 * time.Second,
+		In:          make(chan *Data),
 	}
 	go func(operator *Operator) {
 		for {

monitoring/decision.go

@@ -0,0 +1 @@
+package monitoring

monitoring/monitoring.go

@@ -41,102 +41,145 @@ type Payload struct {
 	Machine       string            `json:"-"`
 }
 
-type probeObject struct {
+type probeWorker struct {
 	sync.Mutex
-	name       string
-	data       chan *Payload
-	stop       chan bool
-	status     probeStatus
-	counter    int
-	lastNormal time.Time
-	timeSerie  *probeTimeSerie
+	name         string
+	data         chan *Payload
+	stop         chan bool
+	status       probeStatus
+	counter      int
+	lastNormal   time.Time
+	timeSerie    *probeTimeSerie
+	headOperator *Operator
+}
+
+// Operator defines the handler for the monitoring
+// It contains the channel to receive the payload from the API server
+// It contains the map of the probes and the list of the probes
+// It contains the dashboard operator to inform the dashboard of the status of the probes
+// Everything that defines the monitoring event loop is here
+type Operator struct {
+	sync.Mutex
+	In                chan *Payload
+	probeMap          sync.Map
+	probeList         []string
+	dashboardOperator *dashboard.Operator
 }
 
 // Handle function handles the payload from the API server
-func Handle(channel chan *Payload, dashboardOperator *dashboard.Operator) {
+func Handle(dashboardOperator *dashboard.Operator) *Operator {
 	log.Debug("Starting monitoring.Handle")
-	var probeMap sync.Map
-	var probeList = make([]string, 0)
-	var timer = time.NewTimer(5 * time.Second)
+	operator := &Operator{
+		In:                make(chan *Payload),
+		probeMap:          sync.Map{},
+		probeList:         make([]string, 0),
+		dashboardOperator: dashboardOperator,
+	}
+	go func(operator *Operator, dashboardOperator *dashboard.Operator) {
+		var timer = time.NewTimer(0)
+		if dashboardOperator != nil {
+			timer = time.NewTimer(operator.dashboardOperator.PollingFreq)
+		}
+		for {
+			select {
+			case <-timer.C:
+				if dashboardOperator == nil {
+					continue
+				}
 
-	for {
-		select {
-		case <-timer.C:
-			if dashboardOperator == nil {
-				continue
-			}
+				operator.dashboardOperator.Lock()
+				operator.informDashboard()
+				timer.Reset(operator.dashboardOperator.PollingFreq)
+				operator.dashboardOperator.Unlock()
 
-			dashboardPayload := &dashboard.Data{
-				Probes: make([]*dashboard.Probe, 0),
-			}
+			case payload := <-operator.In:
+				if loaded, ok := operator.probeMap.Load(payload.Machine); ok {
+					probe, ok := loaded.(*probeWorker)
+					if !ok {
+						log.WithFields(log.Fields{
+							"machine": payload.Machine,
+						}).Fatal("Failed to load probe")
+					}
+
+					probe.informWorker(payload)
+				} else {
+					// Create a new probe
+					newProbe := makeProbe(payload, operator)
 
-			for _, probe := range probeList {
-				if loaded, ok := probeMap.Load(probe); ok {
-					probe := loaded.(*probeObject)
-					probe.Lock()
-					dashboardProbe := &dashboard.Probe{
-						Name:   strings.Clone(probe.name),
-						Status: strings.Clone(probe.status.String()),
+					value, loaded := operator.probeMap.LoadOrStore(payload.Machine, newProbe)
+					if loaded {
+						log.WithFields(log.Fields{
+							"machine": payload.Machine,
+						}).Fatal("Machine already exists")
 					}
-					probe.Unlock()
-					dashboardPayload.Probes = append(dashboardPayload.Probes, dashboardProbe)
-				}
-			}
 
-			dashboardOperator.In <- dashboardPayload
-			timer.Reset(5 * time.Second)
-			continue
-		case payload := <-channel:
-			if loaded, ok := probeMap.Load(payload.Machine); ok {
-				probe := loaded.(*probeObject)
-				if payload.MachineStatus == "delete" {
-					// Delete the probe
-					probe.Lock()
-					for i, name := range probeList {
-						if name == payload.Machine {
-							probeList = append(probeList[:i], probeList[i+1:]...)
-							break
-						}
+					probe, ok := value.(*probeWorker)
+					if !ok {
+						log.WithFields(log.Fields{
+							"machine": payload.Machine,
+						}).Fatal("Failed to load probe")
 					}
-					probe.delete()
-					probeMap.Delete(payload.Machine)
-				} else {
-					// Send the payload to the probe
-					probe.data <- payload
-				}
-			} else {
-				// Create a new probe
-				newProbe := makeProbe(payload)
 
-				value, loaded := probeMap.LoadOrStore(payload.Machine, newProbe)
-				if loaded {
 					log.WithFields(log.Fields{
+						"probe":   probe.name,
 						"machine": payload.Machine,
-					}).Fatal("Machine already exists")
-				}
+						"status":  probe.status,
+					}).Info("Starting probe thread")
 
-				probe, ok := value.(*probeObject)
-				if !ok {
-					log.WithFields(log.Fields{
-						"machine": payload.Machine,
-					}).Fatal("Failed to load probe")
+					operator.Lock()
+					operator.probeList = append(operator.probeList, payload.Machine)
+					operator.Unlock()
+
+					go probe.work()
+
+					probe.informWorker(payload)
 				}
+			}
+		}
+	}(operator, dashboardOperator)
+	return operator
+}
 
-				log.WithFields(log.Fields{
-					"probe":   probe.name,
-					"machine": payload.Machine,
-					"status":  probe.status,
-				}).Info("Starting probe thread")
+func (o *Operator) informDashboard() {
+	dashboardPayload := &dashboard.Data{
+		Probes: make([]*dashboard.Probe, 0),
+	}
+
+	for _, probe := range o.probeList {
+		if loaded, ok := o.probeMap.Load(probe); ok {
+			probe := loaded.(*probeWorker)
+			probe.Lock()
+			dashboardProbe := &dashboard.Probe{
+				Name:   strings.Clone(probe.name),
+				Status: strings.Clone(probe.status.String()),
+			}
+			probe.Unlock()
+			dashboardPayload.Probes = append(dashboardPayload.Probes, dashboardProbe)
+		}
+	}
+
+	o.dashboardOperator.In <- dashboardPayload
+}
 
-				probeList = append(probeList, payload.Machine)
-				go probe.work()
-				probe.data <- payload
+func (p *probeWorker) informWorker(payload *Payload) {
+	if payload.MachineStatus == "delete" {
+		p.Lock()
+		p.headOperator.Lock()
+		for i, name := range p.headOperator.probeList {
+			if name == payload.Machine {
+				p.headOperator.probeList = append(p.headOperator.probeList[:i], p.headOperator.probeList[i+1:]...)
+				break
 			}
 		}
+		p.delete()
+		p.headOperator.probeMap.Delete(payload.Machine)
+		p.headOperator.Unlock()
+	} else {
+		p.data <- payload
 	}
 }
 
-func (p *probeObject) work() {
+func (p *probeWorker) work() {
 	inactivityDelay, err := time.ParseDuration(config.Server.ProbeInactivityDelay)
 	if err != nil {
 		log.WithError(err).Fatal("Failed to parse inactivity delay")
@@ -172,7 +215,7 @@ func (p *probeObject) work() {
 	}
 }
 
-func (p *probeObject) timerIncrement() {
+func (p *probeWorker) timerIncrement() {
 	switch p.status {
 	case normal:
 		p.updateStatus()
@@ -204,7 +247,7 @@ func (p *probeObject) timerIncrement() {
 	}
 }
 
-func (p *probeObject) reset() {
+func (p *probeWorker) reset() {
 	if p.status > normal {
 		log.Infof("Machine %s is back in normal state\n", p.name)
 	}
@@ -213,7 +256,7 @@ func (p *probeObject) reset() {
 	p.lastNormal = time.Now()
 }
 
-func (p *probeObject) updateStatus() {
+func (p *probeWorker) updateStatus() {
 	p.status++
 	p.counter = 0
 	if p.status > normal {
@@ -224,7 +267,7 @@ func (p *probeObject) updateStatus() {
 	log.Infof("Machine %s is now in %s state\n", p.name, p.status.String())
 }
 
-func (p *probeObject) delete() {
+func (p *probeWorker) delete() {
 	log.WithFields(log.Fields{
 		"probe": p.name,
 	}).Info("Deleting probe")
@@ -233,8 +276,8 @@ func (p *probeObject) delete() {
 	close(p.stop)
 }
 
-func makeProbe(originPayload *Payload) *probeObject {
-	return &probeObject{
+func makeProbe(originPayload *Payload, operator *Operator) *probeWorker {
+	return &probeWorker{
 		name:       originPayload.Machine,
 		data:       make(chan *Payload, 1),
 		stop:       make(chan bool),
@@ -249,5 +292,6 @@ func makeProbe(originPayload *Payload) *probeObject {
 			},
 			size: 1,
 		},
+		headOperator: operator,
 	}
 }