# Contains all of the classes that are shared across both the Canary Wrapper and the Persistent Canary Wrapper scripts # If a class can/is reused, then it should be in this file. # Needs to be installed prior to running import boto3 import psutil # Part of standard packages in Python 3.4+ import time import os import json import subprocess import zipfile import datetime # ================================================================================ # Class that holds metric data and has a few utility functions for getting that data in a format we can use for Cloudwatch class DataSnapshot_Metric(): def __init__(self, metric_name, metric_function, metric_dimensions=[], metric_unit="None", metric_alarm_threshold=None, metric_alarm_severity=6, git_hash="", git_repo_name="", reports_to_skip=0, is_percent=False): self.metric_name = metric_name self.metric_function = metric_function self.metric_dimensions = metric_dimensions self.metric_unit = metric_unit self.metric_alarm_threshold = metric_alarm_threshold self.metric_alarm_name = self.metric_name + "-" + git_repo_name + "-" + git_hash self.metric_alarm_description = 'Alarm for metric "' + self.metric_name + '" - git hash: ' + git_hash self.metric_value = None self.reports_to_skip = reports_to_skip self.metric_alarm_severity = metric_alarm_severity self.is_percent = is_percent # Gets the latest metric value from the metric_function callback def get_metric_value(self, psutil_process : psutil.Process): if not self.metric_function is None: self.metric_value = self.metric_function(psutil_process) return self.metric_value # Returns the data needed to send to Cloudwatch when posting metrics def get_metric_cloudwatch_dictionary(self): if (self.reports_to_skip > 0): self.reports_to_skip -= 1 return None # skips sending to Cloudwatch if (self.metric_value == None): return None # skips sending to Cloudwatch return { "MetricName": self.metric_name, "Dimensions": self.metric_dimensions, "Value": self.metric_value, "Unit": self.metric_unit } class DataSnapshot_Dashboard_Widget(): def __init__(self, widget_name, metric_namespace, metric_dimension, cloudwatch_region="us-east-1", widget_period=60) -> None: self.metric_list = [] self.region = cloudwatch_region self.widget_name = widget_name self.metric_namespace = metric_namespace self.metric_dimension = metric_dimension self.widget_period = widget_period def add_metric_to_widget(self, new_metric_name): try: self.metric_list.append(new_metric_name) except Exception as e: print ("[DataSnapshot_Dashboard] ERROR - could not add metric to dashboard widget due to exception!") print ("[DataSnapshot_Dashboard] Exception: " + str(e)) def remove_metric_from_widget(self, existing_metric_name): try: self.metric_list.remove(existing_metric_name) except Exception as e: print ("[DataSnapshot_Dashboard] ERROR - could not remove metric from dashboard widget due to exception!") print ("[DataSnapshot_Dashboard] Exception: " + str(e)) def get_widget_dictionary(self): metric_list_json = [] for metric_name in self.metric_list: metric_list_json.append([self.metric_namespace, metric_name, self.metric_dimension, metric_name]) return { "type":"metric", "properties" : { "metrics" : metric_list_json, "region": self.region, "title": self.widget_name, "period": self.widget_period, }, "width": 14, "height": 10 } # ================================================================================ # Class that keeps track of the metrics registered, sets up Cloudwatch and S3, and sends periodic reports # Is the backbone of the reporting operation class DataSnapshot(): def __init__(self, git_hash=None, git_repo_name=None, git_hash_as_namespace=False, git_fixed_namespace_text="mqtt5_canary", datetime_string=None, output_log_filepath=None, output_to_console=True, cloudwatch_region="us-east-1", cloudwatch_make_dashboard=False, cloudwatch_teardown_alarms_on_complete=True, cloudwatch_teardown_dashboard_on_complete=True, s3_bucket_name="canary-wrapper-bucket", s3_bucket_upload_on_complete=True, lambda_name="CanarySendEmailLambda", metric_frequency=None): # Setting initial values # ================== self.first_metric_call = True self.metrics = [] self.metrics_numbers = [] self.metric_report_number = 0 self.metric_report_non_zero_count = 4 # Needed so we can initialize Cloudwatch alarms, etc, outside of the init function # but before we start sending data. # This boolean tracks whether we have done the post-initialization prior to sending the first report. self.perform_final_initialization = True # Watched by the thread creating the snapshot. Will cause the thread(s) to abort and return an error. self.abort_due_to_internal_error = False self.abort_due_to_internal_error_reason = "" self.abort_due_to_internal_error_due_to_credentials = False self.git_hash = None self.git_repo_name = None self.git_hash_as_namespace = git_hash_as_namespace self.git_fixed_namespace_text = git_fixed_namespace_text self.git_metric_namespace = None self.cloudwatch_region = cloudwatch_region self.cloudwatch_client = None self.cloudwatch_make_dashboard = cloudwatch_make_dashboard self.cloudwatch_teardown_alarms_on_complete = cloudwatch_teardown_alarms_on_complete self.cloudwatch_teardown_dashboard_on_complete = cloudwatch_teardown_dashboard_on_complete self.cloudwatch_dashboard_name = "" self.cloudwatch_dashboard_widgets = [] self.s3_bucket_name = s3_bucket_name self.s3_client = None self.s3_bucket_upload_on_complete = s3_bucket_upload_on_complete self.output_to_file_filepath = output_log_filepath self.output_to_file = False self.output_file = None self.output_to_console = output_to_console self.lambda_client = None self.lambda_name = lambda_name self.datetime_string = datetime_string self.metric_frequency = metric_frequency # ================== # Check for valid credentials # ================== try: tmp_sts_client = boto3.client('sts') tmp_sts_client.get_caller_identity() except Exception as e: print ("[DataSnapshot] ERROR - AWS credentials are NOT valid!") self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "AWS credentials are NOT valid!" self.abort_due_to_internal_error_due_to_credentials = True return # ================== # Git related stuff # ================== if (git_hash == None or git_repo_name == None): print("[DataSnapshot] ERROR - a Git hash and repository name are REQUIRED for the canary wrapper to run!") self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "No Git hash and repository passed!" return self.git_hash = git_hash self.git_repo_name = git_repo_name if (self.git_hash_as_namespace == False): self.git_metric_namespace = self.git_fixed_namespace_text else: if (self.datetime_string == None): git_namespace_prepend_text = self.git_repo_name + "-" + self.git_hash else: git_namespace_prepend_text = self.git_repo_name + "/" + self.datetime_string + "-" + self.git_hash self.git_metric_namespace = git_namespace_prepend_text # ================== # Cloudwatch related stuff # ================== try: self.cloudwatch_client = boto3.client('cloudwatch', self.cloudwatch_region) self.cloudwatch_dashboard_name = self.git_metric_namespace except Exception as e: self.print_message("[DataSnapshot] ERROR - could not make Cloudwatch client due to exception!") self.print_message("[DataSnapshot] Exception: " + str(e)) self.cloudwatch_client = None self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "Could not make Cloudwatch client!" return # ================== # S3 related stuff # ================== try: self.s3_client = boto3.client("s3") except Exception as e: self.print_message("[DataSnapshot] ERROR - could not make S3 client due to exception!") self.print_message("[DataSnapshot] Exception: " + str(e)) self.s3_client = None self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "Could not make S3 client!" return # ================== # Lambda related stuff # ================== try: self.lambda_client = boto3.client("lambda", self.cloudwatch_region) except Exception as e: self.print_message("[DataSnapshot] ERROR - could not make Lambda client due to exception!") self.print_message("[DataSnapshot] Exception: " + str(e)) self.lambda_client = None self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "Could not make Lambda client!" return # ================== # File output (logs) related stuff # ================== if (not output_log_filepath is None): self.output_to_file = True self.output_file = open(self.output_to_file_filepath, "w") else: self.output_to_file = False self.output_file = None # ================== self.print_message("[DataSnapshot] Data snapshot created!") # Cleans the class - closing any files, removing alarms, and sending data to S3. # Should be called at the end when you are totally finished shadowing metrics def cleanup(self, error_occurred=False): if (self.s3_bucket_upload_on_complete == True): self.export_result_to_s3_bucket(copy_output_log=True, log_is_error=error_occurred) self._cleanup_cloudwatch_alarms() if (self.cloudwatch_make_dashboard == True): self._cleanup_cloudwatch_dashboard() self.print_message("[DataSnapshot] Data snapshot cleaned!") if (self.output_file is not None): self.output_file.close() self.output_file = None # Utility function for printing messages def print_message(self, message): if self.output_to_file == True: self.output_file.write(message + "\n") if self.output_to_console == True: print(message, flush=True) # Utility function - adds the metric alarms to Cloudwatch. We do run this right before the first # collection of metrics so we can register metrics before we initialize Cloudwatch def _init_cloudwatch_pre_first_run(self): for metric in self.metrics: if (not metric.metric_alarm_threshold is None): self._add_cloudwatch_metric_alarm(metric) if (self.cloudwatch_make_dashboard == True): self._init_cloudwatch_pre_first_run_dashboard() # Utility function - adds the Cloudwatch Dashboard for the currently running data snapshot def _init_cloudwatch_pre_first_run_dashboard(self): try: # Remove the old dashboard if it exists before adding a new one self._cleanup_cloudwatch_dashboard() new_dashboard_widgets_array = [] for widget in self.cloudwatch_dashboard_widgets: new_dashboard_widgets_array.append(widget.get_widget_dictionary()) new_dashboard_body = { "start": "-PT1H", "widgets": new_dashboard_widgets_array, } new_dashboard_body_json = json.dumps(new_dashboard_body) self.cloudwatch_client.put_dashboard( DashboardName=self.cloudwatch_dashboard_name, DashboardBody= new_dashboard_body_json) self.print_message("[DataSnapshot] Added Cloudwatch dashboard successfully") except Exception as e: self.print_message("[DataSnapshot] ERROR - Cloudwatch client could not make dashboard due to exception!") self.print_message("[DataSnapshot] Exception: " + str(e)) self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "Cloudwatch client could not make dashboard due to exception" return # Utility function - The function that adds each individual metric alarm. def _add_cloudwatch_metric_alarm(self, metric): if self.cloudwatch_client is None: self.print_message("[DataSnapshot] ERROR - Cloudwatch client not setup. Cannot register alarm") return try: self.cloudwatch_client.put_metric_alarm( AlarmName=metric.metric_alarm_name, AlarmDescription=metric.metric_alarm_description, MetricName=metric.metric_name, Namespace=self.git_metric_namespace, Statistic="Maximum", Dimensions=metric.metric_dimensions, Period=60, # How long (in seconds) is an evaluation period? EvaluationPeriods=120, # How many periods does it need to be invalid for? DatapointsToAlarm=1, # How many data points need to be invalid? Threshold=metric.metric_alarm_threshold, ComparisonOperator="GreaterThanOrEqualToThreshold", ) except Exception as e: self.print_message("[DataSnapshot] ERROR - could not register alarm for metric due to exception: " + metric.metric_name) self.print_message("[DataSnapshot] Exception: " + str(e)) # Utility function - removes all the Cloudwatch alarms for the metrics def _cleanup_cloudwatch_alarms(self): if (self.cloudwatch_teardown_alarms_on_complete == True): try: for metric in self.metrics: if (not metric.metric_alarm_threshold is None): self.cloudwatch_client.delete_alarms(AlarmNames=[metric.metric_alarm_name]) except Exception as e: self.print_message("[DataSnapshot] ERROR - could not delete alarms due to exception!") self.print_message("[DataSnapshot] Exception: " + str(e)) # Utility function - removes all Cloudwatch dashboards created def _cleanup_cloudwatch_dashboard(self): if (self.cloudwatch_teardown_dashboard_on_complete == True): try: self.cloudwatch_client.delete_dashboards(DashboardNames=[self.cloudwatch_dashboard_name]) self.print_message("[DataSnapshot] Cloudwatch Dashboards deleted successfully!") except Exception as e: self.print_message("[DataSnapshot] ERROR - dashboard cleaning function failed due to exception!") self.print_message("[DataSnapshot] Exception: " + str(e)) self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "Cloudwatch dashboard cleaning function failed due to exception" return # Returns the results of the metric alarms. Will return a list containing tuples with the following structure: # [Boolean (False = the alarm is in the ALARM state), String (Name of the alarm that is in the ALARM state), int (severity of alarm)] # Currently this function will only return a list of failed alarms, so if the returned list is empty, then it means all # alarms did not get to the ALARM state in Cloudwatch for the registered metrics def get_cloudwatch_alarm_results(self): return self._check_cloudwatch_alarm_states() # Utility function - collects the metric alarm results and returns them in a list. def _check_cloudwatch_alarm_states(self): return_result_list = [] tmp = None for metric in self.metrics: tmp = self._check_cloudwatch_alarm_state_metric(metric) if (tmp[1] != None): # Do not cut a ticket for the "Alive_Alarm" that we use to check if the Canary is running if ("Alive_Alarm" in tmp[1] == False): if (tmp[0] != True): return_result_list.append(tmp) return return_result_list # Utility function - checks each individual alarm and returns a tuple with the following format: # [Boolean (False if the alarm is in the ALARM state, otherwise it is true), String (name of the alarm), Int (severity of alarm)] def _check_cloudwatch_alarm_state_metric(self, metric): alarms_response = self.cloudwatch_client.describe_alarms_for_metric( MetricName=metric.metric_name, Namespace=self.git_metric_namespace, Dimensions=metric.metric_dimensions) return_result = [True, None, metric.metric_alarm_severity] for metric_alarm_dict in alarms_response["MetricAlarms"]: if metric_alarm_dict["StateValue"] == "ALARM": return_result[0] = False return_result[1] = metric_alarm_dict["AlarmName"] break return return_result # Exports a file with the same name as the commit Git hash to an S3 bucket in a folder with the Git repo name. # By default, this file will only contain the Git hash. # If copy_output_log is true, then the output log will be copied into this file, which may be useful for debugging. def export_result_to_s3_bucket(self, copy_output_log=False, log_is_error=False): if (self.s3_client is None): self.print_message("[DataSnapshot] ERROR - No S3 client initialized! Cannot send log to S3") self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "S3 client not initialized and therefore cannot send log to S3" return s3_file = open(self.git_hash + ".log", "w") s3_file.write(self.git_hash) # Might be useful for debugging? if (copy_output_log == True and self.output_to_file == True): # Are we still writing? If so, then we need to close the file first so everything is written to it is_output_file_open_previously = False if (self.output_file != None): self.output_file.close() is_output_file_open_previously = True self.output_file = open(self.output_to_file_filepath, "r") s3_file.write("\n\nOUTPUT LOG\n") s3_file.write("==========================================================================================\n") output_file_lines = self.output_file.readlines() for line in output_file_lines: s3_file.write(line) self.output_file.close() # If we were writing to the output previously, then we need to open in RW mode so we can continue to write to it if (is_output_file_open_previously == True): self.output_to_file = open(self.output_to_file_filepath, "a") s3_file.close() # Upload to S3 try: if (log_is_error == False): if (self.datetime_string == None): self.s3_client.upload_file(self.git_hash + ".log", self.s3_bucket_name, self.git_repo_name + "/" + self.git_hash + ".log") else: self.s3_client.upload_file(self.git_hash + ".log", self.s3_bucket_name, self.git_repo_name + "/" + self.datetime_string + "/" + self.git_hash + ".log") else: if (self.datetime_string == None): self.s3_client.upload_file(self.git_hash + ".log", self.s3_bucket_name, self.git_repo_name + "/Failed_Logs/" + self.git_hash + ".log") else: self.s3_client.upload_file(self.git_hash + ".log", self.s3_bucket_name, self.git_repo_name + "/Failed_Logs/" + self.datetime_string + "/" + self.git_hash + ".log") self.print_message("[DataSnapshot] Uploaded to S3!") except Exception as e: self.print_message("[DataSnapshot] ERROR - could not upload to S3 due to exception!") self.print_message("[DataSnapshot] Exception: " + str(e)) self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "S3 client had exception and therefore could not upload log!" os.remove(self.git_hash + ".log") return # Delete the file when finished os.remove(self.git_hash + ".log") # Sends an email via a special lambda. The payload has to contain a message and a subject # * (REQUIRED) message is the message you want to send in the body of the email # * (REQUIRED) subject is the subject that the email will be sent with def lambda_send_email(self, message, subject): payload = {"Message":message, "Subject":subject} payload_string = json.dumps(payload) try: self.lambda_client.invoke( FunctionName=self.lambda_name, InvocationType="Event", ClientContext="MQTT Wrapper Script", Payload=payload_string ) except Exception as e: self.print_message("[DataSnapshot] ERROR - could not send email via Lambda due to exception!") self.print_message("[DataSnapshot] Exception: " + str(e)) self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "Lambda email function had an exception!" return # Registers a metric to be polled by the Snapshot. # * (REQUIRED) new_metric_name is the name of the metric. Cloudwatch will use this name # * (REQUIRED) new_metric_function is expected to be a pointer to a Python function and will not work if you pass a value/object # * (OPTIONAL) new_metric_unit is the metric unit. There is a list of possible metric unit types on the Boto3 documentation for Cloudwatch # * (OPTIONAL) new_metric_alarm_threshold is the value that the metric has to exceed in order to be registered as an alarm # * (OPTIONAL) new_reports_to_skip is the number of reports this metric will return nothing, but will get it's value. # * Useful for CPU calculations that require deltas # * (OPTIONAL) new_metric_alarm_severity is the severity of the ticket if this alarm is triggered. A severity of 6+ means no ticket. # * (OPTIONAL) is_percent whether or not to display the metric as a percent when printing it (default=false) def register_metric(self, new_metric_name, new_metric_function, new_metric_unit="None", new_metric_alarm_threshold=None, new_metric_reports_to_skip=0, new_metric_alarm_severity=6, is_percent=False): new_metric_dimensions = [] if (self.git_hash_as_namespace == False): git_namespace_prepend_text = self.git_repo_name + "-" + self.git_hash new_metric_dimensions.append( {"Name": git_namespace_prepend_text, "Value": new_metric_name}) else: new_metric_dimensions.append( {"Name": "System_Metrics", "Value": new_metric_name}) new_metric = DataSnapshot_Metric( metric_name=new_metric_name, metric_function=new_metric_function, metric_dimensions=new_metric_dimensions, metric_unit=new_metric_unit, metric_alarm_threshold=new_metric_alarm_threshold, metric_alarm_severity=new_metric_alarm_severity, git_hash=self.git_hash, git_repo_name=self.git_repo_name, reports_to_skip=new_metric_reports_to_skip, is_percent=is_percent ) self.metrics.append(new_metric) # append an empty list so we can track it's metrics over time self.metrics_numbers.append([]) def register_dashboard_widget(self, new_widget_name, metrics_to_add=[], new_widget_period=60): # We need to know what metric dimension to get the metric(s) from metric_dimension_string = "" if (self.git_hash_as_namespace == False): metric_dimension_string = self.git_repo_name + "-" + self.git_hash else: metric_dimension_string = "System_Metrics" widget = self._find_cloudwatch_widget(name=new_widget_name) if (widget == None): widget = DataSnapshot_Dashboard_Widget( widget_name=new_widget_name, metric_namespace=self.git_metric_namespace, metric_dimension=metric_dimension_string, cloudwatch_region=self.cloudwatch_region, widget_period=new_widget_period) self.cloudwatch_dashboard_widgets.append(widget) for metric in metrics_to_add: self.register_metric_to_dashboard_widget(widget_name=new_widget_name, metric_name=metric) def register_metric_to_dashboard_widget(self, widget_name, metric_name, widget=None): if widget is None: widget = self._find_cloudwatch_widget(name=widget_name) if widget is None: print ("[DataSnapshot] ERROR - could not find widget with name: " + widget_name, flush=True) return # Adjust metric name so it has the git hash, repo, etc metric_name_formatted = metric_name widget.add_metric_to_widget(new_metric_name=metric_name_formatted) return def remove_metric_from_dashboard_widget(self, widget_name, metric_name, widget=None): if widget is None: widget = self._find_cloudwatch_widget(name=widget_name) if widget is None: print ("[DataSnapshot] ERROR - could not find widget with name: " + widget_name, flush=True) return widget.remove_metric_from_widget(existing_metric_name=metric_name) return def _find_cloudwatch_widget(self, name): result = None for widget in self.cloudwatch_dashboard_widgets: if widget.widget_name == name: return widget return result # Prints the metrics to the console def export_metrics_console(self): datetime_now = datetime.datetime.now() datetime_string = datetime_now.strftime("%d-%m-%Y/%H:%M:%S") self.print_message("\n[DataSnapshot] Metric report: " + str(self.metric_report_number) + " (" + datetime_string + ")") for metric in self.metrics: if (metric.is_percent == True): self.print_message(" " + metric.metric_name + " - value: " + str(metric.metric_value) + "%") else: self.print_message(" " + metric.metric_name + " - value: " + str(metric.metric_value)) self.print_message("") # Sends all registered metrics to Cloudwatch. # Does NOT need to called on loop. Call post_metrics on loop to send all the metrics as expected. # This is just the Cloudwatch part of that loop. def export_metrics_cloudwatch(self): if (self.cloudwatch_client == None): self.print_message("[DataSnapshot] Error - cannot export Cloudwatch metrics! Cloudwatch was not initialized.") self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "Could not export Cloudwatch metrics due to no Cloudwatch client initialized!" return self.print_message("[DataSnapshot] Preparing to send to Cloudwatch...") metrics_data = [] metric_data_tmp = None for metric in self.metrics: metric_data_tmp = metric.get_metric_cloudwatch_dictionary() if (not metric_data_tmp is None): metrics_data.append(metric_data_tmp) if (len(metrics_data) == 0): self.print_message("[DataSnapshot] INFO - no metric data to send. Skipping...") return try: self.cloudwatch_client.put_metric_data( Namespace=self.git_metric_namespace, MetricData=metrics_data) self.print_message("[DataSnapshot] Metrics sent to Cloudwatch.") except Exception as e: self.print_message("[DataSnapshot] Error - something when wrong posting cloudwatch metrics!") self.print_message("[DataSnapshot] Exception: " + str(e)) self.abort_due_to_internal_error = True self.abort_due_to_internal_error_reason = "Could not export Cloudwatch metrics due to exception in Cloudwatch client!" return # Call this at a set interval to post the metrics to Cloudwatch, etc. # This is the function you want to call repeatedly after you have everything setup. def post_metrics(self, psutil_process : psutil.Process): if (self.perform_final_initialization == True): self.perform_final_initialization = False self._init_cloudwatch_pre_first_run() # Update the metric values internally for i in range(0, len(self.metrics)): metric_value = self.metrics[i].get_metric_value(psutil_process) self.metrics_numbers[i].insert(0, metric_value) # Only keep the last metric_report_non_zero_count results if (len(self.metrics_numbers[i]) > self.metric_report_non_zero_count): amount_to_delete = len(self.metrics_numbers[i]) - self.metric_report_non_zero_count del self.metrics_numbers[i][-amount_to_delete:] # If we have metric_report_non_zero_count amount of metrics, make sure there is at least one # non-zero. If it is all zero, then print a log so we can easily find it if (len(self.metrics_numbers[i]) == self.metric_report_non_zero_count): non_zero_found = False for j in range(0, len(self.metrics_numbers[i])): if (self.metrics_numbers[i][j] != 0.0 and self.metrics_numbers[i][j] != None): non_zero_found = True break if (non_zero_found == False): self.print_message("\n[DataSnapshot] METRIC ZERO ERROR!") self.print_message(f"[DataSnapshot] Metric index {i} has been zero for last {self.metric_report_non_zero_count} reports!") self.print_message("\n") self.metric_report_number += 1 self.export_metrics_console() self.export_metrics_cloudwatch() def output_diagnosis_information(self, dependencies_list): # Print general diagnosis information self.print_message("\n========== Canary Wrapper diagnosis information ==========") self.print_message("\nRunning Canary for repository: " + self.git_repo_name) self.print_message("\t Commit hash: " + self.git_hash) if not dependencies_list == "": self.print_message("\nDependencies:") dependencies_list = dependencies_list.split(";") dependencies_list_found_hash = False for i in range(0, len(dependencies_list)): # There's probably a better way to do this... if (dependencies_list_found_hash == True): dependencies_list_found_hash = False continue self.print_message("* " + dependencies_list[i]) if (i+1 < len(dependencies_list)): self.print_message("\t Commit hash: " + dependencies_list[i+1]) dependencies_list_found_hash = True else: self.print_message("\t Commit hash: Unknown") if (self.metric_frequency != None): self.print_message("\nMetric Snapshot Frequency: " + str(self.metric_frequency) + " seconds") self.print_message("\nMetrics:") for metric in self.metrics: self.print_message("* " + metric.metric_name) if metric.metric_alarm_threshold is not None: self.print_message("\t Alarm Threshold: " + str(metric.metric_alarm_threshold)) self.print_message("\t Alarm Severity: " + str(metric.metric_alarm_severity)) else: self.print_message("\t No alarm set for metric.") self.print_message("\n") self.print_message("==========================================================") self.print_message("\n") # ================================================================================ class SnapshotMonitor(): def __init__(self, wrapper_data_snapshot, wrapper_metrics_wait_time) -> None: self.data_snapshot = wrapper_data_snapshot self.had_internal_error = False self.error_due_to_credentials = False self.internal_error_reason = "" self.error_due_to_alarm = False self.can_cut_ticket = False self.has_cut_ticket = False # A list of all the alarms triggered in the last check, cached for later # NOTE - this is only the alarm names! Not the severity. This just makes it easier to process self.cloudwatch_current_alarms_triggered = [] # Check for errors if (self.data_snapshot.abort_due_to_internal_error == True): self.had_internal_error = True self.internal_error_reason = "Could not initialize DataSnapshot. Likely credentials are not setup!" if (self.data_snapshot.abort_due_to_internal_error_due_to_credentials == True): self.error_due_to_credentials = True self.data_snapshot.cleanup() return # How long to wait before posting a metric self.metric_post_timer = 0 self.metric_post_timer_time = wrapper_metrics_wait_time def register_metric(self, new_metric_name, new_metric_function, new_metric_unit="None", new_metric_alarm_threshold=None, new_metric_reports_to_skip=0, new_metric_alarm_severity=6): try: self.data_snapshot.register_metric( new_metric_name=new_metric_name, new_metric_function=new_metric_function, new_metric_unit=new_metric_unit, new_metric_alarm_threshold=new_metric_alarm_threshold, new_metric_reports_to_skip=new_metric_reports_to_skip, new_metric_alarm_severity=new_metric_alarm_severity) except Exception as e: self.print_message("[SnaptshotMonitor] ERROR - could not register metric in data snapshot due to exception!") self.print_message("[SnaptshotMonitor] Exception: " + str(e)) self.had_internal_error = True self.internal_error_reason = "Could not register metric in data snapshot due to exception" return def register_dashboard_widget(self, new_widget_name, metrics_to_add=[], widget_period=60): self.data_snapshot.register_dashboard_widget(new_widget_name=new_widget_name, metrics_to_add=metrics_to_add, new_widget_period=widget_period) def output_diagnosis_information(self, dependencies=""): self.data_snapshot.output_diagnosis_information(dependencies_list=dependencies) def check_alarms_for_new_alarms(self, triggered_alarms): if len(triggered_alarms) > 0: self.data_snapshot.print_message( "WARNING - One or more alarms are in state of ALARM") old_alarms_still_active = [] new_alarms = [] new_alarms_highest_severity = 6 new_alarm_found = True new_alarm_ticket_description = "Canary has metrics in ALARM state!\n\nMetrics in alarm:\n" for triggered_alarm in triggered_alarms: new_alarm_found = True # Is this a new alarm? for old_alarm_name in self.cloudwatch_current_alarms_triggered: if (old_alarm_name == triggered_alarm[1]): new_alarm_found = False old_alarms_still_active.append(triggered_alarm[1]) new_alarm_ticket_description += "* (STILL IN ALARM) " + triggered_alarm[1] + "\n" new_alarm_ticket_description += "\tSeverity: " + str(triggered_alarm[2]) new_alarm_ticket_description += "\n" break # If it is a new alarm, then add it to our list so we can cut a new ticket if (new_alarm_found == True): self.data_snapshot.print_message(' (NEW) Alarm with name "' + triggered_alarm[1] + '" is in the ALARM state!') new_alarms.append(triggered_alarm[1]) if (triggered_alarm[2] < new_alarms_highest_severity): new_alarms_highest_severity = triggered_alarm[2] new_alarm_ticket_description += "* " + triggered_alarm[1] + "\n" new_alarm_ticket_description += "\tSeverity: " + str(triggered_alarm[2]) new_alarm_ticket_description += "\n" if len(new_alarms) > 0: if (self.can_cut_ticket == True): cut_ticket_using_cloudwatch( git_repo_name=self.data_snapshot.git_repo_name, git_hash=self.data_snapshot.git_hash, git_hash_as_namespace=False, git_fixed_namespace_text=self.data_snapshot.git_fixed_namespace_text, cloudwatch_region="us-east-1", ticket_description="New metric(s) went into alarm for the Canary! Metrics in alarm: " + str(new_alarms), ticket_reason="New metric(s) went into alarm", ticket_allow_duplicates=True, ticket_category="AWS", ticket_item="IoT SDK for CPP", ticket_group="AWS IoT Device SDK", ticket_type="SDKs and Tools", ticket_severity=4) self.has_cut_ticket = True # Cache the new alarms and the old alarms self.cloudwatch_current_alarms_triggered = old_alarms_still_active + new_alarms else: self.cloudwatch_current_alarms_triggered.clear() def monitor_loop_function(self, psutil_process : psutil.Process, time_passed=30): # Check for internal errors if (self.data_snapshot.abort_due_to_internal_error == True): self.had_internal_error = True self.internal_error_reason = "Data Snapshot internal error: " + self.data_snapshot.abort_due_to_internal_error_reason return try: # Poll the metric alarms if (self.had_internal_error == False): # Get a report of all the alarms that might have been set to an alarm state triggered_alarms = self.data_snapshot.get_cloudwatch_alarm_results() self.check_alarms_for_new_alarms(triggered_alarms) except Exception as e: self.print_message("[SnaptshotMonitor] ERROR - exception occurred checking metric alarms!") self.print_message("[SnaptshotMonitor] (Likely session credentials expired)") self.had_internal_error = True self.internal_error_reason = "Exception occurred checking metric alarms! Likely session credentials expired" return if (self.metric_post_timer <= 0): if (self.had_internal_error == False): try: self.data_snapshot.post_metrics(psutil_process) except Exception as e: self.print_message("[SnaptshotMonitor] ERROR - exception occurred posting metrics!") self.print_message("[SnaptshotMonitor] (Likely session credentials expired)") print (e, flush=True) self.had_internal_error = True self.internal_error_reason = "Exception occurred posting metrics! Likely session credentials expired" return # reset the timer self.metric_post_timer += self.metric_post_timer_time # Gather and post the metrics self.metric_post_timer -= time_passed def send_email(self, email_body, email_subject_text_append=None): if (email_subject_text_append != None): self.data_snapshot.lambda_send_email(email_body, "Canary: " + self.data_snapshot.git_repo_name + ":" + self.data_snapshot.git_hash + " - " + email_subject_text_append) else: self.data_snapshot.lambda_send_email(email_body, "Canary: " + self.data_snapshot.git_repo_name + ":" + self.data_snapshot.git_hash) def stop_monitoring(self): # Stub - just added for consistency pass def start_monitoring(self): # Stub - just added for consistency pass def restart_monitoring(self): # Stub - just added for consistency pass def cleanup_monitor(self, error_occurred=False): self.data_snapshot.cleanup(error_occurred=error_occurred) def print_message(self, message): if (self.data_snapshot != None): self.data_snapshot.print_message(message) else: print(message, flush=True) # ================================================================================ class ApplicationMonitor(): def __init__(self, wrapper_application_path, wrapper_application_arguments, wrapper_application_restart_on_finish=True, data_snapshot=None) -> None: self.application_process = None self.application_process_psutil = None self.error_has_occurred = False self.error_due_to_credentials = False self.error_reason = "" self.error_code = 0 self.wrapper_application_path = wrapper_application_path self.wrapper_application_arguments = wrapper_application_arguments self.wrapper_application_restart_on_finish = wrapper_application_restart_on_finish self.data_snapshot=data_snapshot self.stdout_file_path = "Canary_Stdout_File.txt" def start_monitoring(self): self.print_message("[ApplicationMonitor] Starting to monitor application...") if (self.application_process == None): try: canary_command = self.wrapper_application_path + " " + self.wrapper_application_arguments self.application_process = subprocess.Popen(canary_command + " | tee " + self.stdout_file_path, shell=True) self.application_process_psutil = psutil.Process(self.application_process.pid) self.print_message ("[ApplicationMonitor] Application started...") except Exception as e: self.print_message ("[ApplicationMonitor] ERROR - Could not launch Canary/Application due to exception!") self.print_message ("[ApplicationMonitor] Exception: " + str(e)) self.error_has_occurred = True self.error_reason = "Could not launch Canary/Application due to exception" self.error_code = 1 return else: self.print_message("[ApplicationMonitor] ERROR - Monitor already has an application process! Cannot monitor two applications with one monitor class!") def restart_monitoring(self): self.print_message ("[ApplicationMonitor] Restarting monitor application...") if (self.application_process != None): try: self.stop_monitoring() self.start_monitoring() self.print_message("\n[ApplicationMonitor] Restarted monitor application!") self.print_message("================================================================================") except Exception as e: self.print_message("[ApplicationMonitor] ERROR - Could not restart Canary/Application due to exception!") self.print_message("[ApplicationMonitor] Exception: " + str(e)) self.error_has_occurred = True self.error_reason = "Could not restart Canary/Application due to exception" self.error_code = 1 return else: self.print_message("[ApplicationMonitor] ERROR - Application process restart called but process is/was not running!") self.error_has_occurred = True self.error_reason = "Could not restart Canary/Application due to application process not being started initially" self.error_code = 1 return def stop_monitoring(self): self.print_message ("[ApplicationMonitor] Stopping monitor application...") if (not self.application_process == None): self.application_process.terminate() self.application_process.wait() self.print_message ("[ApplicationMonitor] Stopped monitor application!") self.application_process = None self.print_stdout() else: self.print_message ("[ApplicationMonitor] ERROR - cannot stop monitor application because no process is found!") def print_stdout(self): # Print the STDOUT file if (os.path.isfile(self.stdout_file_path)): self.print_message("Just finished Application STDOUT: ") with open(self.stdout_file_path, "r") as stdout_file: self.print_message(stdout_file.read()) os.remove(self.stdout_file_path) def monitor_loop_function(self, time_passed=30): if (self.application_process != None): application_process_return_code = None try: application_process_return_code = self.application_process.poll() except Exception as e: self.print_message("[ApplicationMonitor] ERROR - exception occurred while trying to poll application status!") self.print_message("[ApplicationMonitor] Exception: " + str(e)) self.error_has_occurred = True self.error_reason = "Exception when polling application status" self.error_code = 1 return # If it is not none, then the application finished if (application_process_return_code != None): self.print_message("[ApplicationMonitor] Monitor application has stopped! Processing result...") if (application_process_return_code != 0): self.print_message("[ApplicationMonitor] ERROR - Something Crashed in Canary/Application!") self.print_message("[ApplicationMonitor] Error code: " + str(application_process_return_code)) self.error_has_occurred = True self.error_reason = "Canary application crashed!" self.error_code = application_process_return_code else: # Should we restart? if (self.wrapper_application_restart_on_finish == True): self.print_message("[ApplicationMonitor] NOTE - Canary finished running and is restarting...") self.restart_monitoring() else: self.print_message("[ApplicationMonitor] Monitor application has stopped and monitor is not supposed to restart... Finishing...") self.error_has_occurred = True self.error_reason = "Canary Application Finished" self.error_code = 0 else: self.print_message("[ApplicationMonitor] Monitor application is still running...") def cleanup_monitor(self, error_occurred=False): pass def print_message(self, message): if (self.data_snapshot != None): self.data_snapshot.print_message(message) else: print(message, flush=True) # ================================================================================ class S3Monitor(): def __init__(self, s3_bucket_name, s3_file_name, s3_file_name_in_zip, canary_local_application_path, data_snapshot) -> None: self.s3_client = None self.s3_current_object_version_id = None self.s3_current_object_last_modified = None self.s3_bucket_name = s3_bucket_name self.s3_file_name = s3_file_name self.s3_file_name_only_path, self.s3_file_name_only_extension = os.path.splitext(s3_file_name) self.data_snapshot = data_snapshot self.canary_local_application_path = canary_local_application_path self.s3_file_name_in_zip = s3_file_name_in_zip self.s3_file_name_in_zip_only_path = None self.s3_file_name_in_zip_only_extension = None if (self.s3_file_name_in_zip != None): self.s3_file_name_in_zip_only_path, self.s3_file_name_in_zip_only_extension = os.path.splitext(s3_file_name_in_zip) self.s3_file_needs_replacing = False self.had_internal_error = False self.error_due_to_credentials = False self.internal_error_reason = "" # Check for valid credentials # ================== try: tmp_sts_client = boto3.client('sts') tmp_sts_client.get_caller_identity() except Exception as e: self.print_message("[S3Monitor] ERROR - (S3 Check) AWS credentials are NOT valid!") self.had_internal_error = True self.error_due_to_credentials = True self.internal_error_reason = "AWS credentials are NOT valid!" return # ================== try: self.s3_client = boto3.client("s3") except Exception as e: self.print_message("[S3Monitor] ERROR - (S3 Check) Could not make S3 client") self.had_internal_error = True self.internal_error_reason = "Could not make S3 client for S3 Monitor" return def check_for_file_change(self): try: version_check_response = self.s3_client.list_object_versions( Bucket=self.s3_bucket_name, Prefix=self.s3_file_name_only_path) if "Versions" in version_check_response: for version in version_check_response["Versions"]: if (version["IsLatest"] == True): if (version["VersionId"] != self.s3_current_object_version_id or version["LastModified"] != self.s3_current_object_last_modified): self.print_message("[S3Monitor] Found new version of Canary/Application in S3!") self.print_message("[S3Monitor] Changing running Canary/Application to new one...") # Will be checked by thread to trigger replacing the file self.s3_file_needs_replacing = True self.s3_current_object_version_id = version["VersionId"] self.s3_current_object_last_modified = version["LastModified"] return except Exception as e: self.print_message("[S3Monitor] ERROR - Could not check for new version of file in S3 due to exception!") self.print_message("[S3Monitor] Exception: " + str(e)) self.had_internal_error = True self.internal_error_reason = "Could not check for S3 file due to exception in S3 client" def replace_current_file_for_new_file(self): try: self.print_message("[S3Monitor] Making directory...") if not os.path.exists("tmp"): os.makedirs("tmp") except Exception as e: self.print_message ("[S3Monitor] ERROR - could not make tmp directory to place S3 file into!") self.had_internal_error = True self.internal_error_reason = "Could not make TMP folder for S3 file download" return # Download the file new_file_path = "tmp/new_file" + self.s3_file_name_only_extension try: self.print_message("[S3Monitor] Downloading file...") s3_resource = boto3.resource("s3") s3_resource.meta.client.download_file(self.s3_bucket_name, self.s3_file_name, new_file_path) except Exception as e: self.print_message("[S3Monitor] ERROR - could not download latest S3 file into TMP folder!") self.had_internal_error = True self.internal_error_reason = "Could not download latest S3 file into TMP folder" return # Is it a zip file? if (self.s3_file_name_in_zip != None): self.print_message("[S3Monitor] New file is zip file. Unzipping...") # Unzip it! with zipfile.ZipFile(new_file_path, 'r') as zip_file: zip_file.extractall("tmp/new_file_zip") new_file_path = "tmp/new_file_zip/" + self.s3_file_name_in_zip_only_path + self.s3_file_name_in_zip_only_extension try: # is there a file already present there? if os.path.exists(self.canary_local_application_path) == True: os.remove(self.canary_local_application_path) self.print_message("[S3Monitor] Moving file...") os.replace(new_file_path, self.canary_local_application_path) self.print_message("[S3Monitor] Getting execution rights...") os.system("chmod u+x " + self.canary_local_application_path) except Exception as e: self.print_message("[S3Monitor] ERROR - could not move file into local application path due to exception!") self.print_message("[S3Monitor] Exception: " + str(e)) self.had_internal_error = True self.internal_error_reason = "Could not move file into local application path" return self.print_message("[S3Monitor] New file downloaded and moved into correct location!") self.s3_file_needs_replacing = False def stop_monitoring(self): # Stub - just added for consistency pass def start_monitoring(self): # Stub - just added for consistency pass def restart_monitoring(self): # Stub - just added for consistency pass def cleanup_monitor(self): # Stub - just added for consistency pass def monitor_loop_function(self, time_passed=30): self.check_for_file_change() def print_message(self, message): if (self.data_snapshot != None): self.data_snapshot.print_message(message) else: print(message, flush=True) # ================================================================================ # Cuts a ticket to SIM using a temporary Cloudwatch metric that is quickly created, triggered, and destroyed. # Can be called in any thread - creates its own Cloudwatch client and any data it needs is passed in. # # See (https://w.amazon.com/bin/view/CloudWatchAlarms/Internal/CloudWatchAlarmsSIMTicketing) for more details # on how the alarm is sent using Cloudwatch. def cut_ticket_using_cloudwatch( ticket_description="Description here!", ticket_reason="Reason here!", ticket_severity=5, ticket_category="AWS", ticket_type="SDKs and Tools", ticket_item="IoT SDK for CPP", ticket_group="AWS IoT Device SDK", ticket_allow_duplicates=False, git_repo_name="REPO NAME", git_hash="HASH", git_hash_as_namespace=False, git_fixed_namespace_text="mqtt5_canary", cloudwatch_region="us-east-1"): git_metric_namespace = "" if (git_hash_as_namespace == False): git_metric_namespace = git_fixed_namespace_text else: git_namespace_prepend_text = git_repo_name + "-" + git_hash git_metric_namespace = git_namespace_prepend_text try: cloudwatch_client = boto3.client('cloudwatch', cloudwatch_region) ticket_alarm_name = git_repo_name + "-" + git_hash + "-AUTO-TICKET" except Exception as e: print ("ERROR - could not create Cloudwatch client to make ticket metric alarm due to exception!") print ("Exception: " + str(e), flush=True) return new_metric_dimensions = [] if (git_hash_as_namespace == False): git_namespace_prepend_text = git_repo_name + "-" + git_hash new_metric_dimensions.append( {"Name": git_namespace_prepend_text, "Value": ticket_alarm_name}) else: new_metric_dimensions.append( {"Name": "System_Metrics", "Value": ticket_alarm_name}) ticket_arn = f"arn:aws:cloudwatch::cwa-internal:ticket:{ticket_severity}:{ticket_category}:{ticket_type}:{ticket_item}:{ticket_group}:" if (ticket_allow_duplicates == True): # use "DO-NOT-DEDUPE" so we can run the same commit again and it will cut another ticket. ticket_arn += "DO-NOT-DEDUPE" # In the ticket ARN, all spaces need to be replaced with + ticket_arn = ticket_arn.replace(" ", "+") ticket_alarm_description = f"AUTO CUT CANARY WRAPPER TICKET\n\nREASON: {ticket_reason}\n\nDESCRIPTION: {ticket_description}\n\n" # Register a metric alarm so it can auto-cut a ticket for us try: cloudwatch_client.put_metric_alarm( AlarmName=ticket_alarm_name, AlarmDescription=ticket_alarm_description, MetricName=ticket_alarm_name, Namespace=git_metric_namespace, Statistic="Maximum", Dimensions=new_metric_dimensions, Period=60, # How long (in seconds) is an evaluation period? EvaluationPeriods=1, # How many periods does it need to be invalid for? DatapointsToAlarm=1, # How many data points need to be invalid? Threshold=1, ComparisonOperator="GreaterThanOrEqualToThreshold", # The data above does not really matter - it just needs to be valid input data. # This is the part that tells Cloudwatch to cut the ticket AlarmActions=[ticket_arn] ) except Exception as e: print ("ERROR - could not create ticket metric alarm due to exception!") print ("Exception: " + str(e), flush=True) return # Trigger the alarm so it cuts the ticket try: cloudwatch_client.set_alarm_state( AlarmName=ticket_alarm_name, StateValue="ALARM", StateReason="AUTO TICKET CUT") except Exception as e: print ("ERROR - could not cut ticket due to exception!") print ("Exception: " + str(e), flush=True) return print("Waiting for ticket metric to trigger...", flush=True) # Wait a little bit (2 seconds)... time.sleep(2) # Remove the metric print("Removing ticket metric...", flush=True) cloudwatch_client.delete_alarms(AlarmNames=[ticket_alarm_name]) print ("Finished cutting ticket via Cloudwatch!", flush=True) return # A helper function that gets the majority of the ticket information from the arguments result from argparser. def cut_ticket_using_cloudwatch_from_args( ticket_description="", ticket_reason="", ticket_severity=6, arguments=None): # Do not cut a ticket for a severity of 6+ if (ticket_severity >= 6): return cut_ticket_using_cloudwatch( ticket_description=ticket_description, ticket_reason=ticket_reason, ticket_severity=ticket_severity, ticket_category=arguments.ticket_category, ticket_type=arguments.ticket_type, ticket_item=arguments.ticket_item, ticket_group=arguments.ticket_group, ticket_allow_duplicates=False, git_repo_name=arguments.git_repo_name, git_hash=arguments.git_hash, git_hash_as_namespace=arguments.git_hash_as_namespace)