alert-autoconf

1
from uuid import UUID
2
from datetime import time
3
from enum import Enum
4
from typing import Dict, List, Optional
5
from pydantic import BaseModel, AnyHttpUrl, validator, root_validator
6

7

8
class ContactTypeEnum(Enum):
9
    JIRA = "jira"
10
    MAIL = "mail"
11
    PUSHOVER = "pushover"
12
    SEND_SMS = "send-sms"
13
    SLACK = "slack"
14
    TELEGRAM = "telegram"
15
    TWILIO_SMS = "twilio sms"
16
    TWILIO_VOICE = "twilio voice"
17

18

19
class DaysEnum(Enum):
20
    MON = "Mon"
21
    TUE = "Tue"
22
    WED = "Wed"
23
    THU = "Thu"
24
    FRI = "Fri"
25
    SAT = "Sat"
26
    SUN = "Sun"
27

28

29
class TtlStateEnum(Enum):
30
    DEL = "DEL"
31
    ERROR = "ERROR"
32
    NODATA = "NODATA"
33
    OK = "OK"
34
    WARN = "WARN"
35

36

37
class ParentTriggerRef(BaseModel):
38
    tags: List[str]
39
    name: str
40

41
    def __hash__(self):
42
        return hash((
43
            frozenset(tags),
44
            name,
45
        ))
46

47
    def __eq__(self, other):
48
        return (
49
            set(self.tags) == set(other.tags)
50
            and self.name == other.name
51
        )
52

53

54
class Saturation(BaseModel):
55
    type: str
56
    fallback: Optional[str] = None
57
    parameters: Optional[dict] = None
58

59
    def to_custom_dict(self) -> Dict:
60
        result = {
61
            "type": self.type,
62
        }
63
        if self.fallback is not None:
64
            result["fallback"] = self.fallback
65
        if self.parameters is not None:
66
            result["extra_parameters"] = self.parameters
67
        return result
68

69
    @classmethod
70
    def from_moira_client_model(cls, moira_saturation: "moira_client.models.trigger.Saturation"):
71
        d = moira_saturation.to_dict()
72
        d["parameters"] = d.pop("extra_parameters", None)
73
        return cls(**d)
74

75
    def __hash__(self):
76
        dct = self.to_custom_dict()
77
        return hash(_freeze_dict(dct))
78

79
    def __eq__(self, other):
80
        if isinstance(other, Saturation):
81
            return self.to_custom_dict() == other.to_custom_dict()
82
        else:
83
            raise ValueError("Incomparable types")
84

85

86
def _freeze_dict(dct):
87
    """Tries to freeze a dict to make it hashable."""
88
    result = []
89
    for key, value in dct.items():
90
        if isinstance(value, dict):
91
            value = _freeze_dict(value)
92
        result.append((key, value))
93
    result.sort()
94
    return tuple(result)
95

96

97
class Trigger(BaseModel):
98
    id: Optional[str] = None
99
    name: str
100
    tags: List[str]
101
    targets: List[str]
102
    warn_value: Optional[int] = None
103
    error_value: Optional[int] = None
104
    desc: str = ""
105
    ttl: int = 600
106
    ttl_state: TtlStateEnum = TtlStateEnum.NODATA
107
    expression: Optional[str] = ""
108

109
    is_pull_type: bool = False
110
    dashboard: Optional[AnyHttpUrl] = None
111
    pending_interval: Optional[int] = 0
112

113
    day_disable: List[DaysEnum] = []
114
    time_start: Optional[time] = time(hour=0, minute=0)
115
    time_end: Optional[time] = time(hour=23, minute=59)
116

117
    parents: Optional[List[str]]
118

119
    saturation: Optional[List[Saturation]] = list()
120

121
    @validator("id")
122
    def id_uuid(cls, v):
123
        try:
124
            UUID(v)
125
        except ValueError:
126
            raise
127
        return v
128

129
    @root_validator
130
    def check_thresholds_values(cls, values):
131
        warn_value, error_value = (
132
            values.get('warn_value') is not None,
133
            values.get('error_value') is not None,
134
        )
135
        if warn_value ^ error_value:
136
            raise ValueError('must provide warn_value and error_value')
137

138
        if (
139
            warn_value & error_value
140
            and len(values.get('targets')) > 1
141
            and values.get('expression') is None
142
        ):
143
            raise ValueError('must use single target with warn_value and error_value')
144

145
        return values
146

147
    def to_custom_dict(self) -> Dict:
148
        return {
149
            'name': self.name,
150
            'tags': self.tags,
151
            'targets': self.targets,
152
            'warn_value': self.warn_value,
153
            'error_value': self.error_value,
154
            'desc': self.desc,
155
            'ttl': self.ttl,
156
            'ttl_state': self.ttl_state.value,
157
            'expression': self.expression,
158
            'is_pull_type': self.is_pull_type,
159
            'dashboard': self.dashboard,
160
            'pending_interval': self.pending_interval,
161
            'sched': {
162
                'startOffset': self.time_start.hour * 60 + self.time_start.minute,
163
                'endOffset': self.time_end.hour * 60 + self.time_end.minute,
164
                'tzOffset': 0,
165
                'days': [
166
                    {'name': day.value, 'enabled': day not in self.day_disable}
167
                    for day in DaysEnum
168
                ],
169
            },
170
            'parents': self.parents,
171
            'saturation': [
172
                s.to_custom_dict()
173
                for s in self.saturation
174
            ],
175
        }
176

177

178
class TriggerFile(Trigger):
179
    parents: Optional[List[ParentTriggerRef]]
180

181

182
class Contact(BaseModel):
183
    id: Optional[str] = None
184
    type: ContactTypeEnum
185
    value: str
186
    fallback_value: Optional[str] = None
187

188
    def __hash__(self):
189
        return f"{self.type}:{self.value}:{self.fallback_value}".__hash__()
190

191

192
class Escalation(BaseModel):
193
    contacts: List[Contact]
194
    offset_in_minutes: int = 0
195

196

197
class Subscription(BaseModel):
198
    tags: List[str]
199
    contacts: Optional[List[Contact]] = []
200
    escalations: Optional[List[Escalation]] = []
201
    day_disable: List[DaysEnum] = []
202
    time_start: Optional[time] = time(hour=0, minute=0)
203
    time_end: Optional[time] = time(hour=23, minute=59)
204

205
    def to_custom_dict(self) -> Dict:
206
        return {
207
            'tags': self.tags,
208
            'contacts': [c.id for c in self.contacts],
209
            'escalations': [
210
                {
211
                    'contacts': [c.id for c in e.contacts],
212
                    'offset_in_minutes': e.offset_in_minutes,
213
                }
214
                for e in self.escalations
215
            ],
216
            'sched': {
217
                'startOffset': self.time_start.hour * 60 + self.time_start.minute,
218
                'endOffset': self.time_end.hour * 60 + self.time_end.minute,
219
                'tzOffset': 0,
220
                'days': [
221
                    {'name': day.value, 'enabled': day not in self.day_disable}
222
                    for day in DaysEnum
223
                ],
224
            },
225
        }
226

227

228
class Alerts(BaseModel):
229
    version: float = 1
230
    prefix: str = ""
231
    triggers: List[TriggerFile] = []
232
    alerting: List[Subscription] = []
233