procurement-ingest/lambdas/shared/emf.py

"""Shared CloudWatch EMF emitter for the procurement-ingest pipelines.

A single generic ``emit_metric`` builds the Embedded Metric Format envelope
(``_aws`` block + promoted dimension properties + the metric-value key) and
prints it to stdout, where the Lambda log subscription materializes the metric.
Both the PO and WO email processors emit their parse-outcome metric through
``emit_parse_outcome`` so the load-bearing dimension-set list
``[["ParseMethod"], ["ParseMethod", "TemplateId"]]`` is pinned in exactly ONE
place -- a one-sided dimension change to one pipeline is structurally
impossible.

Envelope byte-equivalence rests on CPython dict insertion order (preserved by
json.dumps with default separators): ``{**properties, metric_name: value}``
yields ``_aws`` first, then the caller's properties in order, then the metric
value last -- matching every inline emitter this module replaced.
"""

import json
from datetime import datetime, timezone

PARSE_METRIC_NAME = "ParseOutcome"

# Two dimension sets are published for every parse-outcome metric: ["ParseMethod"]
# (aggregated across all template ids -- the series the fallback-rate alarms
# query) AND ["ParseMethod", "TemplateId"] (per-template breakdown for Logs
# Insights / dashboards). CloudWatch materializes only the exact dimension sets
# listed here and does NOT auto-aggregate, so an alarm's single-dimension query
# would receive no data unless ["ParseMethod"] is emitted explicitly. Pinned
# ONCE here; both pipelines share it.
_PARSE_DIMENSION_SETS = [["ParseMethod"], ["ParseMethod", "TemplateId"]]


def emit_metric(
    namespace, metric_name, dimension_sets, properties, *, value=1, unit="Count"
):
    """Print one CloudWatch EMF log line for ``metric_name`` in ``namespace``.

    Zero-latency (no PutMetricData API call): the extraction path is async and
    the role already has logs:PutLogEvents. ``properties`` are emitted in caller
    order between the ``_aws`` block and the trailing metric-value key; the keys
    named in ``dimension_sets`` are the promoted (dimensioned) fields, the rest
    ride along as Logs-Insights-queryable properties.
    """
    emf = {
        "_aws": {
            # EMF requires Timestamp (epoch ms); without it CloudWatch may not
            # extract the metric datapoint from the log event.
            "Timestamp": int(datetime.now(timezone.utc).timestamp() * 1000),
            "CloudWatchMetrics": [
                {
                    "Namespace": namespace,
                    "Dimensions": dimension_sets,
                    "Metrics": [{"Name": metric_name, "Unit": unit}],
                }
            ],
        },
        **properties,
        metric_name: value,
    }
    print(json.dumps(emf))


def emit_parse_outcome(namespace, method, template_id, reason_code, id_key, id_value):
    """Emit one parse-outcome EMF line shared by both email processors.

    ParseMethod/TemplateId are the only promoted (dimensioned) fields to keep
    cardinality low; ReasonCode and the pipeline id (``id_key``: ``po_number``
    or ``work_order_id``) ride along as Logs-Insights-queryable properties. See
    ``_PARSE_DIMENSION_SETS`` for why both the single- and two-dimension sets
    are published.
    """
    emit_metric(
        namespace,
        PARSE_METRIC_NAME,
        _PARSE_DIMENSION_SETS,
        {
            "ParseMethod": method,
            "TemplateId": template_id or "unknown",
            "ReasonCode": reason_code or "ok",
            id_key: id_value or "",
        },
    )