oncall-engine/engine/apps/alerts/models/escalation_policy.py

import datetime

from django.conf import settings
from django.core.validators import MinLengthValidator
from django.db import models

from common.ordered_model.ordered_model import OrderedModel
from common.public_primary_keys import generate_public_primary_key, increase_public_primary_key_length


def generate_public_primary_key_for_escalation_policy():
    prefix = "E"
    new_public_primary_key = generate_public_primary_key(prefix)

    failure_counter = 0
    while EscalationPolicy.objects.filter(public_primary_key=new_public_primary_key).exists():
        new_public_primary_key = increase_public_primary_key_length(
            failure_counter=failure_counter, prefix=prefix, model_name="EscalationPolicy"
        )
        failure_counter += 1

    return new_public_primary_key


class EscalationPolicy(OrderedModel):
    order_with_respect_to = ["escalation_chain_id"]

    MAX_TIMES_REPEAT = 5

    (
        STEP_WAIT,
        _DEPRECATED_STEP_NOTIFY,  # only here to keep range intact
        STEP_FINAL_NOTIFYALL,
        STEP_REPEAT_ESCALATION_N_TIMES,
        STEP_FINAL_RESOLVE,
        STEP_NOTIFY_GROUP,
        STEP_NOTIFY_SCHEDULE,
        _DEPRECATED_STEP_NOTIFY_IMPORTANT,  # only here to keep range intact
        STEP_NOTIFY_GROUP_IMPORTANT,
        STEP_NOTIFY_SCHEDULE_IMPORTANT,
        _DEPRECATED_STEP_TRIGGER_CUSTOM_BUTTON,  # only here to keep range intact
        STEP_NOTIFY_USERS_QUEUE,
        STEP_NOTIFY_IF_TIME,
        STEP_NOTIFY_MULTIPLE_USERS,
        STEP_NOTIFY_MULTIPLE_USERS_IMPORTANT,
        STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW,
        STEP_TRIGGER_CUSTOM_WEBHOOK,
        STEP_NOTIFY_TEAM_MEMBERS,
        STEP_NOTIFY_TEAM_MEMBERS_IMPORTANT,
        STEP_DECLARE_INCIDENT,
        STEP_NOTIFY_USERS_QUEUE_IMPORTANT,
    ) = range(21)

    # Must be the same order as previous
    STEP_CHOICES = (
        (STEP_WAIT, "Wait"),
        (_DEPRECATED_STEP_NOTIFY, "Notify User"),
        (STEP_FINAL_NOTIFYALL, "Notify Whole Channel"),
        (STEP_REPEAT_ESCALATION_N_TIMES, "Repeat Escalation (5 times max)"),
        (STEP_FINAL_RESOLVE, "Resolve"),
        (STEP_NOTIFY_GROUP, "Notify Group"),
        (STEP_NOTIFY_SCHEDULE, "Notify Schedule"),
        (_DEPRECATED_STEP_NOTIFY_IMPORTANT, "Notify User (Important)"),
        (STEP_NOTIFY_GROUP_IMPORTANT, "Notify Group (Important)"),
        (STEP_NOTIFY_SCHEDULE_IMPORTANT, "Notify Schedule (Important)"),
        (_DEPRECATED_STEP_TRIGGER_CUSTOM_BUTTON, "Trigger Outgoing Webhook"),
        (STEP_NOTIFY_USERS_QUEUE, "Notify User (next each time)"),
        (STEP_NOTIFY_IF_TIME, "Continue escalation only if time is from"),
        (STEP_NOTIFY_MULTIPLE_USERS, "Notify multiple Users"),
        (STEP_NOTIFY_MULTIPLE_USERS_IMPORTANT, "Notify multiple Users (Important)"),
        (STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW, "Continue escalation if >X alerts per Y minutes"),
        (STEP_TRIGGER_CUSTOM_WEBHOOK, "Trigger Webhook"),
        (STEP_NOTIFY_TEAM_MEMBERS, "Notify all users in a Team"),
        (STEP_NOTIFY_TEAM_MEMBERS_IMPORTANT, "Notify all users in a Team (Important)"),
        (STEP_DECLARE_INCIDENT, "Declare Incident"),
        (STEP_NOTIFY_USERS_QUEUE_IMPORTANT, "Notify User (next each time) (Important)"),
    )

    # Ordered step choices available for internal api.
    # There are not important steps because they are presented as default step with important flag
    INTERNAL_API_STEPS = [
        # Common
        STEP_WAIT,
        STEP_NOTIFY_MULTIPLE_USERS,
        STEP_NOTIFY_TEAM_MEMBERS,
        STEP_NOTIFY_SCHEDULE,
        STEP_FINAL_RESOLVE,
        # Slack
        STEP_FINAL_NOTIFYALL,
        STEP_NOTIFY_GROUP,
        # Other
        STEP_TRIGGER_CUSTOM_WEBHOOK,
        STEP_NOTIFY_USERS_QUEUE,
        STEP_NOTIFY_IF_TIME,
        STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW,
        STEP_REPEAT_ESCALATION_N_TIMES,
        STEP_DECLARE_INCIDENT,
    ]
    # Steps can be stored in db while interacting with internal api
    # Includes important versions of default steps
    INTERNAL_DB_STEPS = [
        STEP_WAIT,
        STEP_FINAL_NOTIFYALL,
        STEP_FINAL_RESOLVE,
        STEP_NOTIFY_GROUP,
        STEP_NOTIFY_SCHEDULE,
        STEP_NOTIFY_GROUP_IMPORTANT,
        STEP_NOTIFY_SCHEDULE_IMPORTANT,
        STEP_NOTIFY_USERS_QUEUE,
        STEP_NOTIFY_IF_TIME,
        STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW,
        STEP_NOTIFY_TEAM_MEMBERS,
        STEP_NOTIFY_TEAM_MEMBERS_IMPORTANT,
        STEP_NOTIFY_MULTIPLE_USERS,
        STEP_NOTIFY_MULTIPLE_USERS_IMPORTANT,
        STEP_TRIGGER_CUSTOM_WEBHOOK,
        STEP_REPEAT_ESCALATION_N_TIMES,
        STEP_DECLARE_INCIDENT,
        STEP_NOTIFY_USERS_QUEUE_IMPORTANT,
    ]

    # Maps internal api's steps choices to their verbal. First string in tuple is display name for existent step.
    # Second one is for option in dropdown.
    INTERNAL_API_STEPS_TO_VERBAL_MAP = {
        # Common steps
        STEP_WAIT: ("Wait {{wait_delay}} minute(s)", "Wait"),
        STEP_NOTIFY_MULTIPLE_USERS: ("Start {{importance}} notification for {{users}}", "Notify users"),
        STEP_NOTIFY_TEAM_MEMBERS: (
            "Start {{importance}} notification for {{team}} team members",
            "Notify all team members",
        ),
        STEP_NOTIFY_SCHEDULE: (
            "Start {{importance}} notification for schedule {{schedule}}",
            "Notify users from on-call schedule",
        ),
        STEP_FINAL_RESOLVE: ("Resolve alert group automatically", "Resolve alert group automatically"),
        # Slack
        STEP_FINAL_NOTIFYALL: (
            "Escalate to all Slack channel members (use with caution)",
            "Escalate to all Slack channel members (use with caution)",
        ),
        STEP_NOTIFY_GROUP: (
            "Start {{importance}} notification for everyone from Slack User Group {{slack_user_group}}",
            "Notify Slack User Group",
        ),
        # Other
        STEP_TRIGGER_CUSTOM_WEBHOOK: ("Trigger webhook {{custom_webhook}}", "Trigger webhook"),
        STEP_NOTIFY_USERS_QUEUE: (
            "Round robin {{importance}} notification for {{users}}",
            "Notify users one by one (round-robin)",
        ),
        STEP_NOTIFY_IF_TIME: (
            "Continue escalation if current UTC time is in {{timerange}}",
            "Continue escalation if current UTC time is in range",
        ),
        STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW: (
            "Continue escalation if >{{num_alerts_in_window}} alerts per {{num_minutes_in_window}} minutes",
            "Continue escalation if >X alerts per Y minutes",
        ),
        STEP_REPEAT_ESCALATION_N_TIMES: (
            "Repeat escalation from the beginning (5 times max)",
            "Repeat escalations from the beginning (5 times max)",
        ),
        STEP_DECLARE_INCIDENT: (
            "Declare Incident with severity {{severity}} (non-default routes only)",
            "Declare Incident (non-default routes only)",
        ),
    }

    STEPS_WITH_NO_IMPORTANT_VERSION_SET = {
        STEP_WAIT,
        STEP_FINAL_NOTIFYALL,
        STEP_FINAL_RESOLVE,
        STEP_TRIGGER_CUSTOM_WEBHOOK,
        STEP_NOTIFY_IF_TIME,
        STEP_REPEAT_ESCALATION_N_TIMES,
        STEP_DECLARE_INCIDENT,
    }

    DEFAULT_TO_IMPORTANT_STEP_MAPPING = {
        STEP_NOTIFY_GROUP: STEP_NOTIFY_GROUP_IMPORTANT,
        STEP_NOTIFY_SCHEDULE: STEP_NOTIFY_SCHEDULE_IMPORTANT,
        STEP_NOTIFY_MULTIPLE_USERS: STEP_NOTIFY_MULTIPLE_USERS_IMPORTANT,
        STEP_NOTIFY_TEAM_MEMBERS: STEP_NOTIFY_TEAM_MEMBERS_IMPORTANT,
        STEP_NOTIFY_USERS_QUEUE: STEP_NOTIFY_USERS_QUEUE_IMPORTANT,
    }
    IMPORTANT_TO_DEFAULT_STEP_MAPPING = {
        STEP_NOTIFY_GROUP_IMPORTANT: STEP_NOTIFY_GROUP,
        STEP_NOTIFY_SCHEDULE_IMPORTANT: STEP_NOTIFY_SCHEDULE,
        STEP_NOTIFY_MULTIPLE_USERS_IMPORTANT: STEP_NOTIFY_MULTIPLE_USERS,
        STEP_NOTIFY_TEAM_MEMBERS_IMPORTANT: STEP_NOTIFY_TEAM_MEMBERS,
        STEP_NOTIFY_USERS_QUEUE_IMPORTANT: STEP_NOTIFY_USERS_QUEUE,
    }

    # Default steps are just usual version of important steps. E.g. notify group - notify group important
    DEFAULT_STEPS_SET = {
        STEP_NOTIFY_GROUP,
        STEP_NOTIFY_SCHEDULE,
        STEP_NOTIFY_MULTIPLE_USERS,
        STEP_NOTIFY_TEAM_MEMBERS,
        STEP_NOTIFY_USERS_QUEUE,
    }

    IMPORTANT_STEPS_SET = {
        STEP_NOTIFY_GROUP_IMPORTANT,
        STEP_NOTIFY_SCHEDULE_IMPORTANT,
        STEP_NOTIFY_MULTIPLE_USERS_IMPORTANT,
        STEP_NOTIFY_TEAM_MEMBERS_IMPORTANT,
        STEP_NOTIFY_USERS_QUEUE_IMPORTANT,
    }

    SLACK_INTEGRATION_REQUIRED_STEPS = [
        STEP_NOTIFY_GROUP,
        STEP_NOTIFY_GROUP_IMPORTANT,
        STEP_FINAL_NOTIFYALL,
    ]

    PUBLIC_STEP_CHOICES = [
        STEP_WAIT,
        STEP_NOTIFY_SCHEDULE,
        STEP_NOTIFY_MULTIPLE_USERS,
        STEP_NOTIFY_TEAM_MEMBERS,
        STEP_NOTIFY_USERS_QUEUE,
        STEP_NOTIFY_GROUP,
        STEP_FINAL_RESOLVE,
        STEP_FINAL_NOTIFYALL,
        STEP_TRIGGER_CUSTOM_WEBHOOK,
        STEP_NOTIFY_IF_TIME,
        STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW,
        STEP_REPEAT_ESCALATION_N_TIMES,
        STEP_DECLARE_INCIDENT,
    ]

    PUBLIC_STEP_CHOICES_MAP = {
        STEP_WAIT: "wait",
        STEP_FINAL_NOTIFYALL: "notify_whole_channel",
        STEP_FINAL_RESOLVE: "resolve",
        STEP_NOTIFY_GROUP: "notify_user_group",
        STEP_NOTIFY_GROUP_IMPORTANT: "notify_user_group",
        STEP_NOTIFY_SCHEDULE: "notify_on_call_from_schedule",
        STEP_NOTIFY_SCHEDULE_IMPORTANT: "notify_on_call_from_schedule",
        STEP_TRIGGER_CUSTOM_WEBHOOK: "trigger_webhook",
        STEP_NOTIFY_USERS_QUEUE: "notify_person_next_each_time",
        STEP_NOTIFY_MULTIPLE_USERS: "notify_persons",
        STEP_NOTIFY_MULTIPLE_USERS_IMPORTANT: "notify_persons",
        STEP_NOTIFY_TEAM_MEMBERS: "notify_team_members",
        STEP_NOTIFY_TEAM_MEMBERS_IMPORTANT: "notify_team_members",
        STEP_NOTIFY_IF_TIME: "notify_if_time_from_to",
        STEP_NOTIFY_IF_NUM_ALERTS_IN_TIME_WINDOW: "notify_if_num_alerts_in_window",
        STEP_REPEAT_ESCALATION_N_TIMES: "repeat_escalation",
        STEP_DECLARE_INCIDENT: "declare_incident",
    }

    public_primary_key = models.CharField(
        max_length=20,
        validators=[MinLengthValidator(settings.PUBLIC_PRIMARY_KEY_MIN_LENGTH + 1)],
        unique=True,
        default=generate_public_primary_key_for_escalation_policy,
    )

    escalation_chain = models.ForeignKey(
        "alerts.EscalationChain", on_delete=models.CASCADE, related_name="escalation_policies"
    )

    notify_to_users_queue = models.ManyToManyField(
        "user_management.User",
        related_name="escalation_policy_notify_queues",
    )

    last_notified_user = models.ForeignKey(
        "user_management.User",
        on_delete=models.SET_NULL,
        related_name="last_notified_in_escalation_policies",
        default=None,
        null=True,
    )

    step = models.IntegerField(choices=STEP_CHOICES, default=None, null=True)

    notify_to_team_members = models.ForeignKey(
        "user_management.Team",
        on_delete=models.SET_NULL,
        related_name="escalation_policies",
        default=None,
        null=True,
    )

    notify_to_group = models.ForeignKey(
        "slack.SlackUserGroup",
        on_delete=models.SET_NULL,
        related_name="escalation_policies",
        default=None,
        null=True,
    )

    notify_schedule = models.ForeignKey(
        "schedules.OnCallSchedule",
        on_delete=models.SET_NULL,
        related_name="escalation_policies",
        null=True,
        default=None,
    )

    custom_webhook = models.ForeignKey(
        "webhooks.Webhook",
        on_delete=models.CASCADE,
        related_name="escalation_policies",
        default=None,
        null=True,
    )

    # Incident severity for declare incident step
    SEVERITY_SET_FROM_LABEL, SEVERITY_SET_FROM_LABEL_DISPLAY_VALUE = ("set-from-label", "from 'severity' label")
    severity = models.CharField(max_length=512, null=True, default=None)

    ONE_MINUTE = datetime.timedelta(minutes=1)
    FIVE_MINUTES = datetime.timedelta(minutes=5)
    FIFTEEN_MINUTES = datetime.timedelta(minutes=15)
    THIRTY_MINUTES = datetime.timedelta(minutes=30)
    HOUR = datetime.timedelta(minutes=60)

    DEFAULT_WAIT_DELAY = datetime.timedelta(minutes=5)

    DURATION_CHOICES = (
        (ONE_MINUTE, "1 min"),
        (FIVE_MINUTES, "5 min"),
        (FIFTEEN_MINUTES, "15 min"),
        (THIRTY_MINUTES, "30 min"),
        (HOUR, "60 min"),
    )

    WEB_DURATION_CHOICES = (
        (ONE_MINUTE, "1"),
        (FIVE_MINUTES, "5"),
        (FIFTEEN_MINUTES, "15"),
        (THIRTY_MINUTES, "30"),
        (HOUR, "60"),
    )

    # the same choices for web, but in integer format for minutes instead of timedelta
    WEB_DURATION_CHOICES_MINUTES = [(choice[0].seconds // 60, choice[1]) for choice in WEB_DURATION_CHOICES]

    wait_delay = models.DurationField(default=None, null=True, choices=DURATION_CHOICES)

    from_time = models.TimeField(null=True, default=None)
    to_time = models.TimeField(null=True, default=None)

    # fields needed for escalation step "Continue escalation if >X alerts per Y minutes"
    num_alerts_in_window = models.PositiveIntegerField(null=True, default=None)
    num_minutes_in_window = models.PositiveIntegerField(null=True, default=None)

    class Meta:
        ordering = ["order"]
        constraints = [
            models.UniqueConstraint(fields=["escalation_chain_id", "order"], name="unique_escalation_policy_order")
        ]

    def __str__(self):
        return f"{self.pk}: {self.step_type_verbal}"

    @property
    def step_type_verbal(self):
        return self.STEP_CHOICES[self.step][1] if self.step is not None else "Empty"

    @property
    def sorted_users_queue(self):
        return sorted(self.notify_to_users_queue.all(), key=lambda user: (user.username or "", user.pk))

    @property
    def slack_integration_required(self) -> bool:
        if self.step in self.SLACK_INTEGRATION_REQUIRED_STEPS:
            return True
        else:
            return False

    @staticmethod
    def get_step_display_name(step):
        step_name = ""
        for step_choice in EscalationPolicy.STEP_CHOICES:
            if step_choice[0] == step:
                step_name = step_choice[1]
                break
        return step_name

    # Insight logs
    @property
    def insight_logs_type_verbal(self):
        return "escalation_policy"

    @property
    def insight_logs_verbal(self):
        return f"Escalation Policy  {self.order} in {self.escalation_chain.insight_logs_verbal}"

    @property
    def insight_logs_serialized(self):
        result = {
            "type": self.step_type_verbal,
            "order": self.order,
        }

        if self.step == EscalationPolicy.STEP_WAIT:
            if self.wait_delay:
                result["wait_delay"] = self.get_wait_delay_display()
        elif self.step in [EscalationPolicy.STEP_NOTIFY_GROUP, EscalationPolicy.STEP_NOTIFY_GROUP_IMPORTANT]:
            if self.notify_to_group:
                result["user_group"] = self.notify_to_group.name
                result["user_group_id"] = self.notify_to_group.public_primary_key
        elif self.step in [
            EscalationPolicy.STEP_NOTIFY_TEAM_MEMBERS,
            EscalationPolicy.STEP_NOTIFY_TEAM_MEMBERS_IMPORTANT,
        ]:
            if self.notify_to_team_members:
                result["team"] = self.notify_to_team_members.name
                result["team_id"] = self.notify_to_team_members.public_primary_key
        elif self.step in [EscalationPolicy.STEP_NOTIFY_SCHEDULE, EscalationPolicy.STEP_NOTIFY_SCHEDULE_IMPORTANT]:
            if self.notify_schedule:
                result["on-call_schedule"] = self.notify_schedule.insight_logs_verbal
                result["on-call_schedule_id"] = self.notify_schedule.public_primary_key
        elif self.step == EscalationPolicy.STEP_TRIGGER_CUSTOM_WEBHOOK:
            if self.custom_webhook:
                result["outgoing_webhook"] = self.custom_webhook.insight_logs_verbal
                result["outgoing_webhook_id"] = self.custom_webhook.public_primary_key
        elif self.step in [
            EscalationPolicy.STEP_NOTIFY_USERS_QUEUE,
            EscalationPolicy.STEP_NOTIFY_MULTIPLE_USERS,
            EscalationPolicy.STEP_NOTIFY_MULTIPLE_USERS_IMPORTANT,
        ]:
            if self.notify_to_users_queue:
                result["notify_users"] = [user.username for user in self.sorted_users_queue]
                result["notify_users_ids"] = [user.public_primary_key for user in self.sorted_users_queue]
        elif self.step == EscalationPolicy.STEP_NOTIFY_IF_TIME:
            if self.from_time:
                result["from_time"] = self.from_time.isoformat() + " (UTC)"
            if self.to_time:
                result["to_time"] = self.to_time.isoformat() + " (UTC)"

        return result

    @property
    def insight_logs_metadata(self):
        return {
            "escalation_chain": self.escalation_chain.insight_logs_verbal,
            "escalation_chain_id": self.escalation_chain.public_primary_key,
        }