Streaming RDS CloudWatch Metrics to Kafka with aiokafka

This page builds an async poller that pulls RDS CloudWatch metrics with aioboto3, turns each data point into a cost-signal record, and publishes it to a Kafka topic through an aiokafka producer with per-instance keying, durable acknowledgements, and clean backpressure handling.

Back to: Real-Time Metric Streaming Setup

CloudWatch is the authoritative source for RDS resource consumption — CPUUtilization, ReadIOPS/WriteIOPS, and FreeableMemory are the four signals that most directly track compute and I/O spend — but the CloudWatch API is a request/response surface, not a stream. To feed a live cost pipeline you need something in between: a poller that calls GetMetricData on a tight cadence and fans the results into a durable log where aggregators can consume them independently. Kafka is that log, and aiokafka lets the producer share the same event loop as the aioboto3 client so a single process can poll many instances without threads. The records this poller emits are the same shape consumed downstream by schema validation for billing data and by the concurrency-controlled async usage parsing workflows that pull provider cost APIs alongside this telemetry.

The polling window is a rolling one: every tick asks CloudWatch for the last few minutes of one-minute-period statistics, so the effective sampling resolution is bounded by the metric period, not the poll interval:

feffective=1max⁡(poll_interval, metric_period)f_{\text{effective}} = \frac{1}{\max(\text{poll\_interval},\ \text{metric\_period})}

Polling faster than the metric period only re-reads the same data points, so the interval below is aligned to the 60-second period.

Prerequisites

Before running the poller, confirm the following are in place.

  • AWS IAM: the identity running the job needs read-only CloudWatch access. Attach a least-privilege policy scoped to cloudwatch:GetMetricData (and cloudwatch:ListMetrics for discovery); this is the same least-privilege posture described in scoping IAM least privilege for Cost Explorer.

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["cloudwatch:GetMetricData", "cloudwatch:ListMetrics"],
      "Resource": "*"
    }
  ]
}

    GetMetricData does not support resource-level ARNs, so Resource: "*" is expected; constrain it with a cloudwatch:namespace condition key if your governance requires it.

  • Python: 3.10 or newer (the code uses asyncio.TaskGroup-style structured shutdown and modern typing).

  • Kafka: a reachable broker (0.10+) and a pre-created topic. Do not rely on auto-topic-creation in production — create it with an explicit partition count so keyed ordering is stable.

  • Libraries: install the async AWS and Kafka clients.

    pip install "aioboto3>=13.0" "aiokafka>=0.11" "pydantic>=2.7"

  • Credentials: resolved from the standard AWS chain (os.environ, instance profile, or IRSA on EKS). Never hard-code keys; the producer reads only KAFKA_BOOTSTRAP from the environment.

Step-by-Step Implementation

The poller acquires an aioboto3 CloudWatch client and an aiokafka producer, then loops: fetch a batch of metric data points, shape each into a cost-signal record, and publish keyed by DB instance. Backpressure is handled by awaiting the send future when the producer buffer fills, and shutdown drains in-flight sends before exiting.

Step 1 — Define the cost-signal record

The record is a small, frozen Pydantic model so the serialized payload is stable and self-describing on the wire. Keying on db_instance guarantees all metrics for one instance land on the same partition, preserving per-instance ordering for stateful aggregators.

from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, ConfigDict

class MetricSignal(BaseModel):
    """One CloudWatch data point shaped as a cost signal."""
    model_config = ConfigDict(frozen=True)

    db_instance: str          # partition key, e.g. "prod-orders-1"
    metric: str               # CPUUtilization | ReadIOPS | WriteIOPS | FreeableMemory
    value: float
    unit: str                 # Percent | Count/Second | Bytes
    timestamp: datetime       # UTC, from CloudWatch
    account: str

    def to_bytes(self) -> bytes:
        # model_dump_json emits ISO-8601 timestamps; encode for Kafka.
        return self.model_dump_json().encode("utf-8")

# Expected:
#   MetricSignal(db_instance="prod-orders-1", metric="CPUUtilization",
#                value=63.4, unit="Percent", timestamp=..., account="1234").to_bytes()
#   -> b'{"db_instance":"prod-orders-1","metric":"CPUUtilization",...}'

Step 2 — Poll CloudWatch with get_metric_data

GetMetricData takes a batch of up to 500 queries in one call, so a single request covers every metric across every instance. Each query needs a unique Id (lowercase, starts with a letter); build a lookup so returned results map back to the instance and metric they came from.

import os
from datetime import datetime, timedelta, timezone
import aioboto3

METRICS = [
    ("CPUUtilization", "Percent"),
    ("ReadIOPS", "Count/Second"),
    ("WriteIOPS", "Count/Second"),
    ("FreeableMemory", "Bytes"),
]

def build_queries(instances: list[str]) -> tuple[list[dict], dict[str, tuple[str, str]]]:
    """Return (MetricDataQueries, id -> (db_instance, metric)) lookup."""
    queries, lookup = [], {}
    for i, inst in enumerate(instances):
        for j, (metric, _unit) in enumerate(METRICS):
            qid = f"q{i}_{j}"
            lookup[qid] = (inst, metric)
            queries.append({
                "Id": qid,
                "MetricStat": {
                    "Metric": {
                        "Namespace": "AWS/RDS",
                        "MetricName": metric,
                        "Dimensions": [
                            {"Name": "DBInstanceIdentifier", "Value": inst}
                        ],
                    },
                    "Period": 60,          # 1-minute resolution
                    "Stat": "Average",
                },
                "ReturnData": True,
            })
    return queries, lookup

async def poll_once(cw, instances: list[str], account: str) -> list[MetricSignal]:
    """One GetMetricData call over a rolling 5-minute window."""
    queries, lookup = build_queries(instances)
    end = datetime.now(timezone.utc)
    start = end - timedelta(minutes=5)
    unit_by_metric = dict(METRICS)

    signals: list[MetricSignal] = []
    paginator_token = None
    while True:
        kwargs = dict(
            MetricDataQueries=queries,
            StartTime=start,
            EndTime=end,
            ScanBy="TimestampDescending",
        )
        if paginator_token:
            kwargs["NextToken"] = paginator_token
        resp = await cw.get_metric_data(**kwargs)
        for result in resp["MetricDataResults"]:
            inst, metric = lookup[result["Id"]]
            # Take only the newest point per query this tick.
            if result["Timestamps"]:
                signals.append(MetricSignal(
                    db_instance=inst,
                    metric=metric,
                    value=result["Values"][0],
                    unit=unit_by_metric[metric],
                    timestamp=result["Timestamps"][0],
                    account=account,
                ))
        paginator_token = resp.get("NextToken")
        if not paginator_token:
            break
    return signals

Expected shape of the returned list on one tick for two instances:

[MetricSignal(db_instance='prod-orders-1', metric='CPUUtilization', value=63.4, ...),
 MetricSignal(db_instance='prod-orders-1', metric='ReadIOPS', value=812.0, ...),
 MetricSignal(db_instance='prod-orders-1', metric='WriteIOPS', value=140.0, ...),
 MetricSignal(db_instance='prod-orders-1', metric='FreeableMemory', value=2.1e9, ...),
 MetricSignal(db_instance='prod-analytics-2', metric='CPUUtilization', value=12.0, ...)]

Step 3 — Publish to Kafka with keyed, durable sends

The producer is configured with acks="all" so a send is only acknowledged once every in-sync replica has the record — the correct durability for cost data you will bill against. enable_idempotence=True prevents duplicates on retry. The key=db_instance argument drives partition assignment so ordering is preserved per instance.

The sequence below shows one tick: the poller reads a batch from CloudWatch, the producer publishes each keyed record, and Kafka acknowledges only after replication.

RDS CloudWatch to Kafka streaming sequence: batch poll, keyed durable publish with acks=all, and buffer backpressureA Poller requests a metric-data batch from CloudWatch and receives data points. For each point it shapes a keyed cost-signal record and calls send on the aiokafka Producer, which appends to the partition selected by the db_instance key on the Kafka broker. Because acks is all, the broker acknowledges only after replicating to in-sync replicas, and the Producer returns record metadata with partition and offset. When the producer buffer is full the send await blocks, applying backpressure to the poll loop.PollerCloudWatchaiokafkaProducerKafka brokerloop[per data point]get_metric_data( batch )data pointsshape keyed recordsend(key=db_instance, acks=all)append to partitionreplicated ackrecord metadata (partition, offset)
import asyncio
import logging
from aiokafka import AIOKafkaProducer
from aiokafka.errors import KafkaError

TOPIC = "rds-cost-signals"

async def run_stream(instances: list[str], account: str, interval: float = 60.0) -> None:
    """Poll RDS metrics and stream cost signals to Kafka until cancelled."""
    session = aioboto3.Session()
    producer = AIOKafkaProducer(
        bootstrap_servers=os.environ["KAFKA_BOOTSTRAP"],
        acks="all",                     # wait for all in-sync replicas
        enable_idempotence=True,        # no duplicates on retry
        linger_ms=50,                   # small batching window
        max_request_size=1_048_576,
        compression_type="lz4",
    )
    await producer.start()
    try:
        async with session.client("cloudwatch") as cw:
            while True:
                tick = asyncio.get_running_loop().time()
                signals = await poll_once(cw, instances, account)
                for sig in signals:
                    try:
                        # await applies backpressure: if the buffer is full,
                        # this blocks until space frees, throttling the poller.
                        await producer.send_and_wait(
                            TOPIC,
                            key=sig.db_instance.encode("utf-8"),
                            value=sig.to_bytes(),
                        )
                    except KafkaError:
                        logging.exception("publish failed for %s/%s",
                                          sig.db_instance, sig.metric)
                logging.info("published %d signals", len(signals))
                # Sleep the remainder of the interval, accounting for poll time.
                elapsed = asyncio.get_running_loop().time() - tick
                await asyncio.sleep(max(0.0, interval - elapsed))
    finally:
        # Drain in-flight sends before the socket closes.
        await producer.stop()

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(run_stream(["prod-orders-1", "prod-analytics-2"], account="123456789012"))

Expected log output over two ticks:

INFO:root:published 8 signals
INFO:root:published 8 signals

Verification

Confirm records are landing with the right key distribution before wiring an aggregator to the topic.

  1. Consume from the tail with the console consumer, printing keys:

    kafka-console-consumer.sh --bootstrap-server "$KAFKA_BOOTSTRAP" \
  --topic rds-cost-signals --property print.key=true --max-messages 4

  2. Expected record shape on the wire (one value line):

    {"db_instance":"prod-orders-1","metric":"CPUUtilization","value":63.4,"unit":"Percent","timestamp":"2026-07-18T14:03:00Z","account":"123456789012"}

  3. Check partition assignment. All four metrics for prod-orders-1 must share a partition — verify with kafka-console-consumer.sh --partition N that a single instance’s keys never split across partitions, which would break per-instance ordering for stateful consumers.

Gotchas & Edge Cases

  • Metric period floors your resolution. RDS standard monitoring publishes at 60-second granularity; a Period below 60 returns empty arrays for many data points. For sub-minute signals you must enable Enhanced Monitoring (which publishes to CloudWatch Logs, not GetMetricData) — do not shrink the poll interval expecting finer data.
  • GetMetricData caps at 500 queries per call. Four metrics per instance means 125 instances per request. Beyond that, page with NextToken or shard instances across pollers; the code above already follows NextToken.
  • CloudWatch data lags 1–3 minutes. The newest data point is often incomplete or absent. The rolling 5-minute window and TimestampDescending scan take the freshest settled point rather than the current minute — treat CloudWatch as near-real-time, not instantaneous, and align consumers accordingly.
  • acks="all" needs min.insync.replicas set on the broker. Without it, a single-replica partition acknowledges immediately and the durability guarantee is illusory. Confirm the topic’s min.insync.replicas is at least 2 for cost data you bill against.
  • Backpressure vs data loss. send_and_wait blocks the poll loop when the buffer fills, which is correct — it slows ingestion rather than dropping records. If you switch to fire-and-forget send() for throughput, you must still await the returned futures or handle KafkaError in a callback, or a broker hiccup silently loses cost signals. The retry discipline here mirrors implementing retry logic for failed metric pulls.
  • Throttling on the CloudWatch side. High-cardinality estates can hit ThrottlingException on GetMetricData; batch aggressively (500 queries per call) and back off on throttle rather than polling per instance — the same rate-limit strategy as handling rate limits when pulling database metrics.

Frequently Asked Questions

Why key on db_instance instead of letting Kafka round-robin?

Keying pins every record for one instance to a single partition, which guarantees ordering for that instance. Stateful aggregators — a rolling CPU average or an IOPS-to-cost integrator — depend on seeing an instance’s data points in timestamp order. Round-robin spreads one instance across partitions, so a consumer group can process its points out of order and corrupt the running total.

Is send_and_wait too slow for a high-cardinality estate?

For a per-tick batch it is fine because linger_ms still batches records under the hood. If you need maximum throughput, collect the futures from producer.send() (non-blocking) and await asyncio.gather(*futures) once per tick — you keep the acks="all" durability while overlapping network round-trips, but you must handle partial failures in the gathered results.

Should I run one poller per instance or one poller for many?

One poller for many. GetMetricData batches up to 500 queries per call, so a single process covers over 100 instances in one request and one Kafka producer connection. Per-instance pollers multiply both CloudWatch API calls (inviting throttling) and producer connections for no benefit.

How do I avoid double-counting when the poller restarts?

Enable idempotent production (enable_idempotence=True, already set) so retries within a session do not duplicate, and make downstream consumers idempotent on (db_instance, metric, timestamp). Because CloudWatch data points are settled and timestamped, a restart that re-reads the same window republishes identical keys that the consumer deduplicates rather than double-counts.

Can I send Decimal values to preserve precision?

CloudWatch returns metrics as floats, so precision loss is already bounded at the source. Serialize as JSON numbers for these gauge-style signals. Reserve Decimal for monetary fields downstream — the schema validation layer coerces cost into Decimal when it joins these usage signals to per-unit rates.

Back to: Real-Time Metric Streaming Setup