#!/usr/bin/env python3 # By: Riasat Ullah # This class represents an instance monitor. It monitors the lifecycle of an instance and conducts # internal updates of the instance as needed. This does not handle user triggered updates. import sys sys.path.append('/var/www/html/taskcallrest/') from cache_queries import cache_cleaner, cache_task_instances from data_syncers import syncer_organizations, syncer_policies, syncer_services, syncer_task_instances, syncer_users from dbqueries import db_accounts, db_events from modules.alert_logger import AlertLogger from modules.event_allocator import EventAllocator from modules.notice_allocator import NoticeAllocator from notices import incident_notices from objects.events import ResolveEvent from taskcallrest import settings from threading import Thread from utils import app_notice, blacklist_messages, constants, internal_alert_manager, logging, mail, times, var_names from utils.communication_vendors import Twilio from utils.db_connection import CONN_POOL, CACHE_CLIENT import argparse import configuration import datetime import psycopg2 import time class InstanceMonitor(Thread): def __init__(self, conn, cache_client, email_creds, taskcall_numbers, vendor_api_clients, monitor_time=None, monitor_region=None, load_instances_from_db=False): self.conn = conn self.cache_client = cache_client self.email_creds = email_creds self.taskcall_numbers = taskcall_numbers self.vendor_api_clients = vendor_api_clients self.monitor_region = monitor_region self.monitor_time = times.get_current_timestamp() if monitor_time is None else monitor_time self.load_instances_from_db = load_instances_from_db # internal environment variables self.open_instances = [] self.org_permissions = dict() self.org_cost_percents = dict() self.policies = dict() self.services = dict() self.user_info = dict() self.allocator = EventAllocator(None, None, None, None, None) self.notifier = NoticeAllocator() self.ex = None monitor_name = '_'.join(['InstanceMonitor', self.monitor_time.strftime(constants.timestamp_format)]) Thread.__init__(self, name=monitor_name) def set_up_environment(self): logging.info(self.name + ' - Setting up environment') self.open_instances = syncer_task_instances.get_instances_to_monitor( self.conn, self.cache_client, self.monitor_time, force_load_from_db=self.load_instances_from_db) if len(self.open_instances) > 0: org_ids = set() parent_policy_ids = set() user_policy_ids = set() task_service_ids = set() for inst_id in self.open_instances: instance = self.open_instances[inst_id] org_ids.add(instance.organization_id) for assignee in instance.assignees: parent_policy_ids.add(assignee.for_policy_id) user_policy_ids.add(assignee.user_policy_id) serv_id = instance.task.service_id() if serv_id is not None: task_service_ids.add(serv_id) self.org_permissions = syncer_organizations.get_organization_permissions( self.conn, self.cache_client, self.monitor_time, list(org_ids), store_misses=True ) self.org_cost_percents = syncer_organizations.get_organization_month_text_call_costs( self.conn, self.cache_client, list(org_ids), store_misses=True ) self.policies = syncer_policies.get_policies( self.conn, self.cache_client, self.monitor_time, list(parent_policy_ids.union(user_policy_ids)), store_misses=True ) if len(task_service_ids) > 0: self.services = syncer_services.get_multiple_services( self.conn, self.cache_client, self.monitor_time, list(task_service_ids), store_misses=True) self.user_info = syncer_users.get_user_policy_info(self.conn, self.cache_client, self.monitor_time, list(user_policy_ids), store_misses=True) self.allocator = EventAllocator(self.policies, self.services, self.user_info, self.org_permissions, self.org_cost_percents) self.notifier = NoticeAllocator(self.user_info, self.taskcall_numbers, self.vendor_api_clients, self.policies, self.services) def run(self): logging.info(self.name + ' - Running monitor') try: self.set_up_environment() if len(self.open_instances) == 0: logging.info(self.name + ' - No open instances were found for monitoring') return else: logging.info(self.name + ' - Monitoring through open instances - ' + str(len(self.open_instances))) self.allocate_triggers_and_remove_non_alertable_instances() # When instances are escalated, the user info of the new assignee may not be available in # self.user_info. Thus, we have to update self.user_info for the additional users. This must # be done prior to handling the notification alerts because if the notification alerts # cannot find the relevant user info, it will skip the alert dispatch and the dispatch # will be delayed till the next time the monitor is run. # Book escalations here!!!! # Ack, Res and Esc codes for a user for an instance are only created in the database. # That is the only source of truth for the codes. Hence, despite the additional db query # overhead, we have to update the instance in the cache with that in the database to # have the user response codes. Although the event allocator handles escalation with the # InstanceState object, it does so without providing the response codes resulting # in "null" values being sent to the user in SMS and voice calls. Thus, this step must # be implemented. self.process_escalations() for inst_id in self.open_instances: instance = self.open_instances[inst_id] if instance.task.details[var_names.to_alert] and instance.status == constants.open_state: self.allocator.handle_notification_alerts(self.monitor_time, instance) self.process_standard_dispatch() if len(self.allocator.triggers) > 0: db_events.book_trigger_events(self.conn, self.allocator.triggers) if len(self.allocator.dispatches) > 0: db_events.book_dispatch_events(self.conn, self.allocator.dispatches) if len(self.allocator.urgency_amendments) > 0: db_events.book_urgency_amendment_events(self.conn, self.allocator.urgency_amendments, is_sys_action=True) if len(self.allocator.updated_instances) > 0: cache_task_instances.store_multiple_instances(self.cache_client, self.allocator.updated_instances) if len(self.notifier.email_messages) > 0: mail.AmazonSesBulkDispatcher(self.notifier.email_messages, self.email_creds).start() if len(self.notifier.push_notices) > 0: app_notice.NoticeSender(self.notifier.push_notices).start() if len(self.notifier.alert_logs) > 0: AlertLogger(self.conn, self.notifier.alert_logs).start() except psycopg2.InterfaceError as e: logging.error('Error from inner scope') logging.exception(str(e)) self.ex = e except Exception as e: logging.error('Error from inner scope') logging.exception(str(e)) self.ex = e def allocate_triggers_and_remove_non_alertable_instances(self): ''' Identify and allocate required triggers and re-triggers, and remove instances that should not be alerted on like instances that should be blacklisted or those that have been open for too long from self.open_instances. ''' del_inst_ids = [] for inst_id in self.open_instances: instance = self.open_instances[inst_id] # The cache should not have any resolved instances in it. However, by any chance if the cache # is not updated properly, we do a check here to ensure that resolved instances are not processed. if instance.task.details[var_names.to_alert] and instance.status != constants.resolved_state: # handle level timeouts if instance.next_alert_timestamp <= self.monitor_time: instance = self.allocator.handle_level_timeout(self.monitor_time, instance) # handle notification alerts # they are only sent out if the instance is in "open" state (un-acknowledged) if instance.status == constants.open_state: if instance.is_over_notified(): if instance.is_unresolved_for_too_long(): res_event = ResolveEvent(inst_id, self.monitor_time, constants.internal) syncer_task_instances.resolve(self.conn, self.cache_client, res_event, instance.organization_id, is_sys_action=True) else: if not syncer_organizations.is_instance_blacklisted( self.conn, self.cache_client, self.monitor_time, instance.organization_id, inst_id ): syncer_task_instances.enable_instance_alerting( self.conn, self.cache_client, self.monitor_time, instance.organization_id, inst_id, to_enable=False ) err_msg = 'Instance ID - ' + str(inst_id) + ': ' + \ blacklist_messages.blmsg_over_notified_instance logging.error(err_msg) syncer_organizations.store_blacklisted_event( self.conn, self.cache_client, self.monitor_time, instance.organization_id, err_msg, inst_id ) # Check if organization is blacklisted. We run this check here # (without conditions) to ensure that the owner of the account is notified as # soon as the account is actually blacklisted syncer_organizations.is_organization_blacklisted( self.conn, self.cache_client, self.monitor_time, instance.organization_id) logging.info('Removing blacklisted instance - ' + str(inst_id)) cache_task_instances.remove_instance(self.cache_client, inst_id) del_inst_ids.append(inst_id) for inst_id in del_inst_ids: del self.open_instances[inst_id] def process_escalations(self): ''' Handle all the escalations, book them in the database and update the cache. ''' if len(self.allocator.escalations) > 0: new_user_pol_ids_to_fetch = [] final_escalations = [] esc_assignees = syncer_task_instances.bulk_system_escalate( self.conn, self.cache_client, self.monitor_time, self.allocator.escalations ) for item in self.allocator.escalations: esc_inst_id = item[0].instance_id if esc_inst_id in esc_assignees: instance = self.open_instances[esc_inst_id] instance.assignees = esc_assignees[esc_inst_id] new_esc_alloc = (item[0], esc_assignees[esc_inst_id]) final_escalations.append(new_esc_alloc) for new_assignee in esc_assignees[esc_inst_id]: if new_assignee.user_policy_id not in self.user_info: new_user_pol_ids_to_fetch.append(new_assignee.user_policy_id) self.allocator.escalations = final_escalations if len(new_user_pol_ids_to_fetch) > 0: self.user_info = {**self.user_info, **syncer_users.get_user_policy_info( self.conn, self.cache_client, self.monitor_time, new_user_pol_ids_to_fetch, store_misses=True )} self.allocator.user_info = self.user_info self.notifier.user_info = self.user_info def process_standard_dispatch(self): ''' Dispatch notifications through the standard channels - email, push notification, SMS and voice call. ''' for event in self.allocator.dispatches: instance = self.open_instances[event.instance_id] # prepare the message to be sent out inst_service_name = None if instance.task.service_id() is None \ else self.services[instance.task.service_id()].service_name u_pid = event.user_policy_id() u_tz = self.user_info[u_pid][var_names.timezone] u_lang = self.user_info[u_pid][var_names.language] if event.event_method == constants.email: subject, message = incident_notices.dispatch_email_content( u_lang, instance.organization_instance_id, times.utc_to_region_time(instance.instance_timestamp, u_tz), instance.get_assignee_names(self.policies), inst_service_name, instance.task.title(), instance.task.text_msg(), instance.task.urgency_level(), instance.get_business_service_names(), u_tz ) self.notifier.handle_bulk_email_dispatch(subject, message, [u_pid], instance_id=instance.instance_id, organization_id=instance.organization_id) elif event.event_method == constants.app: push_title, push_message = incident_notices.dispatch_push_notice_content( u_lang, instance.organization_instance_id, times.utc_to_region_time(instance.instance_timestamp, u_tz), instance.task.title()) self.notifier.handle_app_dispatch(push_title, push_message, policy_id=u_pid, instance_id=instance.instance_id, organization_id=instance.organization_id, is_critical=True if instance.task.urgency_level() in [ constants.high_urgency, constants.critical_urgency] else False) elif event.event_method == constants.text: phone_message = incident_notices.dispatch_sms_content( u_lang, instance.organization_instance_id, instance.task.title(), event.assignee.acknowledgement_code, event.assignee.resolution_code, event.assignee.escalation_code ) self.notifier.handle_text_dispatch(u_pid, phone_message, instance.instance_id, organization_id=instance.organization_id) elif event.event_method == constants.call: phone_message = incident_notices.dispatch_voice_content( u_lang, instance.organization_instance_id, instance.task.title(), event.assignee.acknowledgement_code, event.assignee.resolution_code, event.assignee.escalation_code ) self.notifier.handle_voice_dispatch( u_pid, phone_message, instance.instance_id, u_lang, organization_id=instance.organization_id ) def join(self): Thread.join(self) # Since join() returns in caller thread we re-raise the caught exception if any was caught if self.ex: raise self.ex if __name__ == "__main__": arg_parser = argparse.ArgumentParser() arg_parser.add_argument('--monitor_time', default=None) arg_parser.add_argument('--region', default=None) arg_parser.add_argument('--dont_switch_to_current_time', action='store_true') args = arg_parser.parse_args() start_time = args.monitor_time region = args.region dont_switch_to_current_time = args.dont_switch_to_current_time if start_time is not None: assert (isinstance(start_time, datetime.datetime) or isinstance(start_time, str)) if type(start_time) is str: start_time = datetime.datetime.strptime(start_time, constants.timestamp_format) else: start_time = times.get_current_timestamp() # get a DB connection monitor_conn = CONN_POOL.get_db_conn() monitor_cache = CACHE_CLIENT monitor_date = start_time.date() # get the email account credentials email_credentials = mail.AmazonSesCredentials() # pre text of error message for internal alerting pre_error_title = 'Instance Monitor (' + settings.REGION + ')' # get vendor api clients vendor_clients = {constants.twilio: Twilio(Twilio.get_authentication())} # get phone numbers owned by taskcall across countries vendor_numbers = db_accounts.get_vendor_phones(monitor_conn, start_time) # retrieve organization permissions from the database and store them in cache syncer_organizations.store_organization_permissions_in_cache(monitor_conn, monitor_cache, start_time) # wait time in seconds before next run, refresh minutes for how often text/call usage costs # should be refreshed and cleanup minutes for how often the cache should be cleaned up wait_seconds = 8 refresh_minutes = 12 cleanup_minutes = 30 # We set the last_refresh earlier than start time so that a # refresh happens immediately upon starting the monitor stop = False timestamp = start_time force_instance_load_from_db = True last_refresh = timestamp - datetime.timedelta(minutes=refresh_minutes) last_cleanup = timestamp # Clean the open instances cache completely cache_task_instances.remove_all_instances(monitor_cache) # Clean and reload the blacklisted organizations cache syncer_organizations.store_all_blacklisted_events_in_cache( monitor_conn, monitor_cache, timestamp - datetime.timedelta(days=configuration.blacklist_look_back_period)) while not stop: # If the date changes then get the latest dispatch layers. if timestamp.date() != monitor_date: monitor_date = timestamp.date() email_credentials = mail.AmazonSesCredentials() vendor_numbers = db_accounts.get_vendor_phones(monitor_conn, timestamp) vendor_clients = {constants.twilio: Twilio(Twilio.get_authentication())} try: current_monitor = InstanceMonitor(monitor_conn, monitor_cache, email_credentials, vendor_numbers, vendor_clients, monitor_time=timestamp, monitor_region=region, load_instances_from_db=force_instance_load_from_db) current_monitor.start() current_monitor.join() time.sleep(wait_seconds) if dont_switch_to_current_time: timestamp = timestamp + datetime.timedelta(seconds=wait_seconds) else: timestamp = times.get_current_timestamp() # It is advisable not to make changes to the cache while the thread is still alive. # However, since we wait for a few seconds before moving to the next step, it is okay # to not check if the thread is alive because the db and cache is queried right at the beginning # when the environment for the monitor is set up, which runs really fast. # Refresh the monthly text and call cost percentage stored in the cache every 12 minutes if (timestamp - last_refresh).seconds/60 > refresh_minutes: syncer_organizations.refresh_organization_month_text_call_costs(monitor_conn, monitor_cache) last_refresh = timestamp force_instance_load_from_db = True else: force_instance_load_from_db = False # Clean the cache every 30 minutes if (timestamp - last_cleanup).seconds/60 > cleanup_minutes: cache_cleaner.clean_cached_organization_month_costs(monitor_cache, list(current_monitor.org_cost_percents.keys())) cache_cleaner.clean_cached_policies(monitor_cache, list(current_monitor.policies.keys())) cache_cleaner.clean_cached_services(monitor_cache, list(current_monitor.services.keys())) cache_cleaner.clean_cached_user_info(monitor_cache, list(current_monitor.user_info.keys())) cache_cleaner.clean_cached_workflows(monitor_cache, list(current_monitor.org_cost_percents.keys())) except psycopg2.InterfaceError as e: logging.error('Outer scope - possible connection error') logging.exception(str(e)) internal_alert_manager.dispatch_alerts(pre_error_title + ' - Connection Error', str(e)) try: CONN_POOL.put_db_conn(monitor_conn) except Exception as e: logging.exception(str(e)) finally: logging.info('Trying to open a new connection') monitor_conn = CONN_POOL.get_db_conn() sys.exit(1) except Exception as e: logging.info('Outer scope - unknown error') logging.exception(str(e)) internal_alert_manager.dispatch_alerts(pre_error_title + ' - Unknown Error', str(e)) sys.exit(1)