# By: Riasat Ullah # This file contains functions that will help route tasks/instances according to # custom conditional routing rules, and default de-duplication and urgency amending rules. from cache_queries import cache_task_instances from data_syncers import syncer_routing, syncer_organizations, syncer_services, syncer_task_instances from dbqueries import db_task_instances from exceptions.user_exceptions import MaliciousSource, ServiceUnavailable from modules.alert_logger import AlertLogger from modules import custom_action_manager from modules.notice_allocator import NoticeAllocator from modules.workflow_manager import WorkflowManager from objects.events import ResolveEvent from objects.task_payload import TaskPayload from taskcallrest import settings from threading import Thread from utils import app_notice, constants, errors, key_manager, logging, mail, permissions, times, var_names from uuid import UUID import configuration as configs import datetime class Router(Thread): def __init__(self, conn, cache_client, payload: TaskPayload, routing_conditions=None, org_permission=None, return_queue=None): self.conn = conn self.cache_client = cache_client self.payload = payload self.org_permission = org_permission if org_permission is not None\ else syncer_organizations.get_single_organization_permission(conn, cache_client, payload.timestamp, payload.organization_id) self.task_service = None self.return_queue = return_queue self.routing_conditions = routing_conditions self.additional_policies = [] self.org_business_services = dict() Thread.__init__(self) def run(self): try: # do not create an incident if the organization is blacklisted if syncer_organizations.is_organization_blacklisted( self.conn, self.cache_client, self.payload.timestamp, self.payload.organization_id): raise MaliciousSource(errors.err_org_malicious_activity + ' - ' + str(self.payload.organization_id)) # checking for routing permissions if permissions.has_org_permission(self.org_permission, permissions.ORG_ROUTING_PERMISSION): self.routing_conditions = self.routing_conditions if self.routing_conditions is not None\ else syncer_routing.get_organization_enabled_conditional_routes( self.conn, self.cache_client, self.payload.timestamp, self.payload.organization_id) else: self.routing_conditions = [] # handle all service related function here if self.payload.service_id is not None or self.payload.service_ref_id is not None: self.task_service = syncer_services.get_single_service( self.conn, self.cache_client, self.payload.timestamp, service_id=self.payload.service_id, service_ref_id=self.payload.service_ref_id ) if self.task_service is None: raise MaliciousSource('Unknown service - ' + str(self.payload.organization_id) + ' | ' + str(self.payload.service_id)) else: self.payload.handle_service_level_processes(self.conn, self.task_service) # check for conditional routing (must be called after Service details have been loaded and processed) if self.org_permission is not None and\ permissions.has_org_permission(self.org_permission, permissions.ORG_ROUTING_PERMISSION): self.handle_conditional_routing() # check for de-duplication if self.payload.dedup_key is not None: active_task_ids = db_task_instances.get_active_instance_task_ids_with_dedup_key( self.conn, self.payload.timestamp, self.payload.organization_id, self.payload.dedup_key ) if len(active_task_ids) > 0: self.payload.dedup_alert(active_task_ids[0]) self.additional_policies = self.payload.get_additional_policies(self.conn, self.cache_client) # check if intelligent grouping applies if (self.task_service is None or (self.task_service is not None and self.task_service.allow_grouping)) and\ self.payload.status != constants.grouped_state and self.org_permission is not None and\ permissions.has_org_permission( self.org_permission, permissions.ORG_INTELLIGENT_GROUPING_PERMISSION): self.payload.intelligent_grouping(self.conn, for_minutes=15, additional_policy_ids=self.additional_policies) # Check for impacted business services. This is done at the end to make sure that all # suppressing and grouping effects have been handled first. if self.task_service is not None and self.task_service.business_services is not None and\ self.org_permission is not None and permissions.has_org_permission( self.org_permission, permissions.ORG_BUSINESS_SERVICES_PERMISSION): self.payload.check_for_impacted_business_services(self.conn, self.task_service.business_services) # Create the task and instance(if to be instantiated) inst_id = self.create_task() if not self.payload.to_instantiate and self.payload.resolve_incident_hours is not None: max_timestamp = self.payload.timestamp - datetime.timedelta(hours=self.payload.resolve_incident_hours) pnd_inst_ids = db_task_instances.get_organization_pending_instance_ids( self.conn, self.payload.organization_id, max_timestamp) for pnd_id in pnd_inst_ids: event = ResolveEvent(pnd_id, self.payload.timestamp, constants.internal) syncer_task_instances.resolve( self.conn, self.cache_client, event, org_id=self.payload.organization_id, is_sys_action=True, org_perm=self.org_permission ) # Only send the chat notifications, business services impacted notifications and run workflows # if an instance was created and not automatically resolved by a conditional routing. # Do NOT change this logic. if inst_id is not None and self.payload.to_instantiate and self.payload.status != constants.resolved_state: # send impacted business service notifications if self.task_service is not None and permissions.has_org_permission( self.org_permission, permissions.ORG_BUSINESS_SERVICES_SUBSCRIPTION_PERMISSION): self.queue_up_business_impact_notifications(inst_id) self.send_conference_bridge_notifications(inst_id) # handle custom actions if self.task_service is not None: # Make sure you do not end up in an ongoing loop of incidents being created by vendor and # issues being created by TaskCall. Do not create vendor issue if the incident was created # from the same integration. integ_info = [x for x in self.task_service.service_integrations if x[var_names.integration_type] in configs.allowed_auto_triggering_integrations and x[var_names.integration_id] != self.payload.integration_id] \ if self.task_service.service_integrations is not None else [] if len(integ_info) > 0: inst_obj = db_task_instances.get_instances( self.conn, self.payload.timestamp, inst_id)[inst_id] for item in integ_info: try: integ_key = key_manager.conceal_reference_key(UUID(item[var_names.integration_key])) custom_action_manager.execute_custom_action( self.conn, self.cache_client, self.payload.timestamp, self.payload.organization_id, self.org_permission, inst_id, inst_obj, item, integ_key, None, False, self.payload.dedup_key, constants.lang_en, False, False, True, trigger_info=self.payload.get_displayable_payload() ) except Exception as e: logging.info(str(e)) logging.info('Continuing on...') if settings.CACHE_ON: upd_inst_obj = db_task_instances.get_instances( self.conn, times.get_current_timestamp(), inst_id)[inst_id] cache_task_instances.store_single_instance(self.cache_client, upd_inst_obj) # checking for workflow permissions and execute the ones that should be executed if permissions.has_org_permission(self.org_permission, permissions.ORG_WORKFLOWS_PERMISSION): WorkflowManager(self.conn, self.cache_client, self.payload.organization_id, self.org_permission, inst_id, constants.trigger_event).execute_workflow() if self.return_queue is not None: self.return_queue.put(inst_id) # This will only be needed for the Incidents API. Running the thread as it should be with Thread.start() # will not return the value. However, if the run() method is called directly, then a new thread will # not be created (it will run within the main Thread) and return the instance ID. return inst_id except MaliciousSource as e: logging.error(str(e)) except ServiceUnavailable as e: logging.warning(str(e)) except Exception as e: raise Exception(str(e)) def handle_conditional_routing(self): # effect only the very first conditional routing rule that applies if self.routing_conditions is not None: for route in self.routing_conditions: if self.payload.qualifies_by_time(route) and self.payload.qualifies_by_conditions(route): re_route_srv_id = route.get_re_routing_service_id() new_task_service = None if re_route_srv_id is not None: new_task_service = syncer_services.get_single_service( self.conn, self.cache_client, self.payload.timestamp, service_id=re_route_srv_id) if new_task_service is not None: self.task_service = new_task_service self.payload.process_actions(self.conn, route, new_task_service, self.org_permission) if not route.allow_multiple: break def create_task(self): inst_id = syncer_task_instances.create_task( conn=self.conn, client=self.cache_client, timestamp=self.payload.timestamp, organization_id=self.payload.organization_id, start_date=self.payload.start_date, title=self.payload.title, timezone=self.payload.timezone, task_time=self.payload.task_time, service_id=self.payload.service_id, service_policy_id=self.payload.service_policy_id, assignees=self.additional_policies, created_by=self.payload.created_by, repeat_on=self.payload.repeat, text_msg=self.payload.description, urgency_level=self.payload.urgency_level, trigger_method=self.payload.trigger_method, trigger_info=self.payload.get_displayable_payload(), instantiate=self.payload.to_instantiate, alert=self.payload.to_alert, integration_id=self.payload.integration_id, api_key_id=self.payload.api_key_id, routing_id=self.payload.routing_id, related_task_id=self.payload.related_task_id, task_status=self.payload.status, notes=self.payload.notes, tags=self.payload.tags, dedup_key=self.payload.dedup_key, amended_urgency=self.payload.amended_urgency, next_alert_timestamp=self.payload.next_alert_timestamp, status_update=self.payload.status_update, subscribers=self.payload.subscribers, conference_bridge=self.payload.conference_bridge, impacted_business_services=self.payload.get_impacted_business_service_tuples() )[1] return inst_id def queue_up_business_impact_notifications(self, instance_id): ''' Queue up the notifications that should be sent to business service subscribers for the business services that have been impacted by this incident as per the minimum urgency level set on them. :param instance_id: instance ID ''' if self.payload.impacted_business_services is not None: to_update = False for item in self.payload.impacted_business_services: subscriber_count, page_id = item[2], item[3] if subscriber_count > 0 or page_id is not None: to_update = True break if to_update: syncer_task_instances.store_pending_instance_update( self.cache_client, self.payload.timestamp, self.payload.organization_id, instance_id, constants.trigger_event ) def send_conference_bridge_notifications(self, instance_id): ''' Send the notifications to business service subscribers for the business services that have been impacted by this incident as per the minimum urgency level set on then. :param instance_id: instance ID ''' if self.payload.conference_bridge is not None: notifier = NoticeAllocator() notifier = syncer_task_instances.add_conference_bridge( self.conn, self.cache_client, self.payload.timestamp, notifier, self.payload.organization_id, instance_id, self.payload.conference_bridge, constants.internal, is_sys_action=True) if len(notifier.email_messages) > 0: mail.AmazonSesBulkDispatcher(notifier.email_messages).start() if len(notifier.push_notices) > 0: app_notice.NoticeSender(notifier.push_notices).start() if len(notifier.alert_logs) > 0: AlertLogger(self.conn, notifier.alert_logs).start()