Use 'State' as status

Author: Thibault Richard

api.go

@@ -28,9 +28,16 @@ func HandleEvent(c *gin.Context) {
 		return
 	}
 
+	status, err := extractStatus(data)
+	if err != nil {
+		logrus.WithError(err).WithField("body", data).Error("Fail to extract status while handling state event")
+		c.JSON(400, gin.H{"message": err.Error()})
+		return
+	}
+
 	event := types.Event{
 		ID:        ID,
-		Status:    types.StatusOK,
+		Status:    status,
 		Timestamp: time.Now(),
 		Value:     data,
 	}
@@ -45,17 +52,32 @@ func HandleEvent(c *gin.Context) {
 
 func extractHostAndService(obj map[string]interface{}) (string, error) {
 	host := obj["Host"]
+	if host == nil {
+		host = obj["Node"]
+	}
 	if host == nil {
 		return "", errors.New("Property 'Host' not found")
 	}
 	service := obj["Service"]
 	if service == nil {
-		return "", errors.New("Property 'Host' not found")
+		return "", errors.New("Property 'Service' not found")
 	}
 
 	return service.(string) + "/" + host.(string), nil
 }
 
+func extractStatus(obj map[string]interface{}) (string, error) {
+	status := obj["Status"]
+	if status == nil {
+		status = obj["State"]
+	}
+	if status == nil {
+		return "", errors.New("Property 'Status' not found")
+	}
+
+	return status.(string), nil
+}
+
 // Health return all events with status 500 if at least one event is in error
 func Health(c *gin.Context) {
 	mutex.RLock()

types/events.go

@@ -25,6 +25,10 @@ func (e Event) GetService() string {
 	return e.Value["Service"].(string)
 }
 
+func (e Event) GetStatus() string {
+	return e.Value["Status"].(string)
+}
+
 func (e Event) ToBytes() ([]byte, error) {
 	bytes, err := json.Marshal(e)
 	if err != nil {

watcher.go

@@ -7,38 +7,55 @@ import (
 	"github.com/thbkrkr/toctoc/types"
 )
 
+var (
+	healthTimeoutDuration = time.Second * time.Duration(healthTimeout)
+)
+
 func Watch() {
 	tick := time.NewTicker(time.Second * time.Duration(watchTick))
 	for range tick.C {
+		mutex.Lock()
 		for ns := range events {
 			for _, event := range events[ns] {
-				if time.Since(event.Timestamp) > time.Second*time.Duration(healthTimeout) {
+				if isKO(event) {
 					alert(ns, event)
 				}
 			}
 		}
+		mutex.Unlock()
 	}
 }
 
-func alert(ns string, event types.Event) {
-	mutex.Lock()
-	defer mutex.Unlock()
+func isKO(event types.Event) bool {
+	if event.Status == types.StatusKO {
+		return true
+	}
+	if time.Since(event.Timestamp).Seconds() > float64(healthTimeout) {
+		return true
+	}
+	return false
+}
 
+func alert(ns string, event types.Event) {
 	event.Status = types.StatusKO
 	events[ns][event.ID] = event
 
 	log.WithField("ns", ns).WithField("ID", event.ID).Errorf("No event since %d seconds", healthTimeout)
 
 	if kafkaAlerter {
-		sendAlertToKafka(event)
+		go sendAlertToKafka(event)
 	}
 }
 
 func sendAlertToKafka(event types.Event) {
-	bytes, err := event.ToBytes()
+	msg, err := event.ToBytes()
 	if err != nil {
 		log.WithField("Event", event).Errorf("Fail to marshal alert event")
 		return
 	}
-	go q.Send(bytes)
+	_, _, err = q.Send(msg)
+	if err != nil {
+		log.WithField("Event", event).Errorf("Fail to send event to kafka")
+		return
+	}
 }