#!/usr/bin/env python3 # By: Riasat Ullah # This class represents the business impact monitor job. It monitors business impact related updates and notifications # that need be dispatched for them. import sys sys.path.append('/var/www/html/taskcallrest/') from cache_queries import cache_status_pages, cache_task_instances from data_syncers import syncer_status_pages, syncer_users from dbqueries import db_business_services, db_task_instances from dbqueries.status_pages import db_status_page_posts from modules.alert_logger import AlertLogger from modules.notice_allocator import NoticeAllocator from modules.router import Router from modules import status_page_notifier from objects.task_payload import TaskPayload from taskcallrest import settings from threading import Thread from translators import label_translator as _lt from utils import app_notice, constants, info, internal_alert_manager, label_names as lnm, logging, mail, \ times, var_names from utils.db_connection import CACHE_CLIENT, CONN_POOL import argparse import configuration import datetime import psycopg2 import time class BusinessImpactMonitor(Thread): def __init__(self, conn, cache_client, monitor_time, last_run, email_creds, load_from_db=False, check_for_auto_update=False): self.conn = conn self.cache_client = cache_client self.monitor_time = monitor_time self.last_run = last_run self.email_creds = email_creds self.load_from_db = load_from_db self.check_for_auto_update = check_for_auto_update # internal environment variables self.updated_instances = [] self.updated_posts = [] self.incidents = dict() self.policy_ids = [] self.user_info = dict() self.business_updates = [] self.status_page_updates = dict() self.notifier = NoticeAllocator() self.ex = None Thread.__init__(self) def set_up_environment(self): logging.info('Setting up the environment...') if self.load_from_db: logging.info('Fetching business and status page updates from the database...') # Get the details of instance and posts queued up for notification updates. if settings.CACHE_ON: if not self.load_from_db: self.updated_instances = cache_task_instances.get_all_pending_instance_updates(self.cache_client) self.updated_posts = cache_status_pages.get_all_pending_status_page_updates(self.cache_client) else: self.load_from_db = True # Get the details of business services and associated instances that should be reviewed for updates. if len(self.updated_instances) > 0 or self.load_from_db: inst_ids_to_filter_by = None if self.load_from_db \ else [x[var_names.instance_id] for x in self.updated_instances] self.business_updates = db_business_services.get_business_impact_notification_details( self.conn, self.last_run, self.monitor_time, inst_ids_to_filter_by ) self.extract_data_from_business_updates() if len(self.policy_ids) > 0: self.user_info = syncer_users.get_user_policy_info(self.conn, self.cache_client, self.monitor_time, list(self.policy_ids), store_misses=True) # Get details of the posts that should be reviewed for updates. if len(self.updated_posts) > 0 or self.load_from_db: self.status_page_updates = db_status_page_posts.get_posts_for_monitoring( self.conn, self.monitor_time, self.updated_posts, check_for_auto_update=self.check_for_auto_update ) def run(self): ''' Executes several actions: 1) Notifies business service subscribers (internal) about updates to impacted business services. 2) Create posts on status pages automatically (where allowed) for incidents. 3) Notifies instance subscribers about updates to instances. Whether a business service is impacted by an incident or not, status updates of the incident have to be dispatched to instance subscribers. This is why it is handled differently from business subscribers. 4) Send notifications to status page subscribers for new updates. 5) Update maintenance posts where auto update is enabled when after maintenance begins. ''' logging.info('Running Business Impact Monitor') try: self.set_up_environment() logging.info('Instances with updates - ' + str(len(self.business_updates))) if len(self.business_updates) > 0: self.send_updates_to_business_subscribers() self.process_status_page_auto_posts() self.send_status_updates_to_instance_subscribers() logging.info('Posts with updates - ' + str(len(self.status_page_updates))) if len(self.status_page_updates) > 0 or self.load_from_db: self.send_status_page_notifications() if self.check_for_auto_update: self.update_status_page_maintenance() self.notify_about_overdue_posts() 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() if settings.CACHE_ON: if len(self.updated_instances) > 0: cache_task_instances.remove_pending_instance_updates( self.cache_client, 0, len(self.updated_instances)) if len(self.updated_posts) > 0: cache_status_pages.remove_pending_status_page_updates( self.cache_client, 0, len(self.updated_posts)) 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 send_updates_to_business_subscribers(self): ''' Send notifications to business service subscribers. ''' upd_sent_count = 0 for item in self.business_updates: imp_start = item[var_names.start_timestamp] imp_end = item[var_names.end_timestamp] bus_name = item[var_names.business_service_name] bus_subscribers = item[var_names.subscribers] bus_incidents = item[var_names.incidents] org_id = item[var_names.organization_id] is_impacted = True if self.last_run < imp_start <= self.monitor_time else False has_been_resolved = True if self.last_run < imp_end <= self.monitor_time else False # 1) impacted and resolved soon after: do not do anything 2) new impact 3) resolved # 4) status update if is_impacted and has_been_resolved: continue elif is_impacted: bus_event = constants.trigger_event elif has_been_resolved: bus_event = constants.resolve_event else: found_updates = False for inc in bus_incidents: if inc[var_names.status_update] is not None: for inc_upd in inc[var_names.status_update]: if self.last_run < inc_upd[var_names.timestamp] <= self.monitor_time: found_updates = True if found_updates: bus_event = constants.status_update_event else: continue if bus_subscribers is not None: for p_id in bus_subscribers: if p_id in self.user_info: sub = self.user_info[p_id] self.notifier.handle_impacted_business_services_dispatch( sub[var_names.language], bus_name, bus_event, imp_start, imp_end, bus_incidents, sub[var_names.timezone], email_addr=sub[var_names.email], tokens=sub[var_names.push_token], user_id=sub[var_names.user_id], organization_id=org_id, update_after_time=self.last_run ) upd_sent_count += 1 logging.info('Business subscriber updates sent - ' + str(upd_sent_count)) def process_status_page_auto_posts(self): ''' Create posts on status pages automatically from new incidents, incident resolution, status updates and changes in impacted business services. Auto posts are only made on pages that allow it. Even if post is synced up with an incident, but the status page does not allow auto posting, auto posts will not be made. ''' posts_to_publish = [] for inst_id in self.incidents: inc = self.incidents[inst_id] org_inst_id = inc[var_names.organization_instance_id] resolved_on = inc[var_names.resolved_on] org_id = inc[var_names.organization_id] task_title = inc[var_names.task_title] text_msg = inc[var_names.text_msg] bus_details = inc[var_names.business_services] updates = inc[var_names.status_update] posts = inc[var_names.posts] auto_post_pages = inc[var_names.status_pages] was_resolved = True if resolved_on is not None else False if auto_post_pages is None: continue if posts is None: # New incident for pg_ref_id in auto_post_pages: if not was_resolved: imp_bus_sts = self.get_impacted_business_data_struct(bus_details, False) posts_to_publish.append(( org_id, pg_ref_id, constants.incident, task_title, text_msg, constants.investigating_state, constants.minor_impact, imp_bus_sts, None, org_inst_id) ) else: for post_item in posts: pst_id = post_item[var_names.post_id] pst_last_update_time = post_item[var_names.last_update] max_pending_time = post_item[var_names.pending] post_det = None if pst_id is not None: for upd_pst_id in self.status_page_updates: if upd_pst_id == pst_id: post_det = self.status_page_updates[pst_id] break if post_det is None: continue post_bus_statuses = post_det[var_names.impacted_business_services] post_status = post_det[var_names.status] post_pg_imp = post_det[var_names.page_impact] pg_ref_id = post_det[var_names.page_ref_id] pg_lang = post_det[var_names.language] # Check if any of the pending events occurred between the time the update happened and now. # If they did, then do not process the update anymore as it would imply that it has already # been processed. if pg_ref_id in auto_post_pages and post_det is not None: # Status updates if updates is not None: imp_bus_sts = self.get_impacted_business_data_struct(bus_details, False) for upd in updates: if upd[var_names.timestamp] > pst_last_update_time and ( max_pending_time is None or upd[var_names.timestamp] > max_pending_time): posts_to_publish.append(( org_id, pg_ref_id, constants.status_update, task_title, upd[var_names.status_update], post_status, post_pg_imp, imp_bus_sts, pst_id, org_inst_id) ) # if: resolved else: changes in impacted business services if was_resolved and (max_pending_time is None or resolved_on > max_pending_time): imp_bus_sts = self.get_impacted_business_data_struct(bus_details, True) posts_to_publish.append(( org_id, pg_ref_id, constants.status_update, task_title, _lt.get_label(info.msg_incident_resolved, pg_lang), constants.resolved_state, post_pg_imp, imp_bus_sts, pst_id, org_inst_id) ) else: imp_bus_sts = self.get_impacted_business_data_struct( bus_details, False, max_pending_time, post_bus_statuses) if len(imp_bus_sts) > 0: posts_to_publish.append(( org_id, pg_ref_id, constants.status_update, task_title, _lt.get_label(info.msg_impacted_services_updated, pg_lang), post_status, post_pg_imp, imp_bus_sts, pst_id, org_inst_id) ) logging.info('Auto posting - ' + str(len(posts_to_publish))) if len(posts_to_publish) > 0: for pub in posts_to_publish: syncer_status_pages.create_status_page_post( self.conn, self.cache_client, self.monitor_time, pub[0], pub[1], pub[2], pub[3], pub[4], pub[5], pub[6], next_update_time=None, notify_subscribers=False, impacted_components=pub[7], post_id=pub[8], org_inst_id=pub[9] ) def send_status_updates_to_instance_subscribers(self): ''' Send status updates to instance subscribers. An update is also sent when the incident is resolved. ''' to_query = False if settings.CACHE_ON: filtered_ids = [x[var_names.instance_id] for x in self.updated_instances if x[var_names.event_type] == constants.status_update_event] if len(filtered_ids) > 0: to_query = True else: filtered_ids = None to_query = True if to_query: upd_details = db_task_instances.get_instance_status_update_details( self.conn, self.last_run, self.monitor_time, filtered_ids) subscriber_user_info = dict() all_subscriber_pol_ids = [] for item in upd_details: if item[var_names.subscribers] is not None: all_subscriber_pol_ids += item[var_names.subscribers] if len(all_subscriber_pol_ids) > 0: subscriber_user_info = syncer_users.get_user_policy_info( self.conn, self.cache_client, self.monitor_time, list(all_subscriber_pol_ids), store_misses=True) logging.info('Instances with subscribers - ' + str(len(upd_details))) for item in upd_details: inst_id = item[var_names.instance_id] org_inst_id = item[var_names.organization_instance_id] org_id = item[var_names.organization_id] tsk_title = item[var_names.task_title] inst_updates = item[var_names.status_update] if item[var_names.subscribers] is not None: inst_subscribers = [subscriber_user_info[p_id] for p_id in item[var_names.subscribers] if p_id in subscriber_user_info] new_statuses = [] if inst_updates is None: if item[var_names.resolved_on] is not None: new_statuses.append((item[var_names.resolved_on], None, True)) else: for upd in inst_updates: new_statuses.append((upd[var_names.timestamp], upd[var_names.status_update], False)) for sts_up in new_statuses: self.notifier.handle_status_update_dispatch( inst_id, org_inst_id, sts_up[0], tsk_title, sts_up[1], inst_subscribers, organization_id=org_id, is_resolution_update=sts_up[2] ) def send_status_page_notifications(self): ''' Send post updates to status page subscribers. ''' for pst_id in self.status_page_updates: item = self.status_page_updates[pst_id] if item[var_names.events] is not None and item[var_names.subscribers] is not None: for evn in item[var_names.events]: if self.last_run < evn[var_names.timestamp] <= self.monitor_time: status_page_notifier.send_post_updates_to_subscribers( item[var_names.language], item[var_names.subscribers], item[var_names.page_name], item[var_names.url], item[var_names.logo_url], evn[var_names.timestamp], evn[var_names.event_type], evn[var_names.status], evn[var_names.title], evn[var_names.message], evn[var_names.impacted_business_services] ) def update_status_page_maintenance(self): ''' Auto update maintenance posts that have been configured to auto update. ''' for pst_id in self.status_page_updates: post = self.status_page_updates[pst_id] if post[var_names.is_maintenance]: mnt_start = post[var_names.maintenance_start] mnt_end = post[var_names.maintenance_end] mnt_sts = post[var_names.status] mnt_imp = post[var_names.impacted_business_services] mnt_lang = post[var_names.language] to_update, upd_message = False, post[var_names.text_msg] if self.last_run < mnt_start <= self.monitor_time and mnt_sts == constants.scheduled_state: upd_message = _lt.get_label(info.msg_maintenance_in_progress, mnt_lang) to_update = True elif self.last_run < mnt_end <= self.monitor_time and mnt_sts != constants.completed_state: upd_message = _lt.get_label(info.msg_maintenance_completed, mnt_lang) to_update = True if to_update: syncer_status_pages.create_status_page_post( self.conn, self.cache_client, self.monitor_time, post[var_names.organization_id], post[var_names.page_ref_id], constants.maintenance, post[var_names.title], upd_message, constants.status_update, post[var_names.page_impact], next_update_time=None, notify_subscribers=False, impacted_components=mnt_imp, post_id=post[var_names.post_id], maintenance_start=mnt_start, maintenance_end=mnt_end ) def notify_about_overdue_posts(self): ''' Auto update maintenance posts that have been configured to auto update. ''' for pst_id in self.status_page_updates: post = self.status_page_updates[pst_id] max_pnd_hours = post[var_names.max_pending_hours] if max_pnd_hours is not None: post_time = post[var_names.created_on] pg_post_id = post[var_names.page_post_id] pg_name = post[var_names.page_name] pg_lang = post[var_names.language] duration_hours = int((self.monitor_time - post_time).total_seconds()/3600) if duration_hours > max_pnd_hours: inc_title = ' '.join([ _lt.get_label(lnm.det_post, pg_lang), '#' + str(pg_post_id), _lt.get_label(lnm.dsm_pending_for, pg_lang), str(duration_hours), '(' + _lt.get_label(lnm.ttl_status_page, pg_lang) + ':', pg_name + ')'] ) payload = TaskPayload( self.monitor_time, post[var_names.organization_id], self.monitor_time.date(), inc_title, configuration.standard_timezone, self.monitor_time.time(), text_msg=inc_title, urgency_level=constants.medium_urgency, trigger_method=constants.internal, assignees=[] ) Router(self.conn, self.cache_client, payload).start() def extract_data_from_business_updates(self): ''' Business impact data retrieved from the database is retrieved in reference to business services. This function extracts the data and rearranges in reference to incidents, so it can be used for handling auto posts. Synced up post IDs are also extracted. ''' for item in self.business_updates: if item[var_names.subscribers] is not None: self.policy_ids += item[var_names.subscribers] org_id = item[var_names.organization_id] bus_page_ref_ids = item[var_names.status_pages] bus_incidents = item[var_names.incidents] bus_id = item[var_names.business_service_id] bus_ref_id = item[var_names.business_service_ref_id] bus_details = { var_names.business_service_id: bus_id, var_names.business_service_ref_id: bus_ref_id, var_names.start_timestamp: item[var_names.start_timestamp], var_names.end_timestamp: item[var_names.end_timestamp] } if bus_page_ref_ids is not None and bus_incidents is not None: for inc in bus_incidents: inst_id = inc[var_names.instance_id] inst_time = inc[var_names.instance_timestamp] inst_posts = inc[var_names.posts] if inst_id in self.incidents: if bus_ref_id not in self.incidents[inst_id][var_names.business_services]: self.incidents[inst_id][var_names.business_services].append(bus_details) else: self.incidents[inst_id] = { var_names.organization_id: org_id, var_names.instance_timestamp: inst_time, var_names.instance_id: inst_id, var_names.organization_instance_id: inc[var_names.organization_instance_id], var_names.task_title: inc[var_names.task_title], var_names.text_msg: inc[var_names.text_msg], var_names.urgency_level: inc[var_names.urgency_level], var_names.status: inc[var_names.status], var_names.business_services: [bus_details], var_names.resolved_on: inc[var_names.resolved_on], var_names.status_update: inc[var_names.status_update], var_names.status_pages: bus_page_ref_ids, var_names.posts: inst_posts } if inst_posts is not None: for post_item in inst_posts: if post_item[var_names.post_id] is not None: self.updated_posts.append(post_item[var_names.post_id]) def get_impacted_business_data_struct(self, bus_details, was_resolved, max_pending_time=None, skip_details=None): ''' Get the details of impacted business services in the format expected for new events. :param bus_details: (list of dict) of impacted business service details :param was_resolved: (boolean) True if incident was resolved; False otherwise :param max_pending_time: (datetime.datetime) the datetime when the most recent pending event is from :param skip_details: (list of dict) of business service details to skip :return: (list of dict) new impacted business service details ''' imp_buses, imp_dict = [], dict() for bus in bus_details: to_include, bus_ref, bus_sts = False, bus[var_names.business_service_ref_id], None if was_resolved: to_include = True bus_sts = constants.operational_state else: bus_start = bus[var_names.start_timestamp] bus_end = bus[var_names.end_timestamp] if bus_start <= self.monitor_time < bus_end: if max_pending_time is None or bus_start > max_pending_time or bus_end <= max_pending_time: to_include = True bus_sts = constants.partial_outage_state if to_include: imp_buses.append(bus_ref) imp_dict[bus_ref] = bus_sts if skip_details is not None: skip_dict = dict() for ski in skip_details: skip_dict[ski[var_names.business_service_ref_id]] = ski[var_names.status] diff_buses = set(imp_buses).symmetric_difference(set(list(skip_dict.keys()))) if len(diff_buses) == 0: return [] else: new_imp_dict = dict() for diff_ref in diff_buses: if diff_ref in imp_dict: diff_sts = skip_dict[diff_ref] if diff_ref in skip_dict else imp_dict[diff_ref] new_imp_dict[diff_ref] = diff_sts imp_dict = new_imp_dict imp_struct = [{var_names.business_service_ref_id: x, var_names.status: imp_dict[x]} for x in imp_dict] return imp_struct 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('--timestamp', type=str, default=times.get_current_timestamp()) arg_parser.add_argument('--dont_switch_to_current_time', action='store_true') args = arg_parser.parse_args() start_time = args.timestamp dont_switch_to_current_time = args.dont_switch_to_current_time # pre text of error message for internal alerting pre_error_title = 'Business Impact Monitor (' + settings.REGION + ')' 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() 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() # wait time in seconds before next run; refresh minutes for how often data should be refreshed from db wait_seconds = 45 refresh_minutes = 10 auto_update_minutes = 5 # We set the last_refresh earlier than start time so that a refresh happens immediately upon starting the monitor stop = False timestamp = start_time last_timestamp = timestamp - datetime.timedelta(minutes=15) force_load_from_db = True last_refresh = timestamp force_auto_update = True last_auto_update = timestamp # Clean the current pending updates in the cache cache_task_instances.remove_pending_instance_updates(monitor_cache, 0, -1) cache_status_pages.remove_pending_status_page_updates(monitor_cache, 0, -1) while not stop: if timestamp.date() != monitor_date: monitor_date = timestamp.date() try: current_monitor = BusinessImpactMonitor(monitor_conn, monitor_cache, monitor_time=timestamp, last_run=last_timestamp, email_creds=email_credentials, load_from_db=force_load_from_db, check_for_auto_update=force_auto_update) current_monitor.start() current_monitor.join() last_timestamp = timestamp time.sleep(wait_seconds) if dont_switch_to_current_time: timestamp = timestamp + datetime.timedelta(seconds=wait_seconds) else: timestamp = times.get_current_timestamp() if (timestamp - last_refresh).seconds / 60 > refresh_minutes: force_load_from_db = True last_refresh = timestamp else: force_load_from_db = False if (timestamp - last_auto_update).seconds / 60 > auto_update_minutes: force_auto_update = True last_auto_update = timestamp else: force_auto_update = False 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)