How to use update_alarm_state method in localstack

Best Python code snippet using localstack_python

alarm_scheduler.py

Source:alarm_scheduler.py

...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"]279    while len(collected_datapoints) < evaluation_periods and treat_missing_data in (280        "breaching",281        "notBreaching",282    ):283        # breaching/non-breaching datapoints will be evaluated284        # ignore/missing are not relevant285        collected_datapoints.append(None)286    if is_threshold_exceeded(collected_datapoints, alarm_details):287        update_alarm_state(client, alarm_name, alarm_state, STATE_ALARM, THRESHOLD_CROSSED)288    else:...

comm_monitor.py

Source:comm_monitor.py

...31        self.alarmed = False32        _log.warning("Clearing alarm %s.", self._alarm_handle)33        self._clear_pd.log()34        self._alarm.clear()35    def update_alarm_state(self):36        now = monotonic()37        with self.mutex:38            _log.debug("Deciding whether to change alarm state - alarmed is {}, now is {}, next check time is {}, succeeded count is {}, failed count is {}"39                       .format(self.alarmed, now, self._next_check, self.succeeded, self.failed))40            if (now > self._next_check):41                _log.debug("Checking alarm state")42                if not self.alarmed:43                    if self.succeeded == 0 and self.failed > 0:44                        self.set_alarm()45                    self._next_check = now + 3046                else:47                    if self.succeeded > 0:48                        self.clear_alarm()49                    self._next_check = now + 1550                self.succeeded = self.failed = 051    def inform_success(self):52        self.succeeded += 153        self.update_alarm_state()54    def inform_failure(self):55        self.failed += 1...

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.