Amazon CloudWatch — Metrics with boto3 & the Embedded Metric Format¶

import boto3
from botocore.config import Config

cw = boto3.client("cloudwatch", config=Config(retries={"mode": "standard", "max_attempts": 5}))

import datetime as dt

DIMS = [{"Name": "Service", "Value": "orders-api"}]
now = dt.datetime.now(dt.timezone.utc)

cw.put_metric_data(Namespace="QA/Orders", MetricData=[                       # up to 1000 items, 1 MB per request
    {"MetricName": "Latency", "Value": 120.5, "Unit": "Milliseconds", "Timestamp": now, "Dimensions": DIMS},
    {"MetricName": "Latency", "Values": [80.0, 120.0, 900.0], "Counts": [40, 9, 1],      # 50 observations, 3 numbers sent
     "Unit": "Milliseconds", "Timestamp": now, "Dimensions": DIMS},
    {"MetricName": "QueueDepth", "Unit": "Count", "Timestamp": now, "Dimensions": DIMS,   # pre-aggregated: no percentiles
     "StatisticValues": {"SampleCount": 12, "Sum": 340, "Minimum": 5, "Maximum": 80}},
    {"MetricName": "Heartbeat", "Value": 1, "Unit": "Count", "Dimensions": DIMS,
     "StorageResolution": 1},                                                         # stored per second, custom metrics only
])

# app/metrics.py (excerpt: to_dimensions, check_value, Sample and the UNITS set are in the same module)
def aggregate(samples: list[Sample]) -> list[dict]:
    """Group samples by metric identity and time bucket into Values/Counts items (raw values: percentiles keep working)."""
    groups: dict[tuple, Counter] = defaultdict(Counter)
    for s in samples:
        bucket = 1 if s.high_res else 60                       # high-resolution data keeps 1-second buckets
        start = int(s.timestamp // bucket) * bucket
        groups[(s.name, s.unit, s.dimensions, start, s.high_res)][check_value(s.value, s.unit)] += 1
    items = []
    for (name, unit, dims, start, high_res), counts in groups.items():
        values = sorted(counts)
        for i in range(0, len(values), MAX_VALUES_PER_DATUM):      # one item holds at most 150 distinct values
            chunk = values[i:i + MAX_VALUES_PER_DATUM]
            item = {"MetricName": name, "Unit": unit, "Dimensions": to_dimensions(dict(dims)),
                    "Timestamp": datetime.fromtimestamp(start, tz=timezone.utc),
                    "Values": chunk, "Counts": [float(counts[v]) for v in chunk]}
            if high_res:
                item["StorageResolution"] = 1
            items.append(item)
    return items


def batches(items: list[dict], max_items: int = 1000, max_bytes: int = 800_000) -> Iterator[list[dict]]:
    """Split items so that no request exceeds the item count or an approximate payload size (limit: 1 MB)."""
    current, size = [], 0
    for item in items:
        item_size = len(json.dumps(item, default=str))
        if current and (len(current) >= max_items or size + item_size > max_bytes):
            yield current
            current, size = [], 0
        current.append(item)
        size += item_size
    if current:
        yield current


class MetricsEmitter:
    """`client` is anything with put_metric_data(**kwargs): a boto3 client, a Stubber-wrapped client, a fake."""

    def __init__(self, client, namespace, *, default_dimensions=None, flush_at=500, clock=time.time):
        self._client, self.namespace, self._defaults = client, namespace, dict(default_dimensions or {})
        self._flush_at, self._clock = flush_at, clock
        self._samples: list[Sample] = []
        self._unsent: list[dict] = []
        self._lock = threading.Lock()

    def record(self, name, value, unit="None", *, high_res=False, **dims):
        check_value(value, unit)                                # fail at the call site, not at flush time
        merged = tuple(sorted({**self._defaults, **dims}.items()))
        with self._lock:
            self._samples.append(Sample(name, value, unit, merged, self._clock(), high_res))
            full = len(self._samples) >= self._flush_at
        if full:
            self.flush()

    def count(self, name, value=1, **dims):
        self.record(name, value, "Count", **dims)

    @contextmanager
    def timer(self, name, **dims):
        started = time.perf_counter()
        try:
            yield
        finally:
            self.record(name, (time.perf_counter() - started) * 1000, "Milliseconds", **dims)

    def flush(self) -> int:
        """Send everything buffered. Unsent items stay buffered when a request fails and the error propagates."""
        with self._lock:
            pending, self._samples = self._samples, []
            items, self._unsent = self._unsent + aggregate(pending), []
        chunks = list(batches(items))
        sent = 0
        for i, chunk in enumerate(chunks):
            try:
                self._client.put_metric_data(Namespace=self.namespace, MetricData=chunk)
            except Exception:
                with self._lock:
                    self._unsent = [item for c in chunks[i:] for item in c] + self._unsent
                raise
            sent += len(chunk)
        return sent

    def __enter__(self): return self
    def __exit__(self, *exc): self.flush()

with MetricsEmitter(cw, "QA/Orders", default_dimensions={"Service": "orders-api"}) as metrics:
    metrics.count("OrdersCreated")                          # Count, summed per minute
    with metrics.timer("CreateOrder", Route="/orders"):     # Milliseconds, percentiles available
        ...
# leaving the block flushes; in a long-running service call metrics.flush() from a timer thread or a signal handler

# app/reader.py (excerpt)
def fetch(client, queries, start, end) -> dict[str, list[tuple[datetime, float]]]:
    """GetMetricData with NextToken followed; {query_id: [(timestamp, value), ...]} oldest first."""
    series = {q["Id"]: [] for q in queries if q.get("ReturnData", True)}
    paginator = client.get_paginator("get_metric_data")              # follows NextToken
    for page in paginator.paginate(MetricDataQueries=queries, StartTime=start, EndTime=end,
                                   ScanBy="TimestampAscending"):
        for result in page["MetricDataResults"]:
            if result["StatusCode"] in ("InternalError", "Forbidden"):
                raise RuntimeError(f"{result['Id']}: {result['StatusCode']} {result.get('Messages')}")
            series.setdefault(result["Id"], []).extend(zip(result["Timestamps"], result["Values"]))
    return {k: sorted(v) for k, v in series.items()}                  # pages of one query arrive separately

def stat(query_id, name, stat="Sum"):
    return {"Id": query_id, "ReturnData": False, "MetricStat": {
        "Metric": {"Namespace": "QA/Orders", "MetricName": name, "Dimensions": DIMS}, "Period": 60, "Stat": stat}}

queries = [stat("errors", "OrdersFailed"), stat("total", "OrdersCreated"),
           {"Id": "rate", "Expression": "100 * errors / total", "Label": "failure %"}]
series = fetch(cw, queries, now - dt.timedelta(hours=1), now)        # only "rate" is returned

paginator = cw.get_paginator("list_metrics")
for page in paginator.paginate(Namespace="QA/Orders", Dimensions=[{"Name": "Service"}], RecentlyActive="PT3H"):
    for metric in page["Metrics"]:
        print(metric["MetricName"], {d["Name"]: d["Value"] for d in metric["Dimensions"]})

{"_aws": {"Timestamp": 1700000000000, "CloudWatchMetrics": [{"Namespace": "App/Orders",
  "Dimensions": [["Route"]], "Metrics": [{"Name": "Latency", "Unit": "Milliseconds"}, {"Name": "Orders", "Unit": "Count", "StorageResolution": 1}]}]},
 "Route": "/orders", "Latency": [123.4, 98.0], "Orders": 1, "order_id": "o-1"}

# app/emf.py (core of emf_line; validation of units, counts and size is in the module)
root = {**(properties or {}), **dimensions}                   # properties and dimension values live at the root
for name, (value, unit) in metrics.items():
    root[name] = value                                        # a number, or a list of numbers
root["_aws"] = {"Timestamp": timestamp_ms or int(time.time() * 1000),
                "CloudWatchMetrics": [{"Namespace": namespace, "Dimensions": [sorted(dimensions)],
                                       "Metrics": [{"Name": n, "Unit": u} for n, (_, u) in metrics.items()]}]}
print(json.dumps(root, separators=(",", ":")))                # one line = one log event; stdout is shipped by the driver

import os
os.environ["AWS_EMF_ENVIRONMENT"] = "Local"           # print EMF lines to stdout; omit on Lambda/ECS/EC2 to auto-detect
os.environ["AWS_EMF_NAMESPACE"] = "App/Orders"
os.environ["AWS_EMF_SERVICE_NAME"] = "orders-api"

from aws_embedded_metrics import metric_scope
from aws_embedded_metrics.storage_resolution import StorageResolution


@metric_scope
def handle_order(order_id: str, metrics):
    metrics.set_dimensions({"Route": "/orders"})         # replaces the DEFAULT dimension set (LogGroup, ServiceName, ServiceType)
    metrics.put_metric("Latency", 123.4, "Milliseconds")
    metrics.put_metric("Latency", 98.0, "Milliseconds")   # same name twice becomes an array of values
    metrics.put_metric("Orders", 1, "Count", StorageResolution.HIGH)
    metrics.set_property("order_id", order_id)            # not a dimension