alert-autoconf

1
import logging
2
from typing import List
3

4
import pydantic
5
from redis import Redis
6
import yaml
7

8
from alert_autoconf import models
9

10

11
# Redis keys
12
DEFAULT_PREFIX = "autoconf:defaults:"
13
DEFAULT_TRIGGER_KEY = DEFAULT_PREFIX + "triggers"
14

15

16
class DefaultCondition(pydantic.BaseModel):
17
    tags: List[str]
18

19
    def applies(self, trigger: models.TriggerFile):
20
        return set(self.tags).issubset(trigger.tags)
21

22

23
class DefaultValues(pydantic.BaseModel):
24
    parents: List[models.ParentTriggerRef]
25

26

27
class DefaultRule(pydantic.BaseModel):
28
    condition: DefaultCondition
29
    values: DefaultValues
30

31

32
class DefaultFile(pydantic.BaseModel):
33
    defaults: List[DefaultRule]
34

35

36
def apply_defaults(alerts: models.Alerts, redis: Redis):
37
    """Applies default values stored in Redis.
38
    As of version 0.4.15, the defaults are these:
39
    * all triggers with a MONAD tag get default parents.
40
    """
41

42
    default_file = redis.get(DEFAULT_TRIGGER_KEY)
43
    if default_file is None:
44
        logging.info("Defaults not found in Redis, skipping.")
45
    else:
46
        default_file = yaml.safe_load(default_file)
47
        default_file = DefaultFile(**default_file)
48
        default_rules = default_file.defaults
49

50
        for trigger in alerts.triggers:
51
            for default_rule in default_rules:
52
                if default_rule.condition.applies(trigger):
53
                    trigger.parents = _extend_without_duplicates(trigger.parents, default_rule.values.parents)
54

55

56
def _extend_without_duplicates(this, extension):
57
    if this is None:
58
        this = []
59
    for item in extension:
60
        if item not in this:
61
            this.append(item)
62
    return this
63