# By: Riasat Ullah # This class represents the state of an instance at a point in time. from objects.assignee import Assignee from objects.events import AcknowledgeEvent, AddRespondersEvent, EscalateEvent, Event, NotateEvent, \ ReassignEvent, ResolveEvent, RunWorkflowEvent, SnoozeEvent, TriggerEvent, UrgencyAmendmentEvent from objects.task import Task from utils import constants, helpers, times, var_names import configuration import datetime class InstanceState(object): def __init__(self, instance_id, organization_instance_id, organization_id, instance_timestamp, state_timestamp, state_level, last_alert_timestamp, next_alert_timestamp, status, state_assignees, task: Task, events, notes=None, updates=None, subscribers=None, impacted_business_services=None, resolved_on=None, assignees_history=None, business_services_history=None, conference_bridges=None, synced_vendors=None): ''' InstanceState object :param instance_id: global ID of the instance :param organization_instance_id: organization specific ID of the instance :param organization_id: organization ID :param instance_timestamp: timestamp when the instance occurred :param state_timestamp: timestamp when this new state of the instance developed :param state_level: the level the instance is at now :param last_alert_timestamp: the last time the instance was triggered :param next_alert_timestamp: the next time the instance should be re-triggered :param status: the status the instance is in now - OPEN, ACKNOWLEDGED, RESOLVED :param state_assignees: (list) of Assignee objects of those who the instance is assigned to now :param task: (Task object) the task that triggered the instance :param events: (list) of Event objects :param notes: (list) of notes :param updates: (list) of updates :param subscribers: (list) of user_ids of subscribers :param impacted_business_services: (list of tuples) -> [ (bus serv id, bus serv name), ...] :param resolved_on: timestamp when the instance was resolved :param assignees_history: (list of list) -> [ [user pol id, for pol id, start time, end time], ... ] :param business_services_history: (list of list) -> [ bus serv id, bus name, start, end], ... ] :param conference_bridges: (list) -> [conference phone, conference url] :param synced_vendors: (list of dict) -> [{synced_with: , vendor_id: , vendor_url: }, ...] ''' self.instance_id = instance_id self.organization_instance_id = organization_instance_id self.organization_id = organization_id self.instance_timestamp = instance_timestamp self.timestamp = state_timestamp self.level = state_level self.last_alert_timestamp = last_alert_timestamp self.next_alert_timestamp = next_alert_timestamp self.status = status self.assignees = state_assignees self.task = task self.events = events self.notes = notes self.updates = updates self.subscribers = subscribers self.impacted_business_services = impacted_business_services self.resolved_on = resolved_on self.assignees_history = assignees_history self.business_services_history = business_services_history self.conference_bridges = conference_bridges self.synced_vendors = synced_vendors @staticmethod def create_instance(details): ''' Creates an InstanceState object from a dict of instance details. :param details: (dict) of instance details :return: InstanceState object ''' inst_id = details[var_names.instance_id] return InstanceState( inst_id, details[var_names.organization_instance_id], details[var_names.organization_id], times.get_timestamp_from_string(details[var_names.instance_timestamp]), times.get_timestamp_from_string(details[var_names.state_timestamp]), details[var_names.assignee_level], times.get_timestamp_from_string(details[var_names.last_alert_timestamp]), times.get_timestamp_from_string(details[var_names.next_alert_timestamp]), details[var_names.status], InstanceState.create_instance_assignees(details[var_names.assignees]), Task.create_task(details[var_names.task]), Event.create_events(inst_id, details[var_names.events]), notes=details[var_names.notes] if var_names.notes in details else None, updates=details[var_names.status_update] if var_names.status_update in details else None, subscribers=details[var_names.subscribers] if var_names.subscribers in details else None, impacted_business_services=details[var_names.impacted_business_services] if var_names.impacted_business_services in details else None, resolved_on=times.get_timestamp_from_string(details[var_names.resolved_on]) if var_names.resolved_on in details and details[var_names.resolved_on] is not None else None, assignees_history=details[var_names.assignees_history] if var_names.assignees_history in details else None, synced_vendors=details[var_names.synced_vendors] if var_names.synced_vendors in details else None ) @staticmethod def create_instance_assignees(assignees): ''' Creates assignee objects and puts them in a list given assignees info. :param assignees: (list) of dict of info :return: (list) of Assignee objects ''' assignee_objects = [] for item in assignees: assignee_objects.append(Assignee.create_assignee(item)) return assignee_objects def was_reassigned(self): ''' Confirms if the instance was re-assigned at all or not :return: (boolean) True if it was re-assigned; False otherwise ''' for event in self.events: if isinstance(event, ReassignEvent): return True return False def group_assignee_ids_and_level(self): ''' Gets policy id and level of the assignees that are group policies. :return: (list) of tuples -> [(policy id, level), ...] ''' id_set = set() for assignee in self.assignees: if assignee.for_policy_id != assignee.user_policy_id: id_set.add((assignee.for_policy_id, assignee.assignee_level)) return list(id_set) def user_assignee_ids_and_level(self): ''' Gets policy id and level of the assignees that are user policies. :return: (list) of tuples -> [(policy id, level), ...] ''' id_set = set() for assignee in self.assignees: if assignee.for_policy_id == assignee.user_policy_id: id_set.add((assignee.user_policy_id, assignee.assignee_level)) return list(id_set) def for_policy_ids(self): ''' Get all the for policy ids (main policy ids) the instance is assigned to. :return: (set) of policy ids ''' ids = set() for assignee in self.assignees: ids.add(assignee.for_policy_id) return ids def get_first_dispatch_time(self, user_policy_id=None, group_policy_id=None): ''' Get the first dispatch time. If user policy id and group policy id is not given, then it gets the timestamp of very first event which should be a dispatch event. :param user_policy_id: user policy id to check for :param group_policy_id: group policy id to check for :return: ''' if len(self.events) > 0: if user_policy_id is None and group_policy_id is None: return self.events[0].event_timestamp else: if self.assignees_history is not None: for item in self.assignees_history: if item[0] == user_policy_id or item[1] == group_policy_id: return item[2] return None def get_response_time(self, user_id=None, user_policy_id=None, group_policy_id=None): ''' Gets the time taken by a given user to first respond to an instance. If the user was not the first person to respond, then None will be returned. :param user_id: user_id of the user to check for as the first responder :param user_policy_id: policy ID of the user :param group_policy_id: policy ID of the group :return: (float) time taken in minutes to respond ''' dispatch_time = self.get_first_dispatch_time(user_policy_id, group_policy_id) if dispatch_time is not None: for event in self.events: if isinstance(event, AcknowledgeEvent)\ or isinstance(event, ResolveEvent)\ or isinstance(event, SnoozeEvent): if user_id is None or (user_id is not None and event.event_by == user_id): return helpers.get_minutes_diff_from_timestamps(dispatch_time, event.event_timestamp) return None def get_resolution_time(self, user_id=None, user_policy_id=None, group_policy_id=None): ''' Gets the time taken to resolve an incident. If the incident was not resolved or if a user_id is given, but that user did not resolve the incident, then None will be returned. :param user_id: user_id of the user to check for as the resolver :param user_policy_id: policy ID of the user :param group_policy_id: policy ID of the group :return: (float) time taken in minutes to resolve ''' dispatch_time = self.get_first_dispatch_time(user_policy_id, group_policy_id) if dispatch_time is not None: last_event = self.events[-1] if isinstance(last_event, ResolveEvent): if user_id is None or (user_id is not None and last_event.event_by == user_id): return helpers.get_minutes_diff_from_timestamps(dispatch_time, last_event.event_timestamp) return None def get_time_spent(self, user_policy_id=None, group_policy_id=None): ''' Gets the total time spent on the instance. If the user_id is provided, then it will get the maximum time spent by the user. :param user_policy_id: policy ID of the user :param group_policy_id: policy ID of the group :return: (float) time taken in minutes to resolve ''' if user_policy_id is None and group_policy_id is None: dispatch_time = self.get_first_dispatch_time() last_event_time = self.resolved_on if self.status == constants.resolved_state\ else self.events[-1].event_timestamp return helpers.get_minutes_diff_from_timestamps(dispatch_time, last_event_time) else: total_time = 0 if self.assignees_history is not None: for item in self.assignees_history: if item[0] == user_policy_id or item[1] == group_policy_id: total_time += helpers.get_minutes_diff_from_timestamps(item[2], item[3]) return total_time @staticmethod def get_time_spent_from_assignees_history_list(assignees_history, user_policy_id=None, group_policy_id=None): ''' Gets the amount of time is spent by a user/group on an instance from the dict state of events. :param assignees_history: (list) of instance events as dict; not Instance objects :param user_policy_id: policy id of the user :param group_policy_id: policy ID of the group :return: (datetime.timedelta) representing the interrupted time ''' total_time = 0 for item in assignees_history: if item[0] == user_policy_id or item[1] == group_policy_id: total_time += helpers.get_minutes_diff_from_timestamps(item[2], item[3]) return total_time @staticmethod def check_if_resolved_from_events_dict(instance_events, user_id=None): ''' Checks if an incident was resolved or not. :param instance_events: (list) of instance events as dict; not Instance objects :param user_id: user_id of the user who had to have resolved the incident :return: (boolean) True if the incident was resolved; False otherwise ''' last_event = instance_events[-1] if last_event[var_names.event_type] == constants.resolve_event: if user_id is None or (user_id is not None and last_event[var_names.event_by] == user_id): return True return False def was_dispatched_to(self, user_policy_id=None, group_policy_id=None): ''' Checks if an instance was dispatched to a user/group or not. :param user_policy_id: policy ID of the user :param group_policy_id: policy ID of the group :return: (boolean) -> True if it is; False otherwise ''' for item in self.assignees_history: if user_policy_id is not None and group_policy_id is None and item[0] == user_policy_id: return True elif user_policy_id is None and group_policy_id is not None and item[1] == group_policy_id: return True elif user_policy_id is not None and group_policy_id is not None and\ item[0] == user_policy_id and item[1] == group_policy_id: return True return False def was_escalated(self, escalated_by=None): ''' Checks if an instance was escalated or not. :param escalated_by: user_id of the user who escalated :return: (boolean) True if it was; False otherwise ''' for event in self.events: if isinstance(event, EscalateEvent): if escalated_by is None or (escalated_by is not None and event.event_by == escalated_by): return True return False def was_workflow_run(self, workflow_id): ''' Checks if a workflow was run for this instance or not. :param workflow_id: ID of the workflow to check for :return: (boolean) True if the workflow was run; False otherwise ''' for event in self.events: if isinstance(event, RunWorkflowEvent): if event.workflow_id == workflow_id: return True return False def get_new_responders_events(self): ''' Get all the events that are AddRespondersEvent. :return: (list) of AddRespondersEvent ''' new_responders_events = [] for event in self.events: if isinstance(event, AddRespondersEvent): new_responders_events.append(event) return new_responders_events def get_resolver(self): ''' Get the user_id (or name if in display mode) of the last person who updated the instance. :return: (str) user_id (or name) ''' if len(self.events) > 0: if isinstance(self.events[-1], ResolveEvent): return self.events[-1].event_by return None def is_in_open_state(self): ''' Checks if an instance is in OPEN state or not. :return: (boolean) True is it is; False otherwise ''' if self.status == constants.open_state: return True return False def is_in_acknowledged_state(self): ''' Checks if an instance is in ACKNOWLEDGED state or not. :return: (boolean) True is it is; False otherwise ''' if self.status == constants.acknowledged_state: return True return False def was_deprioritized(self, before_timestamp): ''' Checks if the instance was de-prioritized. :param before_timestamp: timestamp before which the event must have occurred :return: (boolean) True if it was; False otherwise ''' if self.task.urgency_level() <= constants.low_urgency: urg_amend_events = [x for x in self.events if isinstance(x, UrgencyAmendmentEvent)] for event in urg_amend_events: if event.event_timestamp <= before_timestamp and event.event_method == constants.internal: return True return False def get_assignee_names(self, policy_maps=None): ''' Get the list of assignee names from the assignee list of an Instance object. :param policy_maps: (dict) -> policy id mapped to Policy object :return: (str) assignee names ''' name_set = set() if self.assignees is not None: for assig in self.assignees: if assig.display_name is not None: name_set.add(assig.display_name) if len(name_set) == 0 and policy_maps is not None: policy_ids = self.for_policy_ids() for id_ in policy_ids: if id_ in policy_maps: name_set.add(policy_maps[id_].policy_name) return ', '.join(name_set) def get_business_service_names(self): ''' Gets the list of names of the business services that have been impacted by this instance. :return: (str) business service names ''' if self.impacted_business_services is None: return '' return ', '.join([x[1] for x in self.impacted_business_services]) @staticmethod def calculate_next_alert_timestamp(level, for_policies, current_time): ''' Gets the timestamp this instance should be re-triggered next. :param level: the level number to check for in the policies :param for_policies: (list) of policies to sort through to find the lowest wait minutes :param current_time: current UTC timestamp :return: (datetime.datetime) object ''' if len(for_policies) > 0: min_level_minutes = min([pol.get_level_minutes(level) for pol in for_policies]) else: min_level_minutes = configuration.standard_wait_minutes return current_time + datetime.timedelta(minutes=min_level_minutes) @staticmethod def instance_was_resolved(instance_events, user_id=None): ''' Checks if an incident was resolved or not. :param instance_events: (list) of instance events as dict; not Instance objects :param user_id: user_id of the user who had to have resolved the incident :return: (boolean) True if the incident was resolved; False otherwise ''' last_event = instance_events[-1] if last_event[var_names.event_type] == constants.resolve_event: if user_id is None or (user_id is not None and last_event[var_names.event_by] == user_id): return True return False def re_trigger(self, re_trigger_time, next_alert_time): ''' Re-trigger this instance. :param re_trigger_time: time to re-trigger the instance on :param next_alert_time: time to send out the next alerts for this instance ''' self.last_alert_timestamp = re_trigger_time self.next_alert_timestamp = next_alert_time def acknowledge(self, next_alert_time): ''' Acknowledges the instance. :param next_alert_time: next time the instance should trigger again ''' self.status = constants.acknowledged_state self.next_alert_timestamp = next_alert_time def escalate(self, event: EscalateEvent, new_assignees): ''' Escalates the instance. :param event: EscalateEvent :param new_assignees: (list) of Assignee objects ''' self.status = constants.open_state self.level = event.escalate_to_level self.last_alert_timestamp = event.event_timestamp self.next_alert_timestamp = event.next_alert_timestamp self.assignees = new_assignees def snooze(self, event: SnoozeEvent): ''' Snoozes the instance. :param event: SnoozeEvent ''' self.status = constants.acknowledged_state self.next_alert_timestamp = event.event_timestamp + datetime.timedelta(minutes=event.snooze_for) def reassign(self, event_time, new_assignees, next_alert_timestamp): ''' Reassign the instance. :param event_time: time the event occurred :param new_assignees: (list) of Assignee objects :param next_alert_timestamp: the next time the instance should be re-triggered ''' self.status = constants.open_state self.last_alert_timestamp = event_time self.next_alert_timestamp = next_alert_timestamp self.assignees = new_assignees def add_responders(self, new_assignees): ''' Adds new responders to the instance. :param new_assignees: (list) of Assignee objects ''' self.assignees = new_assignees def un_acknowledge(self, event_time, next_alert_time): ''' Un-acknowledge this instance. :param event_time: time the event occurred :param next_alert_time: next time the instance should trigger again ''' self.status = constants.open_state self.last_alert_timestamp = event_time self.next_alert_timestamp = next_alert_time def amend_urgency(self, new_urgency): ''' Amend the urgency of this instance. :param new_urgency: the new urgency to amend to ''' self.task.details[var_names.urgency_level] = new_urgency def merge(self, related_task_id): ''' Merge this instance with another instance. :param related_task_id: task ID of the instance this instance is being merged with ''' self.task.details[var_names.to_alert] = False self.task.details[var_names.related_task_id] = related_task_id def un_merge(self): ''' Merge this instance with another instance. ''' self.task.details[var_names.to_alert] = True self.task.details[var_names.related_task_id] = None def add_subscribers(self, new_subscribers: list): ''' Add new subscribers to the instance. :param new_subscribers: (list) of user IDs of the subscribers ''' if self.subscribers is not None: self.subscribers = self.subscribers + helpers.get_int_list(new_subscribers) def remove_subscribers(self, new_subscribers: list): ''' Remove subscribers from the instance. :param new_subscribers: (list) of user IDs of subscribers to remove ''' if self.subscribers is not None: self.subscribers = list(set(self.subscribers).difference(set(helpers.get_int_list(new_subscribers)))) def add_impacted_business_service(self, business_service_id, business_service_name): ''' Add a new impacted business service to the list of impacted business services. :param business_service_id: ID of the business service :param business_service_name: name of the business service ''' if self.impacted_business_services is not None: self.impacted_business_services.append((business_service_id, business_service_name)) def remove_impacted_business_service(self, business_service_id): ''' Remove an impacted business service from the list of impacted business services. :param business_service_id: ID of the business service ''' if self.impacted_business_services is not None: index_to_remove = None for i in range(0, len(self.impacted_business_services)): if self.impacted_business_services[i][0] == business_service_id: index_to_remove = i break if index_to_remove is not None: del self.impacted_business_services[index_to_remove] def sync_with_vendor(self, integration_id, integration_type_id, vendor_id, vendor_url): ''' Sync the instance with an external vendor's incident equivalent. :param integration_id: ID of the integration that was used to sync up :param integration_type_id: ID of the type of integration that was used :param vendor_id: ID of the incident equivalent of the vendor :param vendor_url: url to get to the vendor equivalent incident ''' if self.synced_vendors is None: self.synced_vendors = [] self.synced_vendors.append({ var_names.integration_id: integration_id, var_names.synced_with: integration_type_id, var_names.vendor_id: vendor_id, var_names.vendor_url: vendor_url }) def edit_title(self, new_title): ''' Edit the title of task of the instance. :param new_title: the new title ''' self.task.details[var_names.task_title] = new_title def get_last_note(self): ''' Get the last note that was added to an instance. :return: (string) note ''' if self.events is not None: note_events = [ev for ev in self.events if isinstance(ev, NotateEvent)] if len(note_events) > 0: return note_events[-1].notes elif self.notes is not None and len(self.notes) > 0: return self.notes[-1][var_names.notes] return None def get_trigger_counts(self): ''' Get the total number of triggers and the number of triggers that have happened in the last hour. :return: (tuple) -> total trigger count, last hour trigger count ''' last_hour_start = self.timestamp - datetime.timedelta(hours=1) total_count, last_hour_count = 0, 0 for event in self.events: if isinstance(event, TriggerEvent): total_count += 1 if event.event_timestamp >= last_hour_start: last_hour_count += 1 return total_count, last_hour_count def get_notification_cycles(self, user_policy_id=None): ''' Get the number of free and paid notification cycles that occurred. :param user_policy_id: (optional) if provided then only the notification count of that user is retrieved :return: (tuple) -> free notification cycles, paid notification cycles ''' total_free, total_paid = 0, 0 for item in self.assignees: if user_policy_id is None or item.user_policy_id == user_policy_id: total_free += item.free_notification_count total_paid += item.paid_notification_count # We divide by 2 because in each free cycle there is an APP and email notification, # and in each paid cycle there is a SMS and voice call notification. return total_free/2, total_paid/2 def is_over_notified(self): ''' Checks if an instance has been over-notified or not. This will be used to stop too many notifications going out. :return: (boolean) -> True if it has been over-notified; False otherwise ''' trigger_count, trigger_count_last_hour = self.get_trigger_counts() num_assignees = len(self.assignees) free_ntf_cycles, paid_ntf_cycles = self.get_notification_cycles() free_cycle_subtractor = round(free_ntf_cycles / num_assignees) paid_cycle_subtractor = round(paid_ntf_cycles / num_assignees) remaining_free_ntf_cycles = num_assignees *\ (configuration.max_idle_free_notification_cycle_count - free_cycle_subtractor) remaining_paid_ntf_cycles = num_assignees *\ (configuration.max_idle_paid_notification_cycle_count - paid_cycle_subtractor) if num_assignees > configuration.max_assignees_per_instance: return True if remaining_free_ntf_cycles <= 0 or remaining_paid_ntf_cycles <= 0: return True if trigger_count_last_hour > configuration.max_hourly_triggers: return True inst_hour_duration = (self.timestamp - self.instance_timestamp).seconds/3600 if inst_hour_duration > 3 and\ trigger_count > configuration.max_avg_hourly_triggers_after_three_hours * inst_hour_duration: return True return False def is_assignee_over_notified(self, user_policy_id): ''' This is a less precise check. It checks if a user has been over notified or not, but uses the overall trigger count in the check. :param user_policy_id: (int) policy ID of the user to check for :return: (boolean) True if the user has been over notified; False otherwise ''' trigger_count, trigger_count_last_hour = self.get_trigger_counts() free_ntf_cycles, paid_ntf_cycles = self.get_notification_cycles(user_policy_id=user_policy_id) if (free_ntf_cycles > trigger_count * configuration.max_idle_free_notification_cycle_count) or\ (paid_ntf_cycles > trigger_count * configuration.max_idle_paid_notification_cycle_count): return True return False def is_unresolved_for_too_long(self): ''' Checks if an unresolved incident has been pending for more than allowed. :return: (boolean) True if it is unresolved; False otherwise ''' if self.resolved_on is None and\ (self.timestamp - self.instance_timestamp).days > configuration.max_unresolved_period: return True return False def to_dict(self): ''' Gets the dict of the InstanceState object. :return: dict of InstanceState ''' data = {var_names.instance_id: self.instance_id, var_names.organization_instance_id: self.organization_instance_id, var_names.organization_id: self.organization_id, var_names.instance_timestamp: self.instance_timestamp, var_names.state_timestamp: self.timestamp, var_names.assignee_level: self.level, var_names.last_alert_timestamp: self.last_alert_timestamp, var_names.next_alert_timestamp: self.next_alert_timestamp, var_names.status: self.status, var_names.assignees: [], var_names.task: self.task.to_dict(), var_names.events: [], var_names.notes: [], var_names.status_update: self.updates, var_names.subscribers: self.subscribers, var_names.impacted_business_services: self.impacted_business_services, var_names.conference_bridges: self.conference_bridges, var_names.synced_vendors: self.synced_vendors, var_names.resolved_on: self.resolved_on} if self.assignees is not None: for assignee in self.assignees: if isinstance(assignee, Assignee): data[var_names.assignees].append(assignee.to_dict()) else: # this will be needed when assignees are sent as [[display name, preferred username], ...] # for instance details on the web and the app data[var_names.assignees].append(assignee) for event in self.events: data[var_names.events].append(event.to_dict()) if self.notes is not None: for item in self.notes: data[var_names.notes].append(item) return data