majuna/app/models/alarms.py

import enum
from datetime import datetime
from typing import List, Any

from app.extensions import db
from app.models.activity import Activity


class AlarmState(enum.Enum):
    UNKNOWN = 0
    OK = 1
    WARNING = 2
    CRITICAL = 3

    @property
    def emoji(self) -> str:
        if self.name == "OK":
            return "😅"
        if self.name == "WARNING":
            return "😟"
        if self.name == "CRITICAL":
            return "🚨"
        return "❓"


class Alarm(db.Model):  # type: ignore
    id = db.Column(db.Integer, primary_key=True)
    target = db.Column(db.String(255), nullable=False)
    aspect = db.Column(db.String(255), nullable=False)
    alarm_state = db.Column(db.Enum(AlarmState), default=AlarmState.UNKNOWN, nullable=False)
    state_changed = db.Column(db.DateTime(), nullable=False)
    last_updated = db.Column(db.DateTime(), nullable=False)
    text = db.Column(db.String(255), nullable=False)

    @classmethod
    def csv_header(cls) -> List[str]:
        return ["id", "target", "alarm_type", "alarm_state", "state_changed", "last_updated", "text"]

    def csv_row(self) -> List[Any]:
        return [getattr(self, x) for x in self.csv_header()]

    def update_state(self, state: AlarmState, text: str) -> None:
        if self.alarm_state is None:
            self.alarm_state = AlarmState.UNKNOWN

        if self.alarm_state != state or self.state_changed is None:
            self.state_changed = datetime.utcnow()
            activity = Activity(activity_type="alarm_state",
                                text=f"[{self.aspect}] {self.state.emoji} Alarm state changed from "
                                     f"{self.alarm_state.name} to {state.name} on {self.target}: {text}.")
            if (self.alarm_state.name in ["WARNING", "CRITICAL"]
                    or state.name in ["WARNING", "CRITICAL"]):
                # Notifications are only sent on recovery from warning/critical state or on entry
                # to warning/critical states. This should reduce alert fatigue.
                activity.notify()
            db.session.add(activity)
        self.alarm_state = state
        self.text = text
        self.last_updated = datetime.utcnow()