There are more AWS SDK examples available in the [AWS Doc SDK Examples](https://github.com/awsdocs/aws-doc-sdk-examples) GitHub repo. # CloudWatch examples using SDK for Python (Boto3) The following code examples show you how to perform actions and implement common scenarios by using the AWS SDK for Python (Boto3) with CloudWatch. *Actions* are code excerpts from larger programs and must be run in context. While actions show you how to call individual service functions, you can see actions in context in their related scenarios. *Scenarios* are code examples that show you how to accomplish specific tasks by calling multiple functions within a service or combined with other AWS services. Each example includes a link to the complete source code, where you can find instructions on how to set up and run the code in context. **Topics** + [Actions](#actions) + [Scenarios](#scenarios) ## Actions ### `DeleteAlarmMuteRule` The following code example shows how to use `DeleteAlarmMuteRule`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def delete_alarm_mute_rule(self, name): """ Deletes an alarm mute rule. :param name: The name of the mute rule. """ try: self.cloudwatch_client.delete_alarm_mute_rule(AlarmMuteRuleName=name) logger.info("Deleted alarm mute rule %s.", name) except ClientError: logger.exception("Couldn't delete alarm mute rule %s.", name) raise ``` + For API details, see [DeleteAlarmMuteRule](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DeleteAlarmMuteRule) in *AWS SDK for Python (Boto3) API Reference*. ### `DeleteAlarms` The following code example shows how to use `DeleteAlarms`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def delete_metric_alarms(self, metric_namespace, metric_name): """ Deletes all of the alarms that are currently watching the specified metric. :param metric_namespace: The namespace of the metric. :param metric_name: The name of the metric. """ try: metric = self.cloudwatch_resource.Metric(metric_namespace, metric_name) metric.alarms.delete() logger.info( "Deleted alarms for metric %s.%s.", metric_namespace, metric_name ) except ClientError: logger.exception( "Couldn't delete alarms for metric %s.%s.", metric_namespace, metric_name, ) raise ``` + For API details, see [DeleteAlarms](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DeleteAlarms) in *AWS SDK for Python (Boto3) API Reference*. ### `DescribeAlarmContributors` The following code example shows how to use `DescribeAlarmContributors`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def describe_alarm_contributors(self, alarm_name): """ Gets the contributors for a PromQL alarm. Each contributor is one series that the alarm's query matched, identified by its label set. This is how you find out *which* hosts, services, or pods are breaching, rather than only that something is. :param alarm_name: The name of the PromQL alarm. :return: The list of contributors. Each contributor has a ContributorId, a ContributorAttributes map of the labels that identify the series, a StateReason, and the time it last changed state. """ contributors = [] try: next_token = None while True: kwargs = {"AlarmName": alarm_name} if next_token is not None: kwargs["NextToken"] = next_token response = self.cloudwatch_client.describe_alarm_contributors(**kwargs) contributors.extend(response["AlarmContributors"]) next_token = response.get("NextToken") if not next_token: break except ClientError: logger.exception("Couldn't get contributors for alarm %s.", alarm_name) raise else: logger.info( "Got %s contributors for alarm %s.", len(contributors), alarm_name ) return contributors ``` + For API details, see [DescribeAlarmContributors](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DescribeAlarmContributors) in *AWS SDK for Python (Boto3) API Reference*. ### `DescribeAlarmsForMetric` The following code example shows how to use `DescribeAlarmsForMetric`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def get_metric_alarms(self, metric_namespace, metric_name): """ Gets the alarms that are currently watching the specified metric. :param metric_namespace: The namespace of the metric. :param metric_name: The name of the metric. :returns: An iterator that yields the alarms. """ metric = self.cloudwatch_resource.Metric(metric_namespace, metric_name) alarm_iter = metric.alarms.all() logger.info("Got alarms for metric %s.%s.", metric_namespace, metric_name) return alarm_iter ``` + For API details, see [DescribeAlarmsForMetric](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DescribeAlarmsForMetric) in *AWS SDK for Python (Boto3) API Reference*. ### `DisableAlarmActions` The following code example shows how to use `DisableAlarmActions`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def enable_alarm_actions(self, alarm_name, enable): """ Enables or disables actions on the specified alarm. Alarm actions can be used to send notifications or automate responses when an alarm enters a particular state. :param alarm_name: The name of the alarm. :param enable: When True, actions are enabled for the alarm. Otherwise, they disabled. """ try: alarm = self.cloudwatch_resource.Alarm(alarm_name) if enable: alarm.enable_actions() else: alarm.disable_actions() logger.info( "%s actions for alarm %s.", "Enabled" if enable else "Disabled", alarm_name, ) except ClientError: logger.exception( "Couldn't %s actions alarm %s.", "enable" if enable else "disable", alarm_name, ) raise ``` + For API details, see [DisableAlarmActions](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DisableAlarmActions) in *AWS SDK for Python (Boto3) API Reference*. ### `EnableAlarmActions` The following code example shows how to use `EnableAlarmActions`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def enable_alarm_actions(self, alarm_name, enable): """ Enables or disables actions on the specified alarm. Alarm actions can be used to send notifications or automate responses when an alarm enters a particular state. :param alarm_name: The name of the alarm. :param enable: When True, actions are enabled for the alarm. Otherwise, they disabled. """ try: alarm = self.cloudwatch_resource.Alarm(alarm_name) if enable: alarm.enable_actions() else: alarm.disable_actions() logger.info( "%s actions for alarm %s.", "Enabled" if enable else "Disabled", alarm_name, ) except ClientError: logger.exception( "Couldn't %s actions alarm %s.", "enable" if enable else "disable", alarm_name, ) raise ``` + For API details, see [EnableAlarmActions](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/EnableAlarmActions) in *AWS SDK for Python (Boto3) API Reference*. ### `GetAlarmMuteRule` The following code example shows how to use `GetAlarmMuteRule`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def get_alarm_mute_rule(self, name): """ Gets the full configuration of an alarm mute rule, including its schedule, the alarms it targets, and whether it is currently SCHEDULED, ACTIVE, or EXPIRED. :param name: The name of the mute rule. :return: The mute rule. """ try: response = self.cloudwatch_client.get_alarm_mute_rule( AlarmMuteRuleName=name ) except ClientError: logger.exception("Couldn't get alarm mute rule %s.", name) raise else: logger.info("Got alarm mute rule %s.", name) return response ``` + For API details, see [GetAlarmMuteRule](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/GetAlarmMuteRule) in *AWS SDK for Python (Boto3) API Reference*. ### `GetMetricStatistics` The following code example shows how to use `GetMetricStatistics`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def get_metric_statistics(self, namespace, name, start, end, period, stat_types): """ Gets statistics for a metric within a specified time span. Metrics are grouped into the specified period. :param namespace: The namespace of the metric. :param name: The name of the metric. :param start: The UTC start time of the time span to retrieve. :param end: The UTC end time of the time span to retrieve. :param period: The period, in seconds, in which to group metrics. The period must match the granularity of the metric, which depends on the metric's age. For example, metrics that are older than three hours have a one-minute granularity, so the period must be at least 60 and must be a multiple of 60. :param stat_types: The type of statistics to retrieve, such as average value or maximum value. :return: The retrieved statistics for the metric. """ try: metric = self.cloudwatch_resource.Metric(namespace, name) stats = metric.get_statistics( StartTime=start, EndTime=end, Period=period, Statistics=stat_types ) logger.info( "Got %s statistics for %s.", len(stats["Datapoints"]), stats["Label"] ) except ClientError: logger.exception("Couldn't get statistics for %s.%s.", namespace, name) raise else: return stats ``` + For API details, see [GetMetricStatistics](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/GetMetricStatistics) in *AWS SDK for Python (Boto3) API Reference*. ### `GetOTelEnrichment` The following code example shows how to use `GetOTelEnrichment`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def get_otel_enrichment_status(self): """ Gets the current OTel enrichment status for the account. :return: The status, either 'Running' or 'Stopped'. """ try: response = self.cloudwatch_client.get_o_tel_enrichment() except ClientError: logger.exception("Couldn't get the OTel enrichment status.") raise else: status = response["Status"] logger.info("OTel enrichment status is %s.", status) return status ``` + For API details, see [GetOTelEnrichment](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/GetOTelEnrichment) in *AWS SDK for Python (Boto3) API Reference*. ### `ListAlarmMuteRules` The following code example shows how to use `ListAlarmMuteRules`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def list_alarm_mute_rules(self, alarm_name=None, statuses=None): """ Lists alarm mute rules in the account. :param alarm_name: When specified, only rules that target this alarm are returned. :param statuses: When specified, only rules in these statuses are returned. Valid values are 'SCHEDULED', 'ACTIVE', and 'EXPIRED'. :return: The list of mute rule summaries. """ summaries = [] try: next_token = None while True: kwargs = {} if alarm_name is not None: kwargs["AlarmName"] = alarm_name if statuses is not None: kwargs["Statuses"] = statuses if next_token is not None: kwargs["NextToken"] = next_token response = self.cloudwatch_client.list_alarm_mute_rules(**kwargs) summaries.extend(response.get("AlarmMuteRuleSummaries", [])) next_token = response.get("NextToken") if not next_token: break except ClientError: logger.exception("Couldn't list alarm mute rules.") raise else: logger.info("Got %s alarm mute rules.", len(summaries)) return summaries ``` + For API details, see [ListAlarmMuteRules](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/ListAlarmMuteRules) in *AWS SDK for Python (Boto3) API Reference*. ### `ListMetrics` The following code example shows how to use `ListMetrics`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def list_metrics(self, namespace, name, recent=False): """ Gets the metrics within a namespace that have the specified name. If the metric has no dimensions, a single metric is returned. Otherwise, metrics for all dimensions are returned. :param namespace: The namespace of the metric. :param name: The name of the metric. :param recent: When True, only metrics that have been active in the last three hours are returned. :return: An iterator that yields the retrieved metrics. """ try: kwargs = {"Namespace": namespace, "MetricName": name} if recent: kwargs["RecentlyActive"] = "PT3H" # List past 3 hours only metric_iter = self.cloudwatch_resource.metrics.filter(**kwargs) logger.info("Got metrics for %s.%s.", namespace, name) except ClientError: logger.exception("Couldn't get metrics for %s.%s.", namespace, name) raise else: return metric_iter ``` + For API details, see [ListMetrics](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/ListMetrics) in *AWS SDK for Python (Boto3) API Reference*. ### `PutAlarmMuteRule` The following code example shows how to use `PutAlarmMuteRule`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def put_alarm_mute_rule( self, name, expression, duration, alarm_names=None, timezone=None, description=None, ): """ Creates or updates an alarm mute rule. While a mute rule is active the targeted alarms keep evaluating and keep transitioning between states, but their configured actions do not fire. This is the supported way to suppress notifications during a known maintenance window instead of disabling alarm actions and hoping someone remembers to turn them back on. :param name: The name of the mute rule. :param expression: When the rule activates. For a recurring window, use a five-field cron expression, 'cron(Minutes Hours Day-of-month Month Day-of-week)', such as 'cron(0 2 * * SUN)' for every Sunday at 2:00 AM. Note that this is five fields, not the six that Amazon EventBridge uses. For a one-time window, use 'at(yyyy-MM-ddThh:mm)', such as 'at(2026-09-05T02:00)'. :param duration: How long the mute window lasts once it activates, in ISO 8601 duration format, from 'PT1M' (one minute) to 'P15D' (15 days). For example, 'PT2H' is two hours and 'P2DT12H' is two days and 12 hours. :param alarm_names: The names of up to 100 alarms to mute. If omitted, the rule applies to all alarms in the account. :param timezone: The time zone the expression is evaluated in, such as 'America/Los_Angeles'. :param description: The description of the mute rule. """ schedule = {"Expression": expression, "Duration": duration} if timezone is not None: schedule["Timezone"] = timezone kwargs = {"Name": name, "Rule": {"Schedule": schedule}} if alarm_names is not None: kwargs["MuteTargets"] = {"AlarmNames": alarm_names} if description is not None: kwargs["Description"] = description try: self.cloudwatch_client.put_alarm_mute_rule(**kwargs) logger.info("Put alarm mute rule %s.", name) except ClientError: logger.exception("Couldn't put alarm mute rule %s.", name) raise ``` + For API details, see [PutAlarmMuteRule](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/PutAlarmMuteRule) in *AWS SDK for Python (Boto3) API Reference*. ### `PutMetricAlarm` The following code example shows how to use `PutMetricAlarm`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). Create an alarm that evaluates a PromQL query against OpenTelemetry metrics. ``` class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def create_promql_alarm( self, alarm_name, query, evaluation_interval, pending_period=300, recovery_period=120, description=None, alarm_actions=None, ): """ Creates an alarm that evaluates a PromQL query. A PromQL alarm differs from a classic metric alarm in a few ways. The query can match many series at once, and each matching series is tracked separately as a *contributor*. Instead of counting breaching periods, you specify durations: a contributor moves to ALARM after it breaches continuously for the pending period, and back to OK after it stops breaching for the recovery period. A PromQL alarm starts in the OK state rather than INSUFFICIENT_DATA. The PromQL evaluation parameters live in the EvaluationCriteria union, which is mutually exclusive with the classic MetricName and Metrics parameters. When you use EvaluationCriteria you must also set EvaluationInterval, and you must not set Period, Statistic, Threshold, ComparisonOperator, EvaluationPeriods, DatapointsToAlarm, or TreatMissingData. :param alarm_name: The name of the alarm. Must be unique within the Region. :param query: The PromQL query to evaluate, such as 'avg(cpu_utilization_percent) > 80'. The comparison belongs in the query itself; there is no separate threshold parameter. :param evaluation_interval: How often, in seconds, to run the query. Valid values are 10, 20, 30, and any multiple of 60, up to 3600. :param pending_period: How long, in seconds, a contributor must breach continuously before it moves to ALARM. :param recovery_period: How long, in seconds, a contributor must stop breaching before it moves back to OK. :param description: The description of the alarm. :param alarm_actions: A list of ARNs to notify when the alarm fires, such as an Amazon SNS topic. """ promql_criteria = { "Query": query, "PendingPeriod": pending_period, "RecoveryPeriod": recovery_period, } kwargs = { "AlarmName": alarm_name, "EvaluationCriteria": {"PromQLCriteria": promql_criteria}, "EvaluationInterval": evaluation_interval, } if description is not None: kwargs["AlarmDescription"] = description if alarm_actions is not None: kwargs["AlarmActions"] = alarm_actions try: self.cloudwatch_client.put_metric_alarm(**kwargs) logger.info("Created PromQL alarm %s for query %s.", alarm_name, query) except ClientError: logger.exception("Couldn't create PromQL alarm %s.", alarm_name) raise ``` Create an alarm that evaluates a single CloudWatch metric. ``` class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def create_metric_alarm( self, metric_namespace, metric_name, alarm_name, stat_type, period, eval_periods, threshold, comparison_op, ): """ Creates an alarm that watches a metric. :param metric_namespace: The namespace of the metric. :param metric_name: The name of the metric. :param alarm_name: The name of the alarm. :param stat_type: The type of statistic the alarm watches. :param period: The period in which metric data are grouped to calculate statistics. :param eval_periods: The number of periods that the metric must be over the alarm threshold before the alarm is set into an alarmed state. :param threshold: The threshold value to compare against the metric statistic. :param comparison_op: The comparison operation used to compare the threshold against the metric. :return: The newly created alarm. """ try: metric = self.cloudwatch_resource.Metric(metric_namespace, metric_name) alarm = metric.put_alarm( AlarmName=alarm_name, Statistic=stat_type, Period=period, EvaluationPeriods=eval_periods, Threshold=threshold, ComparisonOperator=comparison_op, ) logger.info( "Added alarm %s to track metric %s.%s.", alarm_name, metric_namespace, metric_name, ) except ClientError: logger.exception( "Couldn't add alarm %s to metric %s.%s", alarm_name, metric_namespace, metric_name, ) raise else: return alarm ``` + For API details, see [PutMetricAlarm](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/PutMetricAlarm) in *AWS SDK for Python (Boto3) API Reference*. ### `PutMetricData` The following code example shows how to use `PutMetricData`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def put_metric_data(self, namespace, name, value, unit): """ Sends a single data value to CloudWatch for a metric. This metric is given a timestamp of the current UTC time. :param namespace: The namespace of the metric. :param name: The name of the metric. :param value: The value of the metric. :param unit: The unit of the metric. """ try: metric = self.cloudwatch_resource.Metric(namespace, name) metric.put_data( Namespace=namespace, MetricData=[{"MetricName": name, "Value": value, "Unit": unit}], ) logger.info("Put data for metric %s.%s", namespace, name) except ClientError: logger.exception("Couldn't put data for metric %s.%s", namespace, name) raise ``` Put a set of data into a CloudWatch metric. ``` class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def put_metric_data_set(self, namespace, name, timestamp, unit, data_set): """ Sends a set of data to CloudWatch for a metric. All of the data in the set have the same timestamp and unit. :param namespace: The namespace of the metric. :param name: The name of the metric. :param timestamp: The UTC timestamp for the metric. :param unit: The unit of the metric. :param data_set: The set of data to send. This set is a dictionary that contains a list of values and a list of corresponding counts. The value and count lists must be the same length. """ try: metric = self.cloudwatch_resource.Metric(namespace, name) metric.put_data( Namespace=namespace, MetricData=[ { "MetricName": name, "Timestamp": timestamp, "Values": data_set["values"], "Counts": data_set["counts"], "Unit": unit, } ], ) logger.info("Put data set for metric %s.%s.", namespace, name) except ClientError: logger.exception("Couldn't put data set for metric %s.%s.", namespace, name) raise ``` + For API details, see [PutMetricData](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/PutMetricData) in *AWS SDK for Python (Boto3) API Reference*. ### `StartOTelEnrichment` The following code example shows how to use `StartOTelEnrichment`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def start_otel_enrichment(self): """ Turns on OTel enrichment for the account. Once enrichment is running, CloudWatch vended metrics that carry a resource identifier dimension, such as the EC2 CPUUtilization metric with its InstanceId dimension, are decorated with resource ARN and resource tag labels and become queryable with PromQL. Resource tags on telemetry must already be enabled for the account before you call this operation. """ try: # Boto3 splits the OTel prefix when it converts the StartOTelEnrichment # operation name to snake case, so the method is start_o_tel_enrichment. self.cloudwatch_client.start_o_tel_enrichment() logger.info("Started OTel enrichment for this account.") except ClientError: logger.exception("Couldn't start OTel enrichment.") raise ``` + For API details, see [StartOTelEnrichment](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/StartOTelEnrichment) in *AWS SDK for Python (Boto3) API Reference*. ### `StopOTelEnrichment` The following code example shows how to use `StopOTelEnrichment`. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). ``` class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def stop_otel_enrichment(self): """ Turns off OTel enrichment for the account. Existing PromQL alarms are not deleted, but vended metrics stop being enriched with resource ARN and tag labels, so queries that select on those labels stop matching. """ try: self.cloudwatch_client.stop_o_tel_enrichment() logger.info("Stopped OTel enrichment for this account.") except ClientError: logger.exception("Couldn't stop OTel enrichment.") raise ``` + For API details, see [StopOTelEnrichment](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/StopOTelEnrichment) in *AWS SDK for Python (Boto3) API Reference*. ## Scenarios ### Manage custom metrics and alarms The following code example shows how to: + Create an alarm to watch a single CloudWatch metric. + Put data into the metric with `PutMetricData` and trigger the alarm. + Get data from the alarm. + Delete the alarm. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). Create a class that wraps CloudWatch operations. ``` from datetime import datetime, timedelta, timezone import logging from pprint import pprint import random import time import boto3 from botocore.exceptions import ClientError logger = logging.getLogger(__name__) class CloudWatchWrapper: """Encapsulates Amazon CloudWatch functions.""" def __init__(self, cloudwatch_resource): """ :param cloudwatch_resource: A Boto3 CloudWatch resource. """ self.cloudwatch_resource = cloudwatch_resource def put_metric_data_set(self, namespace, name, timestamp, unit, data_set): """ Sends a set of data to CloudWatch for a metric. All of the data in the set have the same timestamp and unit. :param namespace: The namespace of the metric. :param name: The name of the metric. :param timestamp: The UTC timestamp for the metric. :param unit: The unit of the metric. :param data_set: The set of data to send. This set is a dictionary that contains a list of values and a list of corresponding counts. The value and count lists must be the same length. """ try: metric = self.cloudwatch_resource.Metric(namespace, name) metric.put_data( Namespace=namespace, MetricData=[ { "MetricName": name, "Timestamp": timestamp, "Values": data_set["values"], "Counts": data_set["counts"], "Unit": unit, } ], ) logger.info("Put data set for metric %s.%s.", namespace, name) except ClientError: logger.exception("Couldn't put data set for metric %s.%s.", namespace, name) raise def create_metric_alarm( self, metric_namespace, metric_name, alarm_name, stat_type, period, eval_periods, threshold, comparison_op, ): """ Creates an alarm that watches a metric. :param metric_namespace: The namespace of the metric. :param metric_name: The name of the metric. :param alarm_name: The name of the alarm. :param stat_type: The type of statistic the alarm watches. :param period: The period in which metric data are grouped to calculate statistics. :param eval_periods: The number of periods that the metric must be over the alarm threshold before the alarm is set into an alarmed state. :param threshold: The threshold value to compare against the metric statistic. :param comparison_op: The comparison operation used to compare the threshold against the metric. :return: The newly created alarm. """ try: metric = self.cloudwatch_resource.Metric(metric_namespace, metric_name) alarm = metric.put_alarm( AlarmName=alarm_name, Statistic=stat_type, Period=period, EvaluationPeriods=eval_periods, Threshold=threshold, ComparisonOperator=comparison_op, ) logger.info( "Added alarm %s to track metric %s.%s.", alarm_name, metric_namespace, metric_name, ) except ClientError: logger.exception( "Couldn't add alarm %s to metric %s.%s", alarm_name, metric_namespace, metric_name, ) raise else: return alarm def put_metric_data(self, namespace, name, value, unit): """ Sends a single data value to CloudWatch for a metric. This metric is given a timestamp of the current UTC time. :param namespace: The namespace of the metric. :param name: The name of the metric. :param value: The value of the metric. :param unit: The unit of the metric. """ try: metric = self.cloudwatch_resource.Metric(namespace, name) metric.put_data( Namespace=namespace, MetricData=[{"MetricName": name, "Value": value, "Unit": unit}], ) logger.info("Put data for metric %s.%s", namespace, name) except ClientError: logger.exception("Couldn't put data for metric %s.%s", namespace, name) raise def get_metric_statistics(self, namespace, name, start, end, period, stat_types): """ Gets statistics for a metric within a specified time span. Metrics are grouped into the specified period. :param namespace: The namespace of the metric. :param name: The name of the metric. :param start: The UTC start time of the time span to retrieve. :param end: The UTC end time of the time span to retrieve. :param period: The period, in seconds, in which to group metrics. The period must match the granularity of the metric, which depends on the metric's age. For example, metrics that are older than three hours have a one-minute granularity, so the period must be at least 60 and must be a multiple of 60. :param stat_types: The type of statistics to retrieve, such as average value or maximum value. :return: The retrieved statistics for the metric. """ try: metric = self.cloudwatch_resource.Metric(namespace, name) stats = metric.get_statistics( StartTime=start, EndTime=end, Period=period, Statistics=stat_types ) logger.info( "Got %s statistics for %s.", len(stats["Datapoints"]), stats["Label"] ) except ClientError: logger.exception("Couldn't get statistics for %s.%s.", namespace, name) raise else: return stats def get_metric_alarms(self, metric_namespace, metric_name): """ Gets the alarms that are currently watching the specified metric. :param metric_namespace: The namespace of the metric. :param metric_name: The name of the metric. :returns: An iterator that yields the alarms. """ metric = self.cloudwatch_resource.Metric(metric_namespace, metric_name) alarm_iter = metric.alarms.all() logger.info("Got alarms for metric %s.%s.", metric_namespace, metric_name) return alarm_iter def delete_metric_alarms(self, metric_namespace, metric_name): """ Deletes all of the alarms that are currently watching the specified metric. :param metric_namespace: The namespace of the metric. :param metric_name: The name of the metric. """ try: metric = self.cloudwatch_resource.Metric(metric_namespace, metric_name) metric.alarms.delete() logger.info( "Deleted alarms for metric %s.%s.", metric_namespace, metric_name ) except ClientError: logger.exception( "Couldn't delete alarms for metric %s.%s.", metric_namespace, metric_name, ) raise ``` Use the wrapper class to put data in a metric, trigger an alarm that watches the metric, and get data from the alarm. ``` def usage_demo(): print("-" * 88) print("Welcome to the Amazon CloudWatch metrics and alarms demo!") print("-" * 88) logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") cw_wrapper = CloudWatchWrapper(boto3.resource("cloudwatch")) minutes = 20 metric_namespace = "doc-example-metric" metric_name = "page_views" start = datetime.now(timezone.utc) - timedelta(minutes=minutes) print( f"Putting data into metric {metric_namespace}.{metric_name} spanning the " f"last {minutes} minutes." ) for offset in range(0, minutes): stamp = start + timedelta(minutes=offset) cw_wrapper.put_metric_data_set( metric_namespace, metric_name, stamp, "Count", { "values": [ random.randint(bound, bound * 2) for bound in range(offset + 1, offset + 11) ], "counts": [random.randint(1, offset + 1) for _ in range(10)], }, ) alarm_name = "high_page_views" period = 60 eval_periods = 2 print(f"Creating alarm {alarm_name} for metric {metric_name}.") alarm = cw_wrapper.create_metric_alarm( metric_namespace, metric_name, alarm_name, "Maximum", period, eval_periods, 100, "GreaterThanThreshold", ) print(f"Alarm ARN is {alarm.alarm_arn}.") print(f"Current alarm state is: {alarm.state_value}.") print( f"Sending data to trigger the alarm. This requires data over the threshold " f"for {eval_periods} periods of {period} seconds each." ) while alarm.state_value == "INSUFFICIENT_DATA": print("Sending data for the metric.") cw_wrapper.put_metric_data( metric_namespace, metric_name, random.randint(100, 200), "Count" ) alarm.load() print(f"Current alarm state is: {alarm.state_value}.") if alarm.state_value == "INSUFFICIENT_DATA": print(f"Waiting for {period} seconds...") time.sleep(period) else: print("Wait for a minute for eventual consistency of metric data.") time.sleep(period) if alarm.state_value == "OK": alarm.load() print(f"Current alarm state is: {alarm.state_value}.") print( f"Getting data for metric {metric_namespace}.{metric_name} during timespan " f"of {start} to {datetime.now(timezone.utc)} (times are UTC)." ) stats = cw_wrapper.get_metric_statistics( metric_namespace, metric_name, start, datetime.now(timezone.utc), 60, ["Average", "Minimum", "Maximum"], ) print( f"Got {len(stats['Datapoints'])} data points for metric " f"{metric_namespace}.{metric_name}." ) pprint(sorted(stats["Datapoints"], key=lambda x: x["Timestamp"])) print(f"Getting alarms for metric {metric_name}.") alarms = cw_wrapper.get_metric_alarms(metric_namespace, metric_name) for alarm in alarms: print(f"Alarm {alarm.name} is currently in state {alarm.state_value}.") print(f"Deleting alarms for metric {metric_name}.") cw_wrapper.delete_metric_alarms(metric_namespace, metric_name) print("Thanks for watching!") print("-" * 88) ``` + For API details, see the following topics in *AWS SDK for Python (Boto3) API Reference*. + [DeleteAlarms](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DeleteAlarms) + [DescribeAlarmsForMetric](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DescribeAlarmsForMetric) + [DisableAlarmActions](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DisableAlarmActions) + [EnableAlarmActions](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/EnableAlarmActions) + [GetMetricStatistics](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/GetMetricStatistics) + [ListMetrics](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/ListMetrics) + [PutMetricAlarm](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/PutMetricAlarm) + [PutMetricData](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/PutMetricData) ### Send OpenTelemetry metrics and alarm on them with PromQL The following code example shows how to: + Send OTLP metrics to the CloudWatch metrics endpoint with an OpenTelemetry Collector. + Start OpenTelemetry enrichment so CloudWatch correlates those metrics with your resources. + Create an alarm that evaluates a PromQL query across every series the query returns. + Inspect the individual series, called contributors, that put the alarm in ALARM state. + Mute the alarm for a maintenance window, then clean up. **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch#code-examples). Create a class that wraps the CloudWatch OpenTelemetry operations. ``` import logging import time import boto3 from botocore.exceptions import ClientError logger = logging.getLogger(__name__) class CloudWatchOTelWrapper: """Encapsulates the OpenTelemetry-oriented Amazon CloudWatch operations.""" def __init__(self, cloudwatch_client): """ :param cloudwatch_client: A Boto3 CloudWatch client. The OpenTelemetry operations are only available on the client interface, not on the higher-level ``boto3.resource("cloudwatch")`` interface. """ self.cloudwatch_client = cloudwatch_client @classmethod def from_client(cls): """ Creates a wrapper backed by a default CloudWatch client. :return: A CloudWatchOTelWrapper. """ return cls(boto3.client("cloudwatch")) def start_otel_enrichment(self): """ Turns on OTel enrichment for the account. Once enrichment is running, CloudWatch vended metrics that carry a resource identifier dimension, such as the EC2 CPUUtilization metric with its InstanceId dimension, are decorated with resource ARN and resource tag labels and become queryable with PromQL. Resource tags on telemetry must already be enabled for the account before you call this operation. """ try: # Boto3 splits the OTel prefix when it converts the StartOTelEnrichment # operation name to snake case, so the method is start_o_tel_enrichment. self.cloudwatch_client.start_o_tel_enrichment() logger.info("Started OTel enrichment for this account.") except ClientError: logger.exception("Couldn't start OTel enrichment.") raise def get_otel_enrichment_status(self): """ Gets the current OTel enrichment status for the account. :return: The status, either 'Running' or 'Stopped'. """ try: response = self.cloudwatch_client.get_o_tel_enrichment() except ClientError: logger.exception("Couldn't get the OTel enrichment status.") raise else: status = response["Status"] logger.info("OTel enrichment status is %s.", status) return status def create_promql_alarm( self, alarm_name, query, evaluation_interval, pending_period=300, recovery_period=120, description=None, alarm_actions=None, ): """ Creates an alarm that evaluates a PromQL query. A PromQL alarm differs from a classic metric alarm in a few ways. The query can match many series at once, and each matching series is tracked separately as a *contributor*. Instead of counting breaching periods, you specify durations: a contributor moves to ALARM after it breaches continuously for the pending period, and back to OK after it stops breaching for the recovery period. A PromQL alarm starts in the OK state rather than INSUFFICIENT_DATA. The PromQL evaluation parameters live in the EvaluationCriteria union, which is mutually exclusive with the classic MetricName and Metrics parameters. When you use EvaluationCriteria you must also set EvaluationInterval, and you must not set Period, Statistic, Threshold, ComparisonOperator, EvaluationPeriods, DatapointsToAlarm, or TreatMissingData. :param alarm_name: The name of the alarm. Must be unique within the Region. :param query: The PromQL query to evaluate, such as 'avg(cpu_utilization_percent) > 80'. The comparison belongs in the query itself; there is no separate threshold parameter. :param evaluation_interval: How often, in seconds, to run the query. Valid values are 10, 20, 30, and any multiple of 60, up to 3600. :param pending_period: How long, in seconds, a contributor must breach continuously before it moves to ALARM. :param recovery_period: How long, in seconds, a contributor must stop breaching before it moves back to OK. :param description: The description of the alarm. :param alarm_actions: A list of ARNs to notify when the alarm fires, such as an Amazon SNS topic. """ promql_criteria = { "Query": query, "PendingPeriod": pending_period, "RecoveryPeriod": recovery_period, } kwargs = { "AlarmName": alarm_name, "EvaluationCriteria": {"PromQLCriteria": promql_criteria}, "EvaluationInterval": evaluation_interval, } if description is not None: kwargs["AlarmDescription"] = description if alarm_actions is not None: kwargs["AlarmActions"] = alarm_actions try: self.cloudwatch_client.put_metric_alarm(**kwargs) logger.info("Created PromQL alarm %s for query %s.", alarm_name, query) except ClientError: logger.exception("Couldn't create PromQL alarm %s.", alarm_name) raise def describe_alarm_contributors(self, alarm_name): """ Gets the contributors for a PromQL alarm. Each contributor is one series that the alarm's query matched, identified by its label set. This is how you find out *which* hosts, services, or pods are breaching, rather than only that something is. :param alarm_name: The name of the PromQL alarm. :return: The list of contributors. Each contributor has a ContributorId, a ContributorAttributes map of the labels that identify the series, a StateReason, and the time it last changed state. """ contributors = [] try: next_token = None while True: kwargs = {"AlarmName": alarm_name} if next_token is not None: kwargs["NextToken"] = next_token response = self.cloudwatch_client.describe_alarm_contributors(**kwargs) contributors.extend(response["AlarmContributors"]) next_token = response.get("NextToken") if not next_token: break except ClientError: logger.exception("Couldn't get contributors for alarm %s.", alarm_name) raise else: logger.info( "Got %s contributors for alarm %s.", len(contributors), alarm_name ) return contributors def put_alarm_mute_rule( self, name, expression, duration, alarm_names=None, timezone=None, description=None, ): """ Creates or updates an alarm mute rule. While a mute rule is active the targeted alarms keep evaluating and keep transitioning between states, but their configured actions do not fire. This is the supported way to suppress notifications during a known maintenance window instead of disabling alarm actions and hoping someone remembers to turn them back on. :param name: The name of the mute rule. :param expression: When the rule activates. For a recurring window, use a five-field cron expression, 'cron(Minutes Hours Day-of-month Month Day-of-week)', such as 'cron(0 2 * * SUN)' for every Sunday at 2:00 AM. Note that this is five fields, not the six that Amazon EventBridge uses. For a one-time window, use 'at(yyyy-MM-ddThh:mm)', such as 'at(2026-09-05T02:00)'. :param duration: How long the mute window lasts once it activates, in ISO 8601 duration format, from 'PT1M' (one minute) to 'P15D' (15 days). For example, 'PT2H' is two hours and 'P2DT12H' is two days and 12 hours. :param alarm_names: The names of up to 100 alarms to mute. If omitted, the rule applies to all alarms in the account. :param timezone: The time zone the expression is evaluated in, such as 'America/Los_Angeles'. :param description: The description of the mute rule. """ schedule = {"Expression": expression, "Duration": duration} if timezone is not None: schedule["Timezone"] = timezone kwargs = {"Name": name, "Rule": {"Schedule": schedule}} if alarm_names is not None: kwargs["MuteTargets"] = {"AlarmNames": alarm_names} if description is not None: kwargs["Description"] = description try: self.cloudwatch_client.put_alarm_mute_rule(**kwargs) logger.info("Put alarm mute rule %s.", name) except ClientError: logger.exception("Couldn't put alarm mute rule %s.", name) raise def get_alarm_mute_rule(self, name): """ Gets the full configuration of an alarm mute rule, including its schedule, the alarms it targets, and whether it is currently SCHEDULED, ACTIVE, or EXPIRED. :param name: The name of the mute rule. :return: The mute rule. """ try: response = self.cloudwatch_client.get_alarm_mute_rule( AlarmMuteRuleName=name ) except ClientError: logger.exception("Couldn't get alarm mute rule %s.", name) raise else: logger.info("Got alarm mute rule %s.", name) return response def list_alarm_mute_rules(self, alarm_name=None, statuses=None): """ Lists alarm mute rules in the account. :param alarm_name: When specified, only rules that target this alarm are returned. :param statuses: When specified, only rules in these statuses are returned. Valid values are 'SCHEDULED', 'ACTIVE', and 'EXPIRED'. :return: The list of mute rule summaries. """ summaries = [] try: next_token = None while True: kwargs = {} if alarm_name is not None: kwargs["AlarmName"] = alarm_name if statuses is not None: kwargs["Statuses"] = statuses if next_token is not None: kwargs["NextToken"] = next_token response = self.cloudwatch_client.list_alarm_mute_rules(**kwargs) summaries.extend(response.get("AlarmMuteRuleSummaries", [])) next_token = response.get("NextToken") if not next_token: break except ClientError: logger.exception("Couldn't list alarm mute rules.") raise else: logger.info("Got %s alarm mute rules.", len(summaries)) return summaries def delete_alarm_mute_rule(self, name): """ Deletes an alarm mute rule. :param name: The name of the mute rule. """ try: self.cloudwatch_client.delete_alarm_mute_rule(AlarmMuteRuleName=name) logger.info("Deleted alarm mute rule %s.", name) except ClientError: logger.exception("Couldn't delete alarm mute rule %s.", name) raise def stop_otel_enrichment(self): """ Turns off OTel enrichment for the account. Existing PromQL alarms are not deleted, but vended metrics stop being enriched with resource ARN and tag labels, so queries that select on those labels stop matching. """ try: self.cloudwatch_client.stop_o_tel_enrichment() logger.info("Stopped OTel enrichment for this account.") except ClientError: logger.exception("Couldn't stop OTel enrichment.") raise def delete_alarms(self, alarm_names): """ Deletes the specified alarms. :param alarm_names: The names of the alarms to delete. """ try: self.cloudwatch_client.delete_alarms(AlarmNames=alarm_names) logger.info("Deleted alarms %s.", ", ".join(alarm_names)) except ClientError: logger.exception("Couldn't delete alarms %s.", ", ".join(alarm_names)) raise ``` Use the wrapper class to alarm on OpenTelemetry metrics with a PromQL query, inspect the contributors to the alarm, and mute it. ``` def usage_demo(): """ Walks through the OpenTelemetry metrics workflow in CloudWatch: turn on enrichment, alarm on a PromQL query, inspect the contributors that matched, mute the alarm for a maintenance window, then clean up. This scenario assumes OpenTelemetry metrics are already flowing into the account, either from an OpenTelemetry collector, the CloudWatch agent, or the ADOT SDK. """ print("-" * 88) print("Welcome to the Amazon CloudWatch OpenTelemetry metrics demo!") print("-" * 88) logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") cw = CloudWatchOTelWrapper.from_client() alarm_name = "doc-example-promql-high-cpu" mute_rule_name = "doc-example-maintenance-window" print("Checking whether OTel enrichment is on for this account.") status = cw.get_otel_enrichment_status() started_enrichment_here = False if status == "Stopped": print("Enrichment is stopped. Starting it so vended metrics accept PromQL.") cw.start_otel_enrichment() started_enrichment_here = True else: print("Enrichment is already running. Leaving it alone.") query = 'avg by (host_name) (cpu_utilization_percent{service_name="checkout"}) > 80' print(f"\nCreating a PromQL alarm on: {query}") cw.create_promql_alarm( alarm_name, query, evaluation_interval=30, pending_period=300, recovery_period=120, description="Average CPU over 80% per host for the checkout service.", ) print( "The alarm evaluates every 30 seconds. A host moves to ALARM after breaching " "for 300 seconds straight, and back to OK after 120 seconds clean." ) print("\nWaiting a moment for the first evaluation, then listing contributors.") time.sleep(30) contributors = cw.describe_alarm_contributors(alarm_name) if not contributors: print( "No contributors yet. The query matched no series, which usually means " "no OTel metrics with these labels have arrived. Send some OTel metrics " "through the OTLP endpoint and run this again." ) for contributor in contributors: labels = ", ".join( f"{key}={value}" for key, value in sorted(contributor["ContributorAttributes"].items()) ) print(f" {contributor['ContributorId']}: {labels}") print(f" reason: {contributor['StateReason']}") print(f"\nMuting {alarm_name} for a weekly two-hour maintenance window.") cw.put_alarm_mute_rule( mute_rule_name, expression="cron(0 2 * * SUN)", duration="PT2H", alarm_names=[alarm_name], timezone="America/Los_Angeles", description="Suppress checkout CPU pages during Sunday patching.", ) rule = cw.get_alarm_mute_rule(mute_rule_name) print(f"Mute rule status is {rule.get('Status')}.") print( "While the window is active the alarm keeps evaluating and still changes " "state; only its actions are suppressed." ) print(f"\nMute rules targeting {alarm_name}:") for summary in cw.list_alarm_mute_rules(alarm_name=alarm_name): print(f" {summary.get('AlarmMuteRuleArn')} ({summary.get('Status')})") print("\nCleaning up.") cw.delete_alarm_mute_rule(mute_rule_name) cw.delete_alarms([alarm_name]) if started_enrichment_here: print("Stopping OTel enrichment, since this demo started it.") cw.stop_otel_enrichment() print("\nThanks for watching!") print("-" * 88) ``` Configure an OpenTelemetry Collector to send the OTLP metrics that this example alarms on to the CloudWatch metrics endpoint. Metric ingestion over OTLP is not an AWS SDK operation, so this half of the example is collector configuration rather than SDK code. ``` # Purpose # # An OpenTelemetry Collector configuration that sends OTLP metrics to the Amazon # CloudWatch metrics endpoint. Once metrics land in CloudWatch they are queryable with # PromQL in Query Studio, and you can alarm on them with the PromQL alarm operations # shown in cloudwatch_otel.py. # # Metric ingestion over OTLP is deliberately NOT an AWS SDK operation. There is no # boto3 call that sends OTLP metrics. You send them with one of the following, in # rough order of how integrated the CloudWatch experience is: # # 1. The CloudWatch agent (recommended). An AWS-managed OpenTelemetry Collector with # CloudWatch components pre-built. Adds entity correlation, runtime metrics, and # Container Insights support. # 2. An upstream OpenTelemetry Collector. What this file configures. # 3. A custom OpenTelemetry Collector build. # 4. The AWS Distro for OpenTelemetry (ADOT) SDK, with no collector at all. # # Prerequisites # # * An OpenTelemetry Collector release that includes the sigv4authextension. The # contrib distribution does. See # https://github.com/open-telemetry/opentelemetry-collector-releases/releases # * AWS credentials the collector can resolve. On Amazon EC2, attach the # CloudWatchAgentServerPolicy managed policy to the instance role. On Amazon EKS, # bind that policy to the collector's service account with IRSA. On premises, run # `aws configure` for an IAM user that has the same policy. # # Run the collector with: # # otelcol-contrib --config otlp_collector_config.yaml # # Then point your instrumented application at http://localhost:4318. # # For more information, see # https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLPEndpoint.html receivers: # The CloudWatch OTLP endpoints are HTTP 1.1 only and do not support gRPC. You can # still accept gRPC from your applications here and let the collector translate, but # this example keeps the receiver HTTP-only to mirror what is sent upstream. otlp: protocols: http: endpoint: "0.0.0.0:4318" processors: # Batch to stay inside the endpoint's per-request limits: 1 MB uncompressed and # 1,000 datapoints, counted as the sum across ResourceMetrics, ScopeMetrics, and # Metrics. A batch of 200 leaves comfortable headroom. batch: send_batch_size: 200 timeout: 10s exporters: otlphttp: tls: insecure: false # Pattern: https://monitoring.{region}.amazonaws.com/v1/metrics endpoint: "https://monitoring.us-east-1.amazonaws.com/v1/metrics" # Only gzip and none are supported. compression: gzip auth: authenticator: sigv4auth extensions: # SigV4 is the recommended authentication method, and the only one supported for # traces. For hosts outside AWS you can instead use the bearertokenauth extension # against the metrics or logs endpoints; see # https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLP-MetricsBearerTokenAuth.html sigv4auth: # The metrics endpoint signs as "monitoring". The logs endpoint signs as "logs" # and the traces endpoint as "xray". service: "monitoring" region: "us-east-1" service: extensions: [sigv4auth] pipelines: metrics: receivers: [otlp] processors: [batch] exporters: [otlphttp] # Endpoint limits worth designing around, all per account and Region: # # Maximum TPS 500 # New series creation rate 1,000,000 per 10-minute window # Maximum request size 1 MB uncompressed # Maximum datapoint count 1,000 per request # Maximum metadata size 40 KB of labels and values per series per datapoint # Maximum label count 150 across Resource, Scope, and Datapoint attributes # Timestamp window at most 10 minutes in the future, 14 days in the past # # Exceeding the TPS or new-series limits returns 429. The size, count, metadata, # label, and timestamp limits return 400, and a request whose metrics are only # partially invalid returns 200 with the valid metrics ingested. ``` + For API details, see the following topics in *AWS SDK for Python (Boto3) API Reference*. + [DeleteAlarmMuteRule](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DeleteAlarmMuteRule) + [DeleteAlarms](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DeleteAlarms) + [DescribeAlarmContributors](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/DescribeAlarmContributors) + [GetAlarmMuteRule](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/GetAlarmMuteRule) + [GetOTelEnrichment](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/GetOTelEnrichment) + [ListAlarmMuteRules](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/ListAlarmMuteRules) + [PutAlarmMuteRule](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/PutAlarmMuteRule) + [PutMetricAlarm](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/PutMetricAlarm) + [StartOTelEnrichment](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/StartOTelEnrichment) + [StopOTelEnrichment](https://docs.aws.amazon.com/goto/boto3/monitoring-2010-08-01/StopOTelEnrichment)