Coverage for mcp/metric_readers/cloudwatch.py: 100%
30 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-15 15:07 +0000
1"""Read-only CloudWatch datapoint reader.
3This module turns a single CloudWatch ``GetMetricStatistics`` request into one
4numeric scalar a metric-threshold check can read. It has two pieces:
6* :func:`select_most_recent` — a pure helper that, given a non-empty list of
7 CloudWatch datapoints, picks the one with the latest timestamp and returns
8 its statistic value alongside that timestamp in ISO-8601 form. No I/O, so it
9 is trivial to test in isolation.
10* :func:`get_datapoint` — the thin boto3 wrapper. It constructs a
11 region-scoped CloudWatch client, issues one read-only
12 ``GetMetricStatistics`` call bounded to the requested window, and hands the
13 returned datapoints to :func:`select_most_recent`. Every failure mode —
14 an empty result, an unreachable endpoint, a credentials problem, or any
15 other client error — is translated into a :class:`MetricReaderError` with a
16 stable code so the calling tool can render a structured error envelope
17 instead of crashing.
19``boto3`` is imported lazily inside :func:`get_datapoint` (mirroring the
20lazy-import convention used by the SSM helpers and the stacks CLI) so this
21module's import surface stays free of the SDK and tests can monkeypatch
22``boto3.client`` without dragging the SDK into unrelated test runs.
23"""
25from __future__ import annotations
27from datetime import datetime
28from typing import Any
30from .shape import ErrorCode, MetricReaderError
32# CloudWatch ``Error.Code`` values that mean "the request was rejected because
33# of credentials or permissions" rather than a transient transport failure.
34# These map to the ``unauthorized`` discriminator so an operator can tell an
35# access problem apart from an unreachable endpoint without reading the message.
36_UNAUTHORIZED_ERROR_CODES = frozenset(
37 {
38 "AccessDenied",
39 "AccessDeniedException",
40 "AuthFailure",
41 "ExpiredToken",
42 "ExpiredTokenException",
43 "InvalidAccessKeyId",
44 "InvalidClientTokenId",
45 "SignatureDoesNotMatch",
46 "UnauthorizedOperation",
47 "UnrecognizedClientException",
48 }
49)
52def select_most_recent(datapoints: list[dict[str, Any]], statistic: str) -> tuple[float, str]:
53 """Pick the latest datapoint and return its value and ISO timestamp.
55 Given a non-empty list of CloudWatch datapoints (each a dict carrying a
56 ``Timestamp`` and the requested statistic key), return a
57 ``(value, iso_timestamp)`` tuple where ``value`` is the chosen
58 datapoint's ``statistic`` reading coerced to a float and ``iso_timestamp``
59 is its timestamp in ISO-8601 form. Selection is deterministic: the
60 datapoint with the maximum ``Timestamp`` wins, and ``GetMetricStatistics``
61 never emits two datapoints sharing a timestamp within one period.
63 Args:
64 datapoints: A non-empty list of CloudWatch datapoint dicts.
65 statistic: The statistic key to read from the chosen datapoint
66 (for example ``"Average"`` or ``"Sum"``).
68 Returns:
69 A ``(value, iso_timestamp)`` tuple for the latest datapoint.
70 """
71 most_recent = max(datapoints, key=lambda dp: dp["Timestamp"])
72 value = float(most_recent[statistic])
73 timestamp = most_recent["Timestamp"]
74 iso_timestamp = timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp)
75 return value, iso_timestamp
78def get_datapoint(
79 *,
80 metric_name: str,
81 namespace: str,
82 dimensions: dict[str, str] | None,
83 region: str,
84 period: int,
85 statistic: str,
86 start_time: datetime,
87 end_time: datetime,
88) -> tuple[float, str]:
89 """Read one CloudWatch datapoint for a metric in a named region.
91 Constructs a region-scoped CloudWatch client and issues a single
92 read-only ``GetMetricStatistics`` request bounded to
93 ``[start_time, end_time]`` for the supplied metric, namespace,
94 dimensions, period, and statistic. The dimensions mapping is passed to
95 CloudWatch unchanged, as a list of ``{"Name": ..., "Value": ...}`` pairs.
97 Args:
98 metric_name: The CloudWatch metric name.
99 namespace: The CloudWatch namespace the metric lives in.
100 dimensions: Name/value dimension pairs, or ``None`` for no dimensions.
101 region: The AWS region to scope the CloudWatch client to.
102 period: The aggregation period, in seconds.
103 statistic: The statistic to request (for example ``"Average"``).
104 start_time: The inclusive start of the lookback window.
105 end_time: The end of the lookback window.
107 Returns:
108 A ``(value, iso_timestamp)`` tuple for the latest datapoint in the
109 window, as produced by :func:`select_most_recent`.
111 Raises:
112 MetricReaderError: With :attr:`ErrorCode.NO_DATAPOINTS` when the
113 window holds no datapoints, or :attr:`ErrorCode.AWS_UNREACHABLE`
114 (carrying a ``kind`` discriminator of ``unreachable``,
115 ``unauthorized``, or ``client_error``) when the request cannot be
116 completed.
117 """
118 import boto3
119 from botocore.exceptions import (
120 BotoCoreError,
121 ClientError,
122 EndpointConnectionError,
123 )
125 dimension_pairs = [{"Name": key, "Value": value} for key, value in (dimensions or {}).items()]
127 try:
128 client = boto3.client("cloudwatch", region_name=region)
129 response = client.get_metric_statistics(
130 Namespace=namespace,
131 MetricName=metric_name,
132 Dimensions=dimension_pairs,
133 StartTime=start_time,
134 EndTime=end_time,
135 Period=period,
136 Statistics=[statistic],
137 )
138 except EndpointConnectionError as exc:
139 # A subclass of BotoCoreError; caught first so it keeps its own,
140 # more specific "unreachable" classification.
141 raise MetricReaderError(
142 ErrorCode.AWS_UNREACHABLE,
143 {"kind": "unreachable", "region": region, "message": str(exc)},
144 ) from exc
145 except ClientError as exc:
146 aws_error_code = exc.response.get("Error", {}).get("Code", "")
147 kind = "unauthorized" if aws_error_code in _UNAUTHORIZED_ERROR_CODES else "client_error"
148 raise MetricReaderError(
149 ErrorCode.AWS_UNREACHABLE,
150 {
151 "kind": kind,
152 "region": region,
153 "aws_error_code": aws_error_code,
154 "message": str(exc),
155 },
156 ) from exc
157 except BotoCoreError as exc:
158 raise MetricReaderError(
159 ErrorCode.AWS_UNREACHABLE,
160 {"kind": "unreachable", "region": region, "message": str(exc)},
161 ) from exc
163 datapoints = response.get("Datapoints", [])
164 if not datapoints:
165 raise MetricReaderError(
166 ErrorCode.NO_DATAPOINTS,
167 {
168 "metric_name": metric_name,
169 "namespace": namespace,
170 "region": region,
171 "statistic": statistic,
172 },
173 )
175 return select_most_recent(datapoints, statistic)