graphicek/server/services/alerts_evaluator_service.go

package services

import (
	"basic-sensor-receiver/integrations"
	"basic-sensor-receiver/models"
	"database/sql"
	"encoding/json"
	"fmt"
	"time"
)

type AlertsEvaluatorService struct {
	ctx *Context
}

func evaluateSensorValueCondition(condition *models.AlertConditionSensorValue, value *sensorValue) bool {
	switch condition.Condition {
	case "less":
		return value.Value < condition.Value
	case "more":
		return value.Value > condition.Value
	case "equal":
		return value.Value == condition.Value
	}

	return false
}

func (s *AlertsEvaluatorService) EvaluateAlerts() error {
	alerts, err := s.ctx.Services.Alerts.GetList()

	if err != nil {
		return err
	}

	for _, alert := range alerts {
		err := s.EvaluateAlert(&alert)

		if err != nil {
			return err
		}
	}

	return nil

}

func unitToSeconds(unit string) float64 {
	switch unit {
	case "s":
		return 1
	case "m":
		return 60
	case "h":
		return 3600
	case "d":
		return 86400
	}

	return 0
}

func (s *AlertsEvaluatorService) EvaluateAlert(alert *models.AlertItem) error {
	condition := map[string]interface{}{}
	err := json.Unmarshal([]byte(alert.Condition), &condition)
	if err != nil {
		return err
	}

	// TODO: This flow is a bit cumbersome, consider refactoring
	newStatus := alert.LastStatus
	lastValue := float64(0)
	conditionMet := false
	sensorId := int64(-1)
	var sensorValueCondition *models.AlertConditionSensorValue
	var sensorLastContactCondition *models.AlertConditionSensorLastContact

	switch condition["type"].(string) {
	case models.AlertConditionSensorValueType:
		{
			sensorValueCondition = &models.AlertConditionSensorValue{
				SensorId:  int64(condition["sensorId"].(float64)),
				Condition: condition["condition"].(string),
				Value:     condition["value"].(float64),
			}

			value, err := s.ctx.Services.SensorValues.GetLatest(sensorValueCondition.SensorId, time.Now().Unix())
			lastValue = value.Value
			sensorId = int64(sensorValueCondition.SensorId)

			if err != nil {
				if err == sql.ErrNoRows {
					lastValue = float64(0)
					conditionMet = false
				} else {
					return fmt.Errorf("error getting sensor value: %v", err)
				}
			} else {
				conditionMet = evaluateSensorValueCondition(sensorValueCondition, value)
			}

			break
		}

	case models.AlertConditionSensorLastContactType:
		{
			sensorLastContactCondition = &models.AlertConditionSensorLastContact{
				SensorId:  int64(condition["sensorId"].(float64)),
				Value:     condition["value"].(float64),
				ValueUnit: condition["valueUnit"].(string),
			}

			value, err := s.ctx.Services.SensorValues.GetLatest(sensorLastContactCondition.SensorId, time.Now().Unix())
			lastValue = float64(value.Timestamp)
			sensorId = sensorLastContactCondition.SensorId

			if err != nil {
				if err == sql.ErrNoRows {
					lastValue = float64(0)
				} else {
					return fmt.Errorf("error getting sensor last contact value: %v", err)
				}
			}

			conditionInSec := int64(sensorLastContactCondition.Value * unitToSeconds(sensorLastContactCondition.ValueUnit))

			conditionMet = time.Now().Unix()-value.Timestamp > conditionInSec

			break
		}

	}

	/*
		Status flow:
		  ok -> alert_pending
			alert_pending -> alerting OR alert_pending -> ok
			alerting -> ok_pending
			ok_pending -> ok OR ok_pending -> alerting
	*/
	if conditionMet {
		if alert.LastStatus == models.AlertStatusOk {
			newStatus = models.AlertStatusAlertPending
		} else if alert.LastStatus == models.AlertStatusAlertPending {
			if time.Now().Unix()-alert.LastStatusAt >= alert.TriggerInterval {
				newStatus = models.AlertStatusAlerting
			}
		}
	} else {
		if alert.LastStatus == models.AlertStatusAlerting {
			newStatus = models.AlertStatusOkPending
		} else if alert.LastStatus == models.AlertStatusOkPending {
			if time.Now().Unix()-alert.LastStatusAt >= alert.TriggerInterval {
				newStatus = models.AlertStatusOk
			}
		}
	}

	if newStatus != alert.LastStatus || newStatus == models.AlertStatusOk {
		s.ctx.DB.Exec("UPDATE alerts SET last_status = ?, last_status_at = ? WHERE id = ?", newStatus, time.Now().Unix(), alert.Id)

		if newStatus == models.AlertStatusAlerting || newStatus == models.AlertStatusOk {
			sensor, err := s.ctx.Services.Sensors.GetById(sensorId)
			if err != nil {
				return fmt.Errorf("error getting sensor detail (%d): %v", sensorId, err)
			}

			contactPoint, err := s.ctx.Services.ContactPoints.GetById(alert.ContactPointId)
			if err != nil {
				return fmt.Errorf("error getting contact point detail: %v", err)
			}

			dispatchService, err := contactPoint.getService(s.ctx)
			if err != nil {
				return fmt.Errorf("error getting dispatch service: %v", err)
			}

			// Dispatch alert resolved event when we are transitioning ok_pending -> ok
			if newStatus == models.AlertStatusOk && alert.LastStatus == models.AlertStatusOkPending {
				err = dispatchService.ProcessEvent(&integrations.ContactPointEvent{
					Type: integrations.ContactPointEventAlertResolved,
					AlertResolvedEvent: &integrations.ContactPointAlertResolvedEvent{
						Alert:                      alert,
						Sensor:                     sensor,
						SensorValueCondition:       sensorValueCondition,
						SensorLastContactCondition: sensorLastContactCondition,
						LastValue:                  lastValue,
					},
				}, contactPoint.TypeConfig)
			}

			// Dispatch alert triggered event when we are transitioning alert_pending -> alerting
			if newStatus == models.AlertStatusAlerting && alert.LastStatus == models.AlertStatusAlertPending {
				err = dispatchService.ProcessEvent(&integrations.ContactPointEvent{
					Type: integrations.ContactPointEventAlertTriggered,
					AlertTriggeredEvent: &integrations.ContactPointAlertTriggeredEvent{
						Alert:                      alert,
						Sensor:                     sensor,
						SensorValueCondition:       sensorValueCondition,
						SensorLastContactCondition: sensorLastContactCondition,
						LastValue:                  lastValue,
					},
				}, contactPoint.TypeConfig)
			}

			if err != nil {
				return fmt.Errorf("error dispatching alert: %v", err)
			}
		}
	}

	return nil
}