# By: Riasat Ullah # This class is a helper to the instance_monitor class. It helps allocate both internal and external # events correctly - triggers, internally handled prioritization/de-prioritization, dispatches and escalations. from objects.assignee import Assignee from objects.events import DispatchEvent, EscalateEvent, TriggerEvent, UrgencyAmendmentEvent from objects.instance_state import InstanceState from utils import constants, helpers, permissions, var_names import configuration as configs class EventAllocator(object): def __init__(self, all_policies, services, user_info, org_permissions, org_cost_percents): self.all_policies = all_policies self.services = services self.user_info = user_info self.org_permissions = org_permissions self.org_cost_percents = org_cost_percents self.triggers = [] self.dispatches = [] self.escalations = [] self.urgency_amendments = [] self.updated_instances = [] def handle_level_timeout(self, current_time, instance: InstanceState): ''' Handles the situation where the next alert timestamp set for the instance has timed out. This function sets up re-triggers, escalation and re-prioritization as appropriate. :param current_time: timestamp to check on :param instance: InstanceState object :return Instance object (new state after updates) ''' serv_id = instance.task.service_id() inst_serv = self.services[serv_id] if (serv_id is not None and serv_id in self.services) else None to_re_trigger_open_instance = False # If instance was de-prioritized when service was not in support and the service is set to re-prioritize it, # then re-trigger the alert at "high" urgency without escalating once the service is back in support. But # do not send alerts if the service is not back in support. At the time of de-prioritization the next alert # timestamp is set to be the next support start. So, de-prioritized instances that do not need re-prioritization # at the moment will ideally not be passed here from the instance monitor. However, if the support start of # the service changes, then we would need to handle that here. Nevertheless, doing an additional check # here does no harm. if instance.is_in_open_state(): if inst_serv is not None and not inst_serv.in_support_mode(instance.instance_timestamp)\ and instance.was_deprioritized(current_time): if inst_serv.on_hours_reprioritize and inst_serv.in_support_mode(current_time): new_urgency = constants.high_urgency instance.amend_urgency(new_urgency) self.urgency_amendments.append(UrgencyAmendmentEvent(instance.instance_id, current_time, constants.internal, new_urgency)) to_re_trigger_open_instance = True else: # If the instance had not been de-prioritized due to the service being out of service # and the instance is still in open state, then escalate. prev_esc_count = len(self.escalations) self.handle_instance_escalation(current_time, instance) if len(self.escalations) == prev_esc_count: to_re_trigger_open_instance = True # If the instance's acknowledged state has timed out, or it had been set to re-trigger # while in an open state, then re-trigger it. if (instance.is_in_acknowledged_state() and (inst_serv is None or (inst_serv is not None and inst_serv.re_trigger_minutes is not None)))\ or to_re_trigger_open_instance: for_policies = [self.all_policies[p_id] for p_id in instance.for_policy_ids() if p_id in self.all_policies] next_alert = InstanceState.calculate_next_alert_timestamp(instance.level, for_policies, current_time) instance.un_acknowledge(current_time, next_alert) self.triggers.append(TriggerEvent(instance.instance_id, current_time, constants.internal, next_alert)) self.updated_instances.append(instance) return instance def handle_instance_escalation(self, current_time, instance: InstanceState, escalated_by=None, access_method=constants.internal): ''' Handles escalation to the next level. The escalation happens only for group assignees when there is a higher level to assign to. :param current_time: timestamp this is being run at :param instance: InstanceState object :param escalated_by: (int) who the event is being escalated by :param access_method: (str) method by which this escalation is being made ''' # Use a set to make sure that an assignee does not get booked in more than once new_assignees = set() group_policy_ids = set() next_level = instance.level + 1 for assignee in instance.assignees: # only for group assignees user_policyid and for_policyid are not the same if assignee.user_policy_id != assignee.for_policy_id and assignee.for_policy_id in self.all_policies: parent_policy = self.all_policies[assignee.for_policy_id] # If the group has a higher level then escalate to the next level ONLY for the group assignee; # not for the other assignees of the instance next_level, next_level_users = parent_policy.get_next_available_level_on_call(next_level) for user_info in next_level_users: if user_info[2] not in [x.user_policy_id for x in new_assignees]: new_assignees.add(Assignee(assignee.for_policy_id, user_info[2], next_level)) group_policy_ids.add(assignee.for_policy_id) if len(group_policy_ids) > 0: new_assignees = list(new_assignees) group_policy_ids = list(group_policy_ids) next_alert = InstanceState.calculate_next_alert_timestamp( next_level, [self.all_policies[x] for x in group_policy_ids], current_time) esc_event = EscalateEvent(instance.instance_id, current_time, access_method, escalated_by, next_level, next_alert, group_policy_ids) # This step is only used to avoid empty or missing escalations. It cannot provide the Ack, Res and Esc # codes that are sent with SMS and voice calls. They are "null". This has to be adjusted in the monitor. instance.escalate(esc_event, new_assignees) self.escalations.append((esc_event, new_assignees)) self.updated_instances.append(instance) def handle_notification_alerts(self, current_time, instance: InstanceState): ''' Handles the notification alerts that need to be sent out to users based on their notification rules. :param current_time: timestamp this is being run at :param instance: InstanceState object ''' org_perm = self.org_permissions[instance.organization_id] org_cost = self.org_cost_percents[instance.organization_id]\ if instance.organization_id in self.org_cost_percents else 0 inst_updated = False for user_assignee in instance.assignees: if user_assignee.user_policy_id in self.user_info and\ not instance.is_assignee_over_notified(user_assignee.user_policy_id): user_assignee_info = self.user_info[user_assignee.user_policy_id] if permissions.has_user_permission(user_assignee_info[var_names.user_permissions], permissions.USER_INCIDENTS_RESPOND_PERMISSION): # find the minutes that have passed since the last notification was sent out last_notice = user_assignee.last_notification_timestamp minutes_passed_until_last_notice = helpers.get_minutes_diff_from_timestamps( instance.last_alert_timestamp, last_notice) if last_notice is not None else 0 minutes_passed_until_now = helpers.get_minutes_diff_from_timestamps( instance.last_alert_timestamp, current_time) # find the alert notification rule set for the user for this instance's urgency level notice_rules = user_assignee_info[var_names.notification_rules][instance.task.urgency_level() - 1] # Sort the minute buffers in ascending order so that we can stop at the first match. # Notification rules are saved as json in the database. So, the minutes need to be converted # to int from the str they are saved as in json. all_buffers = sorted([x for x in list(notice_rules.keys())]) if notice_rules is not None else [] # break out of the loop at the first match for minutes_buffer in all_buffers: if last_notice is None or\ (minutes_passed_until_last_notice < float(minutes_buffer) <= minutes_passed_until_now): notice_methods = notice_rules[minutes_buffer] for method in notice_methods: # Do not permit texts and voice calls if the cost usage is over the acceptable cost # threshold or if the organization does not have the correct permissions if method == constants.text: if org_cost > configs.max_alerting_cost_percent_threshold or\ not permissions.has_org_permission( org_perm, permissions.ORG_SMS_PERMISSION): break user_assignee.paid_notification_count += 1 elif method == constants.call: if org_cost > configs.max_alerting_cost_percent_threshold or\ not permissions.has_org_permission( org_perm, permissions.ORG_VOICE_CALL_PERMISSION): break user_assignee.paid_notification_count += 1 else: user_assignee.free_notification_count += 1 self.dispatches.append(DispatchEvent(instance.instance_id, current_time, method, user_assignee.to_dispatch_tuple(), user_assignee)) user_assignee.last_notification_timestamp = current_time inst_updated = True break if inst_updated: if instance not in self.updated_instances: self.updated_instances.append(instance)