SES bounce rate creeping toward account review

"""Alert on SES bounce and complaint rates before AWS acts on them.

AWS reviews at 5% bounces / 0.1% complaints and can pause sending at 10% / 0.5%.
Alerting at those numbers means alerting when it is already too late, so the
thresholds here default well below and are configurable.
"""
import argparse
import datetime as dt
import logging
import sys

import boto3

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("ses_reputation_watch")

AWS_REVIEW_BOUNCE = 0.05
AWS_REVIEW_COMPLAINT = 0.001


def fetch(cw, metric, days):
    end = dt.datetime.now(dt.timezone.utc)
    points = cw.get_metric_statistics(
        Namespace="AWS/SES",
        MetricName=metric,
        StartTime=end - dt.timedelta(days=days),
        EndTime=end,
        Period=86400,
        Statistics=["Maximum"],
    )["Datapoints"]
    return [p["Maximum"] for p in sorted(points, key=lambda p: p["Timestamp"])]


def judge(series, threshold, label):
    """Pure decision function over a daily series.

    Reports level AND direction, because a flat 3% is a list to clean next sprint
    while a 1% that became 3% this week is an incident happening now.
    """
    if not series:
        return [f"{label}: no data (has the account sent anything?)"]
    latest = series[-1]
    problems = []
    if latest >= threshold:
        problems.append(f"{label}: {latest:.3%} is at or over your {threshold:.3%} threshold")
    if len(series) >= 4:
        earlier = sum(series[:len(series) // 2]) / (len(series) // 2)
        recent = sum(series[len(series) // 2:]) / (len(series) - len(series) // 2)
        if earlier > 0 and recent > earlier * 1.5:
            problems.append(
                f"{label}: rising fast, {earlier:.3%} -> {recent:.3%} across the window")
    return problems


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--region", default="us-east-1")
    ap.add_argument("--days", type=int, default=14)
    ap.add_argument("--max-bounce", type=float, default=0.02)
    ap.add_argument("--max-complaint", type=float, default=0.0005)
    args = ap.parse_args()

    cw = boto3.client("cloudwatch", region_name=args.region)
    problems = []
    for metric, threshold, label in (
        ("Reputation.BounceRate", args.max_bounce, "bounce rate"),
        ("Reputation.ComplaintRate", args.max_complaint, "complaint rate"),
    ):
        series = fetch(cw, metric, args.days)
        if series:
            log.info("%s: latest %.3%%, peak %.3%% over %d days",
                     label, series[-1] * 100, max(series) * 100, args.days)
        problems += judge(series, threshold, label)

    for p in problems:
        log.error(p)
    if problems:
        log.error("AWS reviews at %.0f%% bounces / %.1f%% complaints",
                  AWS_REVIEW_BOUNCE * 100, AWS_REVIEW_COMPLAINT * 100)
    return 1 if problems else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Alert on SES bounce and complaint rates before AWS acts on them.
 *
 * AWS reviews at 5% bounces / 0.1% complaints and can pause sending at 10% / 0.5%.
 * Alerting at those numbers means alerting when it is already too late, so the
 * thresholds here default well below and are configurable.
 */
import { CloudWatchClient, GetMetricStatisticsCommand } from '@aws-sdk/client-cloudwatch';

const AWS_REVIEW_BOUNCE = 0.05;
const AWS_REVIEW_COMPLAINT = 0.001;

async function fetchSeries(cw, MetricName, days) {
  const EndTime = new Date();
  const StartTime = new Date(EndTime.getTime() - days * 86400_000);
  const out = await cw.send(new GetMetricStatisticsCommand({
    Namespace: 'AWS/SES', MetricName, StartTime, EndTime,
    Period: 86400, Statistics: ['Maximum'],
  }));
  return (out.Datapoints ?? [])
    .sort((a, b) => a.Timestamp - b.Timestamp)
    .map((p) => p.Maximum);
}

/**
 * Pure decision function over a daily series.
 *
 * Reports level AND direction, because a flat 3% is a list to clean next sprint
 * while a 1% that became 3% this week is an incident happening now.
 */
export function judge(series, threshold, label) {
  if (!series.length) return [`${label}: no data (has the account sent anything?)`];
  const pct = (n) => `${(n * 100).toFixed(3)}%`;
  const latest = series.at(-1);
  const problems = [];
  if (latest >= threshold) {
    problems.push(`${label}: ${pct(latest)} is at or over your ${pct(threshold)} threshold`);
  }
  if (series.length >= 4) {
    const half = Math.floor(series.length / 2);
    const mean = (a) => a.reduce((x, y) => x + y, 0) / a.length;
    const earlier = mean(series.slice(0, half));
    const recent = mean(series.slice(half));
    if (earlier > 0 && recent > earlier * 1.5) {
      problems.push(`${label}: rising fast, ${pct(earlier)} -> ${pct(recent)} across the window`);
    }
  }
  return problems;
}

async function main() {
  const region = process.env.AWS_REGION ?? 'us-east-1';
  const days = Number(process.env.DAYS ?? 14);
  const maxBounce = Number(process.env.MAX_BOUNCE ?? 0.02);
  const maxComplaint = Number(process.env.MAX_COMPLAINT ?? 0.0005);
  const cw = new CloudWatchClient({ region });

  const problems = [];
  for (const [metric, threshold, label] of [
    ['Reputation.BounceRate', maxBounce, 'bounce rate'],
    ['Reputation.ComplaintRate', maxComplaint, 'complaint rate'],
  ]) {
    problems.push(...judge(await fetchSeries(cw, metric, days), threshold, label));
  }
  problems.forEach((p) => console.error(p));
  if (problems.length) {
    console.error(`AWS reviews at ${AWS_REVIEW_BOUNCE * 100}% bounces / ${AWS_REVIEW_COMPLAINT * 100}% complaints`);
  }
  process.exit(problems.length ? 1 : 0);
}

if (import.meta.url === `file://${process.argv[1]}`) main();

from ses_reputation_watch import judge


def test_empty_series_is_reported_not_ignored():
    assert judge([], 0.02, "bounce rate")


def test_under_threshold_and_flat_is_quiet():
    assert judge([0.01, 0.01, 0.011, 0.01], 0.02, "bounce rate") == []


def test_over_threshold_is_reported():
    problems = judge([0.01, 0.01, 0.01, 0.03], 0.02, "bounce rate")
    assert any("threshold" in p for p in problems)


def test_a_sharp_rise_is_caught_below_the_threshold():
    """0.9% is under a 2% threshold, but tripling in a week is the real signal."""
    problems = judge([0.003, 0.003, 0.009, 0.009], 0.02, "bounce rate")
    assert any("rising fast" in p for p in problems)


def test_short_series_does_not_claim_a_trend():
    assert judge([0.001, 0.001], 0.02, "bounce rate") == []

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { judge } from './ses-reputation-watch.mjs';

test('an empty series is reported, not ignored', () => {
  assert.ok(judge([], 0.02, 'bounce rate').length);
});

test('under threshold and flat is quiet', () => {
  assert.deepEqual(judge([0.01, 0.01, 0.011, 0.01], 0.02, 'bounce rate'), []);
});

test('over threshold is reported', () => {
  const p = judge([0.01, 0.01, 0.01, 0.03], 0.02, 'bounce rate');
  assert.ok(p.some((x) => x.includes('threshold')));
});

test('a sharp rise is caught below the threshold', () => {
  const p = judge([0.003, 0.003, 0.009, 0.009], 0.02, 'bounce rate');
  assert.ok(p.some((x) => x.includes('rising fast')));
});