xref: /aosp_15_r20/external/aws-crt-java/codebuild/CanaryWrapper_Classes.py (revision 3c7ae9de214676c52d19f01067dc1a404272dc11)

# 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)