Building usage dashboards and budget alerts

The preceding lessons produced structured request logs (Lesson 1), a cost model (Lesson 2), and usage aggregation by user and feature (Lesson 3). This lesson connects those pieces into two concrete operational tools: a dashboard that makes current usage and cost visible at a glance, and an alerting system that proactively notifies someone before spend exceeds an acceptable threshold.

Why Dashboards and Alerts Are Different Tools

A dashboard is a pull-based tool — someone opens it and looks at the current state. It is useful for periodic review, debugging a specific concern, or answering an ad hoc question ("how much did we spend on the summarization feature last week?"). An alert is a push-based tool — it notifies someone automatically when a defined condition is met, without anyone needing to remember to check. It is useful for catching problems that would otherwise go unnoticed until a monthly bill arrives.

Both are necessary because they cover different failure modes. Without a dashboard, understanding why an alert fired requires digging through raw logs. Without alerts, a cost spike goes unnoticed until someone happens to check the dashboard — potentially after the damage (an unexpectedly large bill) is already done.

Designing a Dashboard Data Layer

A dashboard should be built on top of the aggregation logic from Lesson 3, not duplicate it. The dashboard's job is to query and present that data, not to recompute it differently.

from dataclasses import dataclass
from datetime import date, timedelta


@dataclass
class DashboardSnapshot:
    period_start: date
    period_end: date
    total_cost: float
    total_requests: int
    cost_by_feature: dict[str, float]
    top_users_by_cost: list[tuple[str, float]]
    error_rate: float


def build_dashboard_snapshot(
    aggregator: "UsageAggregator",
    error_count: int,
    total_count: int,
    period_start: date,
    period_end: date,
) -> DashboardSnapshot:
    top_users = [(user_id, summary.cost) for user_id, summary in aggregator.top_users_by_cost(n=5)]
    total_cost = sum(summary.cost for summary in aggregator._by_feature.values())
    error_rate = error_count / total_count if total_count > 0 else 0.0

    return DashboardSnapshot(
        period_start=period_start,
        period_end=period_end,
        total_cost=round(total_cost, 4),
        total_requests=total_count,
        cost_by_feature=aggregator.cost_by_feature(),
        top_users_by_cost=top_users,
        error_rate=round(error_rate, 4),
    )

build_dashboard_snapshot reuses UsageAggregator from Lesson 3 directly rather than reimplementing cost summation — this is deliberate: keeping one canonical place where cost is aggregated means the dashboard, the alerts below, and any billing report all agree with each other by construction, instead of risking three slightly different implementations drifting out of sync over time. Bundling period_start and period_end into the returned DashboardSnapshot makes each snapshot self-describing, which matters once you start storing snapshots historically and need to know, without external context, exactly what time window a given number covers.

Rendering a Simple Text Dashboard

A dashboard does not need to be an elaborate web application to be useful. A well-formatted text or console report, generated on demand or on a schedule, is often sufficient for an internal engineering or operations audience.

def render_dashboard_text(snapshot: DashboardSnapshot) -> str:
    lines = [
        f"Usage Report: {snapshot.period_start} to {snapshot.period_end}",
        f"Total cost: ${snapshot.total_cost:.2f}",
        f"Total requests: {snapshot.total_requests}",
        f"Error rate: {snapshot.error_rate:.2%}",
        "",
        "Cost by feature:",
    ]
    for feature, cost in sorted(snapshot.cost_by_feature.items(), key=lambda kv: kv[1], reverse=True):
        lines.append(f"  {feature}: ${cost:.2f}")

    lines.append("")
    lines.append("Top users by cost:")
    for user_id, cost in snapshot.top_users_by_cost:
        lines.append(f"  {user_id}: ${cost:.2f}")

    return "\n".join(lines)

Sorting cost_by_feature by cost descending before rendering (reverse=True) means the most expensive feature always appears first, regardless of insertion order in the underlying dictionary — this is a small detail that matters for readability, since a reader scanning a report wants the biggest cost driver immediately visible, not buried partway down an arbitrarily ordered list. Formatting cost with :.2f and error rate with :.2% produces human-readable output ($142.30, 2.10%) directly, rather than raw floats that would need mental conversion by the reader.

Designing Budget Alert Rules

An alert needs three things: a condition to check, a threshold that defines when the condition is a problem, and an action to take when it fires. Keeping these three concerns separate makes the alerting system easy to extend with new rules later.

from enum import Enum


class AlertSeverity(Enum):
    WARNING = "warning"
    CRITICAL = "critical"


@dataclass
class AlertRule:
    name: str
    check: "Callable[[DashboardSnapshot], bool]"
    severity: AlertSeverity
    message_template: str


def daily_cost_exceeds(threshold: float):
    def check(snapshot: DashboardSnapshot) -> bool:
        return snapshot.total_cost > threshold
    return check


def error_rate_exceeds(threshold: float):
    def check(snapshot: DashboardSnapshot) -> bool:
        return snapshot.error_rate > threshold
    return check


BUDGET_RULES = [
    AlertRule(
        name="daily_cost_warning",
        check=daily_cost_exceeds(100.0),
        severity=AlertSeverity.WARNING,
        message_template="Daily cost ${cost:.2f} exceeded warning threshold of $100",
    ),
    AlertRule(
        name="daily_cost_critical",
        check=daily_cost_exceeds(250.0),
        severity=AlertSeverity.CRITICAL,
        message_template="Daily cost ${cost:.2f} exceeded critical threshold of $250",
    ),
    AlertRule(
        name="error_rate_warning",
        check=error_rate_exceeds(0.05),
        severity=AlertSeverity.WARNING,
        message_template="Error rate {error_rate:.2%} exceeded 5%",
    ),
]

daily_cost_exceeds and error_rate_exceeds are factory functions that return a check closure — this pattern lets the same underlying comparison logic be reused with different thresholds (a warning threshold at $100, a critical threshold at $250) without duplicating the comparison code itself. Defining WARNING and CRITICAL as distinct severities, rather than a single generic "alert," matters operationally: a warning might go to a shared team channel for awareness, while a critical alert might page someone directly, and conflating the two either causes alert fatigue (everything pages) or missed emergencies (nothing pages).

Evaluating Rules and Producing Alerts

With rules defined declaratively, evaluating them against a snapshot is a simple, uniform loop — adding a new rule never requires touching this evaluation logic.

@dataclass
class TriggeredAlert:
    rule_name: str
    severity: AlertSeverity
    message: str


def evaluate_alerts(snapshot: DashboardSnapshot, rules: list[AlertRule]) -> list[TriggeredAlert]:
    triggered = []
    for rule in rules:
        if rule.check(snapshot):
            message = rule.message_template.format(
                cost=snapshot.total_cost,
                error_rate=snapshot.error_rate,
            )
            triggered.append(TriggeredAlert(rule.name, rule.severity, message))
    return triggered


def send_alerts(alerts: list[TriggeredAlert]) -> None:
    for alert in alerts:
        # In production this would call a notification service (email, Slack, PagerDuty).
        print(f"[{alert.severity.value.upper()}] {alert.rule_name}: {alert.message}")

evaluate_alerts iterates over every rule and checks it independently, so multiple alerts can fire from a single snapshot (for example, both the cost warning and the error rate warning at once) — each is evaluated and reported on its own merits rather than the first match short-circuiting the rest. Separating evaluate_alerts (which decides what fired) from send_alerts (which decides how to notify) means the notification mechanism can be swapped or extended — adding a Slack integration, say — without touching the rule evaluation logic at all.

Testing the Dashboard and Alerting Logic

Both the dashboard snapshot construction and the alert evaluation are pure data transformations and should be tested with constructed inputs, never a live usage aggregator connected to real logs.

def test_evaluate_alerts_fires_only_exceeded_rules():
    low_snapshot = DashboardSnapshot(
        period_start=date(2026, 1, 1),
        period_end=date(2026, 1, 1),
        total_cost=50.0,
        total_requests=1000,
        cost_by_feature={},
        top_users_by_cost=[],
        error_rate=0.01,
    )
    high_cost_snapshot = DashboardSnapshot(
        period_start=date(2026, 1, 2),
        period_end=date(2026, 1, 2),
        total_cost=300.0,
        total_requests=1000,
        cost_by_feature={},
        top_users_by_cost=[],
        error_rate=0.01,
    )

    low_alerts = evaluate_alerts(low_snapshot, BUDGET_RULES)
    high_alerts = evaluate_alerts(high_cost_snapshot, BUDGET_RULES)

    assert low_alerts == []
    assert len(high_alerts) == 2  # both cost warning and cost critical fire
    assert any(a.severity == AlertSeverity.CRITICAL for a in high_alerts)
    print("PASS: alerts fire only when thresholds are actually exceeded")


def test_render_dashboard_text_sorts_features_by_cost():
    snapshot = DashboardSnapshot(
        period_start=date(2026, 1, 1),
        period_end=date(2026, 1, 7),
        total_cost=30.0,
        total_requests=500,
        cost_by_feature={"low_cost_feature": 5.0, "high_cost_feature": 25.0},
        top_users_by_cost=[],
        error_rate=0.0,
    )
    text = render_dashboard_text(snapshot)
    high_index = text.index("high_cost_feature")
    low_index = text.index("low_cost_feature")
    assert high_index < low_index
    print("PASS: dashboard lists the more expensive feature first")


test_evaluate_alerts_fires_only_exceeded_rules()
test_render_dashboard_text_sorts_features_by_cost()

The first test constructs two snapshots deliberately positioned on either side of the alert thresholds ($50 versus $300, against thresholds of $100 and $250) to confirm the boundary behavior is correct in both directions — a snapshot below every threshold produces no alerts, and one above both cost thresholds produces exactly two. The second test checks the position of each feature name in the rendered text rather than just checking both names are present, which is what actually verifies the sort order claim rather than just the presence of the data.

Common Mistakes

Building a dashboard that recomputes aggregation independently from the alerting system. Two separate aggregation implementations will eventually disagree, at which point neither number can be fully trusted; both should read from the same underlying aggregation logic, as shown here with UsageAggregator.

Setting a single alert threshold with no severity distinction. Treating every threshold breach as equally urgent either causes alert fatigue if the threshold is too sensitive, or misses genuinely urgent problems if it is calibrated too loosely to reduce noise.

Alerting on cost alone, without an error-rate or latency signal. A cost spike can be caused by a legitimate traffic increase or by a bug causing retry storms or unusually verbose output; an error-rate alert running alongside a cost alert helps distinguish these cases faster.

Best Practices

Reuse the same aggregation logic across dashboards, alerts, and billing. A single source of truth for "how much did this cost" prevents the reports different stakeholders see from silently disagreeing with each other.

Tune alert thresholds using historical data, not guesses. Look at several weeks of the cost_trend data from Lesson 3 to set a threshold that reliably catches genuine anomalies without firing on normal day-to-day variation.

Keep alert rules declarative and centrally defined. A list of AlertRule objects, as shown in BUDGET_RULES, is easier to review, test, and extend than threshold checks scattered as inline conditionals throughout the codebase.