Ingest from AWS CloudWatch

You will learn how to get AWS CloudWatch logs, alarms, and metrics into Cordo.

CloudWatch has no direct route to Cordo. Everything goes through a small shim — usually a Lambda function. This page explains why, then gives a working shim for each of the three signals.

Before you start

1. Create a stream — Create and manage streams.
2. Mint a stream API key (lax_…) — Manage stream API keys.

Keep the key in AWS Secrets Manager or SSM Parameter Store rather than in the Lambda source.

What CloudWatch can push

CloudWatch emits three unrelated things, and each has its own delivery mechanism:

Signal	How it leaves AWS	Typical target
Logs	A subscription filter on a log group	Lambda, Kinesis Data Firehose, Kinesis Data Streams
Alarms	SNS topic, or a CloudWatch Alarm State Change event on the EventBridge default bus	Email, HTTPS, Lambda, SQS, EventBridge targets
Metrics	Pull only — plus Metric Streams, which forwards to Firehose	GetMetricData, or Firehose delivery

Metrics are the odd one out. They are pull-based, so nothing leaves CloudWatch unless you ask for it, with a single exception: CloudWatch Metric Streams continuously forwards datapoints to Kinesis Data Firehose, which delivers to S3 or a supported partner. That is the only push route for metrics.

Why nothing lands directly in Cordo

Cordo authenticates ingest in exactly two ways:

Ingest path	Auth
POST /v1/streams/{id}/ingest (JSON)	X-API-Key header
POST /v1/streams/{id}/ingest/text (plain text)	X-API-Key header
POST /v1/otlp/v1/logs (OTLP Logs)	X-API-Key header, or Authorization: Bearer
POST /_bulk (OpenSearch-compatible)	HTTP Basic, password = stream API key

An SNS HTTPS subscription sends its own x-amz-sns-* headers and does not let you add an X-API-Key. A Firehose HTTP endpoint destination sends X-Amz-Firehose-Access-Key (or SigV4), also with no arbitrary header. Cordo has no query-string or unauthenticated ingest path, so both get a 401.

Hence the shim. It is not a workaround for a gap in Cordo — it is where you reshape AWS's payloads into the fields Cordo learns from.

Logs only for OTLP

Cordo accepts OTLP Logs only. Do not point a metrics or traces pipeline at the Cordo OTLP endpoint; those requests fail and a collector will retry and fill its logs with errors. See OpenTelemetry ingest.

Logs: subscription filter → Lambda → Cordo

This is the most reliable path, and the only one where you control the shape of each event.

1. Create the Lambda

CloudWatch Logs invokes the function with a gzipped, base64-encoded batch of log events. Set three environment variables before deploying: CORDO_HOST, STREAM_ID, and CORDO_API_KEY.

import base64
import datetime
import gzip
import json
import os
import re
import urllib.request

CORDO_HOST = os.environ["CORDO_HOST"]   # e.g. api.cordo.cloud
STREAM_ID  = os.environ["STREAM_ID"]
API_KEY    = os.environ["CORDO_API_KEY"]  # lax_<prefix>_<secret>

# Plain-text lines carry no level field, so sniff the common conventions.
LEVEL_PROBE = re.compile(r"\b(ERROR|ERR|WARN|WARNING|INFO|DEBUG|TRACE|FATAL)\b")


def iso(timestamp_ms):
    """CloudWatch timestamps are epoch milliseconds; Cordo wants ISO 8601."""
    return datetime.datetime.fromtimestamp(
        timestamp_ms / 1000, datetime.timezone.utc
    ).isoformat()


def sniff_level(message):
    match = LEVEL_PROBE.search(message)
    return match.group(1).lower() if match else "info"


def to_event(log_group, log_stream, entry):
    message = entry.get("message", "")
    event = {
        "@timestamp": iso(entry["timestamp"]),
        "message": message,
        "service": log_group.rsplit("/", 1)[-1],
        "log_group": log_group,
        "log_stream": log_stream,
    }
    # Structured JSON log lines: keep the fields, don't double-wrap them.
    try:
        parsed = json.loads(message)
    except (ValueError, TypeError):
        parsed = None
    if isinstance(parsed, dict):
        event.update(parsed)
        event["message"] = parsed.get("message") or parsed.get("msg") or message
        # Leave level/severity to Cordo's normalisation (ERROR, warn, W, 5, …).
        event.setdefault("level", parsed.get("level") or parsed.get("severity")
                         or sniff_level(message))
    else:
        event["level"] = sniff_level(message)
    return event


def handler(event, context):
    payload = json.loads(gzip.decompress(base64.b64decode(event["awslogs"]["data"])))
    log_group = payload["logGroup"]
    log_stream = payload.get("logStream", "")
    events = [to_event(log_group, log_stream, e) for e in payload["logEvents"]]
    if not events:
        return

    request = urllib.request.Request(
        f"https://{CORDO_HOST}/v1/streams/{STREAM_ID}/ingest",
        data=json.dumps(events).encode("utf-8"),
        headers={"Content-Type": "application/json", "X-API-Key": API_KEY},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        response.read()

The function needs only logs:CreateLogGroup/CreateLogStream/PutLogEvents on its own log group, plus outbound HTTPS. It does not read from CloudWatch.

2. Give CloudWatch permission to invoke it

aws lambda add-permission \
  --function-name cordo-cloudwatch-forwarder \
  --statement-id cordo-cwlogs \
  --principal logs.amazonaws.com \
  --action lambda:InvokeFunction \
  --source-arn "arn:aws:logs:ap-southeast-2:123456789012:log-group:/aws/lambda/my-api:*" \
  --source-account 123456789012

The trailing :* matters — subscription filters pass a versioned log stream name.

3. Attach the subscription filter

aws logs put-subscription-filter \
  --log-group-name /aws/lambda/my-api \
  --filter-name cordo \
  --filter-pattern "" \
  --destination-arn arn:aws:lambda:ap-southeast-2:123456789012:function:cordo-cloudwatch-forwarder

An empty --filter-pattern "" forwards everything. Narrow it with a pattern such as ?ERROR ?WARN ?"Timed out" when a group is too chatty for the budgets below.

Epoch milliseconds are not event time

The iso() conversion is deliberate. Cordo derives producer event time from an ISO 8601 string under @timestamp, timestamp, time, ts, event_time, occurred_at, or the log.* / event.* variants — and it never treats a JSON number as an epoch. If you leave CloudWatch's milliseconds in place, every event falls back to receipt time, so a batch replayed after a Lambda retry looks like it happened now and quietly distorts change and baseline monitors.

Alternative: no Lambda

Vector has an aws_cloudwatch_logs source that manages the subscription filter, Firehose, and S3 plumbing itself and can emit to Cordo's /ingest with an X-API-Key header — see Ingest with Vector and Fluent Bit. It is a service you have to run and operate, but it removes the handler code.

Alarms

An alarm is a state machine (OK → ALARM → INSUFFICIENT_DATA). State changes can leave CloudWatch in two ways.

EventBridge API destination

Since 2020, every alarm state change is published to the default EventBridge bus as a CloudWatch Alarm State Change event, with no SNS in the middle. An API destination can attach arbitrary headers, which is what unlocks direct delivery.

1. In EventBridge, create a connection with an API key authorisation, and store the stream API key as the value.
2. Create an API destination with:
3. Endpoint: https://<your-cordo-host>/v1/streams/<stream-id>/ingest
4. Method: POST
5. Header: X-API-Key: lax_<prefix>_<secret> (from the connection)
6. Create a rule on the default bus:

{
  "source": ["aws.cloudwatch"],
  "detail-type": ["CloudWatch Alarm State Change"]
}

The alarm envelope (detail.alarmName, detail.state.value, detail.state.reason, detail.configuration.metrics) becomes one event per delivery.

EventBridge delivers the envelope as-is, so those events will not cluster into Cordo patterns on their own — the payload has no message, level, or service key. If you want alarm events to read well, use the Lambda below instead, or add an input transformer on the rule to flatten the envelope into message, service, and level.

SNS → Lambda

Use this when you also want the alarm reshaped into clean fields:

import datetime
import json
import os
import re
import urllib.request

# CloudWatch writes StateChangeTime as 2026-10-27T04:00:00.000+0000 — an offset
# without a colon, which is not RFC 3339. Insert the colon.
OFFSET_NO_COLON = re.compile(r"([+-]\d{2})(\d{2})$")


def rfc3339(value):
    """Normalise AWS timestamps; return None when the value is unusable."""
    if not value:
        return None
    candidate = OFFSET_NO_COLON.sub(r"\1:\2", str(value).strip())
    try:
        # fromisoformat's 'Z' handling differs before Python 3.11; Lambda has 3.9+.
        parsed = datetime.datetime.fromisoformat(candidate.replace("Z", "+00:00"))
    except ValueError:
        return None
    return parsed.astimezone(datetime.timezone.utc).isoformat()


def handler(event, context):
    for record in event["Records"]:
        sns = record["Sns"]
        try:
            alarm = json.loads(sns["Message"])
        except (ValueError, TypeError):
            alarm = {"Subject": sns.get("Subject", ""), "Message": sns.get("Message", "")}

        state = alarm.get("NewStateValue", "UNKNOWN")
        cordo_event = {
            "service": "cloudwatch-alarms",
            "level": "error" if state == "ALARM" else "info",
            "message": f"{alarm.get('AlarmName', 'alarm')} → {state}: "
                       f"{alarm.get('NewStateReason', '')}".strip(),
            "alarm_name": alarm.get("AlarmName"),
            "alarm_state": state,
            "previous_state": alarm.get("OldStateValue"),
            "region": alarm.get("Region"),
            "account": alarm.get("AWSAccountId"),
        }
        timestamp = rfc3339(alarm.get("StateChangeTime") or sns.get("Timestamp"))
        if timestamp:
            cordo_event["@timestamp"] = timestamp

        request = urllib.request.Request(
            f"https://{os.environ['CORDO_HOST']}/v1/streams/{os.environ['STREAM_ID']}/ingest",
            data=json.dumps([cordo_event]).encode("utf-8"),
            headers={
                "Content-Type": "application/json",
                "X-API-Key": os.environ["CORDO_API_KEY"],
            },
            method="POST",
        )
        urllib.request.urlopen(request, timeout=10)

Always set @timestamp from StateChangeTime, not from the SNS envelope's own Timestamp: the latter is when SNS delivered the notification, and a retry or a queue backlog would shift it away from when the alarm actually fired.

Cordo is not an alarm receiver

Cordo's notification channels — webhook, email, Slack, PagerDuty — are all outbound. Nothing can POST an alarm into Cordo and have Cordo relay it to PagerDuty. Alarms arrive as events; Cordo's own monitors then decide what is worth paging on. If you want to keep AWS paging you, leave the SNS subscription to PagerDuty in place and send CloudWatch to Cordo as a parallel record.

Metrics: Metric Streams → Firehose → Lambda

Metric Streams is the one CloudWatch signal with no log-shaped payload, so flatten each datapoint into one event.

Use the json output format. The two OpenTelemetry formats (opentelemetry0.7, opentelemetry1.0) are binary protobuf, each record a length-delimited ExportMetricsServiceRequest — and their points are Summary points, carrying count, sum, and quantiles rather than a single value. Either you vendor a protobuf parser, or you use json and get a newline-delimited object per datapoint. This guide uses json.

aws cloudwatch put-metric-stream \
  --name cordo \
  --firehose-arn arn:aws:firehose:ap-southeast-2:123456789012:deliverystream/cordo \
  --role-arn arn:aws:iam::123456789012:role/cordo-metric-stream \
  --output-format json \
  --include-filters '[{"Namespace":"AWS/EC2"}]'

Each Firehose record holds several newline-separated JSON objects, each one datapoint:

{"metric_stream_name":"cordo","account_id":"123456789012","region":"ap-southeast-2","namespace":"AWS/EC2","metric_name":"DiskWriteOps","dimensions":{"InstanceId":"i-123"},"timestamp":1611929698000,"value":{"count":3.0,"sum":20.0,"max":18.0,"min":0.0},"unit":"Seconds"}

So the shim splits on newlines and reads value as an object — not resourceMetrics, and not a scalar.

import base64
import datetime
import gzip
import json
import os
import urllib.request


def iso(epoch_ms):
    """Metric Streams timestamps are epoch milliseconds."""
    return datetime.datetime.fromtimestamp(
        int(epoch_ms) / 1000, datetime.timezone.utc
    ).isoformat()


def to_event(point):
    namespace = point.get("namespace", "")
    name = point.get("metric_name", "")
    if not name:
        return None  # not a metric-stream datapoint
    value = point.get("value") or {}
    # CloudWatch sends SUM, MIN, MAX, and SAMPLECOUNT per interval. Sum/count is
    # the mean over the interval; fall back to sum when count is absent.
    total = value.get("sum")
    count = value.get("count")
    observed = total / count if total is not None and count else total
    if observed is None:
        observed = value.get("max")

    event = {
        "@timestamp": iso(point.get("timestamp", 0)),
        "service": namespace or "cloudwatch",
        "level": "info",
        "message": f"{namespace}/{name}={observed}",
        "metric": f"{namespace}/{name}",
        "value": observed,
        "metric_name": name,
        "unit": point.get("unit", ""),
        "dimensions": point.get("dimensions") or {},
        "account_id": point.get("account_id", ""),
        "region": point.get("region", ""),
        "count": count,
        "sum": total,
        "min": value.get("min"),
        "max": value.get("max"),
    }
    # Drop absent statistics so the field catalog does not learn null samples.
    return {k: v for k, v in event.items() if v is not None}


def handler(event, context):
    out = []
    for record in event["records"]:
        raw = base64.b64decode(record["data"])
        try:
            raw = gzip.decompress(raw)
        except OSError:
            pass
        for line in raw.decode("utf-8", "replace").splitlines():
            line = line.strip()
            if not line:
                continue
            try:
                event = to_event(json.loads(line))
            except (ValueError, TypeError):
                continue  # skip a malformed line rather than fail the batch
            if event:
                out.append(event)

    if out:
        request = urllib.request.Request(
            f"https://{os.environ['CORDO_HOST']}/v1/streams/{os.environ['STREAM_ID']}/ingest",
            data=json.dumps(out).encode("utf-8"),
            headers={
                "Content-Type": "application/json",
                "X-API-Key": os.environ["CORDO_API_KEY"],
            },
            method="POST",
        )
        urllib.request.urlopen(request, timeout=10)

Keep batches under 1 MiB

A Firehose record can exceed the native ingest limit, and this shim posts the whole batch in one request. Set the delivery stream's buffer size to 1 MiB or less so a single record cannot outgrow the request budget.

Metric Streams can emit OTel 0.7 or 1.0, in JSON or Parquet. Pick JSON, and match flatten() to the version you selected — 0.7 nests datapoints under sum/gauge/histogram with the same shape but wraps them in instrumentationLibraryMetrics instead of scopeMetrics. Parquet needs a reader that the Lambda runtime does not ship.

Because Cordo records its own metrics from query aggregates, the useful shape here is one event per datapoint with a stable metric name and a numeric value. You can then record a metric over that query and monitor it like any log-derived signal — see Detect and alert.

Payload and budget limits

Limit	Value
Native JSON / plain-text body	1 MiB per request
OpenSearch _bulk, OTLP Logs body	10 MiB per request
Events per stream	12,000 per minute (default; configurable)
Events per API key	12,000 per minute (default)
Bytes per stream	64 MiB per minute (default)

A subscription filter batch is bounded, so logs are usually fine. Metric Streams is not — a busy account can emit far more datapoints than 12,000 per minute. Trim the metric set in AWS first: the metric stream's inclusion filters (choose namespaces and specific metric names) and its additional statistics setting both cut volume at the source. If you still need more headroom, raise the ingest budgets with the Cordo operator.

Rejected requests are recorded as an Ingest refused reason on the stream, so check that before assuming a Lambda is broken.

Verify

1. Open Events for the stream. Each log line (or alarm, or datapoint) should appear as one event.
2. Open Fields. You should see service, level, message, @timestamp, and any JSON keys your log lines carry.
3. Run a query such as level = error to confirm filtering works.
4. Open Patterns. Log messages should cluster into templates with the service name from log_group. If everything is one giant pattern, the shim is probably sending the whole CloudWatch envelope instead of a message string.

Next

• Google Cloud equivalent: Ingest from Google Cloud
• Method chooser: Ingest your logs
• Field and message conventions: Patterns and findings
• Turn the data into alerts: Detect and alert
• Return to Workflows