How to use is_triggering_premature_alarm method in localstack

Best Python code snippet using localstack_python

alarm_scheduler.py

Source:alarm_scheduler.py

...147    sum_breaching = evaluated_datapoints.count(True)148    if sum_breaching >= datapoints_to_alarm:149        return True150    return False151def is_triggering_premature_alarm(metric_values: List[float], alarm_details: MetricAlarm) -> bool:152    """153    Checks if a premature alarm should be triggered.154    https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/AlarmThatSendsEmail.html#CloudWatch-alarms-avoiding-premature-transition:155    [...] alarms are designed to always go into ALARM state when the oldest available breaching datapoint during the Evaluation156    Periods number of data points is at least as old as the value of Datapoints to Alarm, and all other more recent data157    points are breaching or missing. In this case, the alarm goes into ALARM state even if the total number of datapoints158    available is lower than M (Datapoints to Alarm).159    This alarm logic applies to M out of N alarms as well.160    """161    treat_missing_data = alarm_details.get("TreatMissingData", "missing")162    if treat_missing_data not in ("missing", "ignore"):163        return False164    datapoints_to_alarm = alarm_details.get("DatapointsToAlarm", 1)165    if datapoints_to_alarm > 1:166        comparison_operator = alarm_details["ComparisonOperator"]167        threshold = alarm_details["Threshold"]168        oldest_datapoints = metric_values[:-datapoints_to_alarm]169        if oldest_datapoints.count(None) == len(oldest_datapoints):170            if metric_values[-datapoints_to_alarm] and COMPARISON_OPS.get(comparison_operator)(171                metric_values[-datapoints_to_alarm], threshold172            ):173                values = list(filter(None, metric_values[len(oldest_datapoints) :]))174                if all(175                    COMPARISON_OPS.get(comparison_operator)(value, threshold) for value in values176                ):177                    return True178    return False179def collect_metric_data(alarm_details: MetricAlarm, client: "CloudWatchClient") -> List[float]:180    """181    Collects the metric data for the evaluation interval.182    :param alarm_details: the alarm details as returned by describe_alarms183    :param client: the cloudwatch client184    :return: list with data points185    """186    metric_values = []187    evaluation_periods = alarm_details["EvaluationPeriods"]188    period = alarm_details["Period"]189    # From the docs: "Whenever an alarm evaluates whether to change state, CloudWatch attempts to retrieve a higher number of data190    # points than the number specified as Evaluation Periods."191    # No other indication, try to calculate a reasonable value:192    magic_number = max(math.floor(evaluation_periods / 3), 2)193    collected_periods = evaluation_periods + magic_number194    now = datetime.utcnow().replace(tzinfo=timezone.utc)195    metric_query = generate_metric_query(alarm_details)196    # get_metric_data needs to be run in a loop, so we also collect empty data points on the right position197    for i in range(0, collected_periods):198        start_time = now - timedelta(seconds=period)199        end_time = now200        metric_data = client.get_metric_data(201            MetricDataQueries=[metric_query], StartTime=start_time, EndTime=end_time202        )["MetricDataResults"][0]203        val = metric_data["Values"]204        # oldest datapoint should be at the beginning of the list205        metric_values.insert(0, val[0] if val else None)206        now = start_time207    return metric_values208def update_alarm_state(209    client: "CloudWatchClient",210    alarm_name: str,211    current_state: str,212    desired_state: str,213    reason: str = DEFAULT_REASON,214) -> None:215    """Updates the alarm state, if the current_state is different than the desired_state216    :param client: the cloudwatch client217    :param alarm_name: the name of the alarm218    :param current_state: the state the alarm is currently in219    :param desired_state: the state the alarm should have after updating220    :param reason: reason why the state is set, will be used to for set_alarm_state221    """222    if current_state == desired_state:223        return224    client.set_alarm_state(AlarmName=alarm_name, StateValue=desired_state, StateReason=reason)225def calculate_alarm_state(alarm_arn: str) -> None:226    """227    Calculates and updates the state of the alarm228    :param alarm_arn: the arn of the alarm to be evaluated229    """230    alarm_details = get_metric_alarm_details_for_alarm_arn(alarm_arn)231    client = get_cloudwatch_client_for_region_of_alarm(alarm_arn)232    metric_values = collect_metric_data(alarm_details, client)233    alarm_name = alarm_details["AlarmName"]234    alarm_state = alarm_details["StateValue"]235    treat_missing_data = alarm_details.get("TreatMissingData", "missing")236    empty_datapoints = metric_values.count(None)237    if empty_datapoints == len(metric_values):238        if treat_missing_data == "missing":239            update_alarm_state(240                client,241                alarm_name,242                alarm_state,243                STATE_INSUFFICIENT_DATA,244                f"{INSUFFICIENT_DATA}: empty datapoints",245            )246        elif treat_missing_data == "breaching":247            update_alarm_state(248                client,249                alarm_name,250                alarm_state,251                STATE_ALARM,252                f"{THRESHOLD_CROSSED}: empty datapoints - treated as breaching",253            )254        elif treat_missing_data == "notBreaching":255            update_alarm_state(256                client,257                alarm_name,258                alarm_state,259                STATE_OK,260                f"{THRESHOLD_OK}: empty datapoints - treated as notBreaching",261            )262        # 'ignore': keep the same state263        return264    if is_triggering_premature_alarm(metric_values, alarm_details):265        if treat_missing_data == "missing":266            update_alarm_state(267                client,268                alarm_name,269                alarm_state,270                STATE_ALARM,271                f"{THRESHOLD_CROSSED}: premature alarm for missing datapoints",272            )273        # for 'ignore' the state should be retained274        return275    # collect all non-empty datapoints from the evaluation interval276    collected_datapoints = [val for val in reversed(metric_values) if val is not None]277    # adding empty data points until amount of data points == "evaluation periods"278    evaluation_periods = alarm_details["EvaluationPeriods"]...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.