a trunk sheds calls at its CPS limit and the average hides it

"""Report whether outbound call bursts are hitting a Twilio trunk CPS ceiling.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. The repair is printed, never performed,
because this script holds a credential to an account that can place calls and
spend money.
"""
import argparse
import datetime as dt
import email.utils
import logging
import os
import sys

import requests

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

HOST = "https://api.twilio.com"
BASE = HOST + "/2010-04-01"
MONITOR = "https://monitor.twilio.com/v1"
TRUNKING = "https://trunking.twilio.com/v1"

CPS_EXCEEDED = 32001
CPS_WARNING = 32012


def second_bucket(value):
    """Floor a Twilio timestamp to a whole UTC second, as an ISO string.

    start_time comes back in RFC 2822 form on the 2010-04-01 API. ISO is
    accepted too so the same function can be pointed at other resources. An
    unparseable value returns "" rather than a guess, because a timestamp
    silently bucketed to the epoch would drag the peak somewhere meaningless.
    """
    v = str(value or "").strip()
    if not v:
        return ""
    parsed = None
    if "," in v:
        try:
            parsed = email.utils.parsedate_to_datetime(v)
        except (TypeError, ValueError):
            parsed = None
    if parsed is None:
        try:
            parsed = dt.datetime.fromisoformat(v.replace("Z", "+00:00"))
        except ValueError:
            return ""
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=dt.timezone.utc)
    return parsed.astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def burst_profile(timestamps):
    """Fold call start times into the shape a CPS ceiling is enforced against.

    Returns a dict with the total parsed, the busiest one-second bucket and when
    it was, how many seconds carried any traffic at all, and the span from first
    call to last. Bucketing to the minute instead would divide the peak by sixty
    and produce a reassuring number that answers a different question.
    """
    buckets = {}
    for t in timestamps:
        key = second_bucket(t)
        if not key:
            continue
        buckets[key] = buckets.get(key, 0) + 1
    if not buckets:
        return {"calls": 0, "peak": 0, "at": "", "active_seconds": 0, "span_seconds": 0}
    at = max(sorted(buckets), key=lambda k: buckets[k])
    keys = sorted(buckets)
    first = dt.datetime.strptime(keys[0], "%Y-%m-%dT%H:%M:%SZ")
    last = dt.datetime.strptime(keys[-1], "%Y-%m-%dT%H:%M:%SZ")
    return {"calls": sum(buckets.values()),
            "peak": buckets[at],
            "at": at,
            "active_seconds": len(buckets),
            "span_seconds": int((last - first).total_seconds()) + 1}


def verdict(profile, ceiling, alerts=0, warnings=0, burst_ratio=4):
    """Judge a burst profile against a CPS ceiling. Pure, so it tests offline.

    ceiling is the trunk's calls-per-second allowance. No read API reports it,
    so it is supplied by whoever runs this rather than discovered.

    Returns (state, detail).
    """
    calls = profile.get("calls", 0)
    if not calls:
        return ("no-calls", "no calls with a readable start_time in this window.")

    peak = profile.get("peak", 0)
    span = max(profile.get("span_seconds", 0), 1)
    mean = calls / float(span)

    if alerts:
        return ("shedding",
                "%d call(s) rejected with %d: the peak was %d call(s) in the "
                "second at %s against a ceiling of %d, while the mean over the "
                "window was %.2f per second and hid all of it."
                % (alerts, CPS_EXCEEDED, peak, profile.get("at"), ceiling, mean))

    if peak > ceiling:
        return ("over-ceiling",
                "peak of %d call(s) at %s is above the ceiling of %d with no "
                "%d alert in the window, so either the ceiling is higher than "
                "the value given here or the calls were spread across trunks."
                % (peak, profile.get("at"), ceiling, CPS_EXCEEDED))

    if peak == ceiling:
        return ("at-ceiling",
                "peak of %d call(s) at %s sits exactly on the ceiling: nothing "
                "was lost this time and a batch one call larger will be."
                % (peak, profile.get("at")))

    if warnings:
        return ("warned",
                "%d %d warning(s) at LogLevel=warning with a peak of %d against "
                "a ceiling of %d. That is the notice that comes before the "
                "shedding, and it is the one an error-only sweep drops."
                % (warnings, CPS_WARNING, peak, ceiling))

    if peak >= burst_ratio * mean and peak >= 2:
        return ("bursty",
                "peak of %d call(s) at %s against a mean of %.2f per second: "
                "under the ceiling of %d today, but the traffic arrives in "
                "bursts and no hourly average will ever show it."
                % (peak, profile.get("at"), mean, ceiling))

    return ("within-ceiling",
            "peak of %d call(s) in one second against a ceiling of %d, mean "
            "%.2f per second over %d second(s)."
            % (peak, ceiling, mean, span))


def get(session, url, **params):
    r = session.get(url, params=params, timeout=30)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Twilio: check TWILIO_ACCOUNT_SID and that the "
                         "API key belongs to that account with read access"
                         % r.status_code)
    r.raise_for_status()
    return r.json()


def list_calls(session, account, since, limit):
    """Page the calls. next_page_uri here is a path, and there is no ErrorCode
    filter on this resource, so everything is bucketed client-side."""
    url = "%s/Accounts/%s/Calls.json" % (BASE, account)
    params = {"StartTime>=": since, "PageSize": 1000}
    out = []
    while url and len(out) < limit:
        body = get(session, url, **params)
        out.extend(body.get("calls", []))
        nxt = body.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


def sweep_alerts(session, since, limit, levels):
    """Both log levels, merged on sid.

    32001 is an error and 32012 is a warning. A sweep hard-coded to the error
    level sees the calls you lost and never the warning that preceded them.
    """
    seen = {}
    for level in levels:
        url = MONITOR + "/Alerts"
        params = {"LogLevel": level, "StartDate": since, "PageSize": 1000}
        got = 0
        while url and got < limit:
            page = get(session, url, **params)
            for a in page.get("alerts", []):
                seen.setdefault(a.get("sid"), a)
                got += 1
            url = (page.get("meta") or {}).get("next_page_url")
            params = {}
    return list(seen.values())


def count_trunks(session):
    """How many trunks the traffic could be spread across. One paginated read."""
    url = TRUNKING + "/Trunks"
    params = {"PageSize": 100}
    total = 0
    while url:
        page = get(session, url, **params)
        total += len(page.get("trunks", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return total


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=1,
                    help="window to measure; keep it short, this reads every call")
    ap.add_argument("--cps", type=int, default=1,
                    help="the trunk's calls-per-second ceiling, which no read API "
                         "reports: use the value Twilio gave you")
    ap.add_argument("--max-calls", type=int, default=20000,
                    help="stop after this many calls")
    ap.add_argument("--errors-only", action="store_true",
                    help="skip the warning level, which drops 32012 entirely")
    args = ap.parse_args()

    account = os.environ.get("TWILIO_ACCOUNT_SID")
    key = os.environ.get("TWILIO_API_KEY")
    secret = os.environ.get("TWILIO_API_SECRET")
    if not (account and key and secret):
        log.error("set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET "
                  "(an API Key with read access, not the auth token)")
        return 2

    session = requests.Session()
    session.auth = (key, secret)

    days = min(args.days, 30)
    since = (dt.date.today() - dt.timedelta(days=days)).isoformat()
    levels = ["error"] if args.errors_only else ["error", "warning"]

    alerts = sweep_alerts(session, since, 10000, levels)
    exceeded = [a for a in alerts
                if str(a.get("error_code") or "").strip() == str(CPS_EXCEEDED)]
    warned = [a for a in alerts
              if str(a.get("error_code") or "").strip() == str(CPS_WARNING)]

    calls = list_calls(session, account, since, args.max_calls)
    profile = burst_profile(c.get("start_time") for c in calls)
    state, detail = verdict(profile, args.cps, len(exceeded), len(warned))

    log.info("%d call(s) over %d day(s) across %d trunk(s)",
             len(calls), days, count_trunks(session))
    if state in ("within-ceiling", "no-calls"):
        log.info("%-15s %s", state, detail)
        return 0

    log.warning("%-15s %s", state, detail)
    log.warning("  repair: rate-limit the dialer below %d call(s) per second, "
                "spread the campaign across additional trunks, or ask Twilio "
                "Support to raise the trunk's CPS", args.cps)
    log.warning("  measure again over a window containing a real campaign: a "
                "peak taken on a quiet afternoon confirms nothing")
    return 1


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

/**
 * Report whether outbound call bursts are hitting a Twilio trunk CPS ceiling.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The repair is printed, never
 * performed.
 */
const HOST = 'https://api.twilio.com';
const BASE = `${HOST}/2010-04-01`;
const MONITOR = 'https://monitor.twilio.com/v1';
const TRUNKING = 'https://trunking.twilio.com/v1';

const CPS_EXCEEDED = 32001;
const CPS_WARNING = 32012;

/**
 * Floor a Twilio timestamp to a whole UTC second, as an ISO string. start_time
 * comes back in RFC 2822 form on the 2010-04-01 API; ISO is accepted too. An
 * unparseable value returns '' rather than a guess, because a timestamp
 * silently bucketed to the epoch would drag the peak somewhere meaningless.
 */
export function secondBucket(value) {
  const v = String(value ?? '').trim();
  if (!v) return '';
  const ms = Date.parse(v);
  if (Number.isNaN(ms)) return '';
  return new Date(Math.floor(ms / 1000) * 1000).toISOString().replace('.000Z', 'Z');
}

/**
 * Fold call start times into the shape a CPS ceiling is enforced against.
 * Bucketing to the minute instead would divide the peak by sixty and produce a
 * reassuring number that answers a different question.
 */
export function burstProfile(timestamps) {
  const buckets = new Map();
  for (const t of timestamps) {
    const key = secondBucket(t);
    if (!key) continue;
    buckets.set(key, (buckets.get(key) ?? 0) + 1);
  }
  if (buckets.size === 0) {
    return { calls: 0, peak: 0, at: '', active_seconds: 0, span_seconds: 0 };
  }
  const keys = [...buckets.keys()].sort();
  let at = keys[0];
  for (const k of keys) if (buckets.get(k) > buckets.get(at)) at = k;
  const first = Date.parse(keys[0]);
  const last = Date.parse(keys[keys.length - 1]);
  let calls = 0;
  for (const n of buckets.values()) calls += n;
  return {
    calls,
    peak: buckets.get(at),
    at,
    active_seconds: buckets.size,
    span_seconds: Math.round((last - first) / 1000) + 1,
  };
}

/**
 * Judge a burst profile against a CPS ceiling. `ceiling` is supplied rather than
 * discovered: no read API reports a trunk's calls-per-second allowance. Pure.
 * Returns [state, detail].
 */
export function verdict(profile, ceiling, alerts = 0, warnings = 0, burstRatio = 4) {
  const calls = profile.calls ?? 0;
  if (!calls) return ['no-calls', 'no calls with a readable start_time in this window.'];

  const peak = profile.peak ?? 0;
  const span = Math.max(profile.span_seconds ?? 0, 1);
  const mean = calls / span;

  if (alerts) {
    return ['shedding',
      `${alerts} call(s) rejected with ${CPS_EXCEEDED}: the peak was ${peak} ` +
      `call(s) in the second at ${profile.at} against a ceiling of ${ceiling}, ` +
      `while the mean over the window was ${mean.toFixed(2)} per second and hid ` +
      'all of it.'];
  }

  if (peak > ceiling) {
    return ['over-ceiling',
      `peak of ${peak} call(s) at ${profile.at} is above the ceiling of ` +
      `${ceiling} with no ${CPS_EXCEEDED} alert in the window, so either the ` +
      'ceiling is higher than the value given here or the calls were spread ' +
      'across trunks.'];
  }

  if (peak === ceiling) {
    return ['at-ceiling',
      `peak of ${peak} call(s) at ${profile.at} sits exactly on the ceiling: ` +
      'nothing was lost this time and a batch one call larger will be.'];
  }

  if (warnings) {
    return ['warned',
      `${warnings} ${CPS_WARNING} warning(s) at LogLevel=warning with a peak of ` +
      `${peak} against a ceiling of ${ceiling}. That is the notice that comes ` +
      'before the shedding, and it is the one an error-only sweep drops.'];
  }

  if (peak >= burstRatio * mean && peak >= 2) {
    return ['bursty',
      `peak of ${peak} call(s) at ${profile.at} against a mean of ` +
      `${mean.toFixed(2)} per second: under the ceiling of ${ceiling} today, but ` +
      'the traffic arrives in bursts and no hourly average will ever show it.'];
  }

  return ['within-ceiling',
    `peak of ${peak} call(s) in one second against a ceiling of ${ceiling}, mean ` +
    `${mean.toFixed(2)} per second over ${span} second(s).`];
}

function authHeader(key, secret) {
  return `Basic ${Buffer.from(`${key}:${secret}`).toString('base64')}`;
}

async function get(auth, url, params = {}) {
  const u = new URL(url);
  for (const [k, v] of Object.entries(params)) u.searchParams.set(k, v);
  const res = await fetch(u, { headers: { Authorization: auth } });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Twilio: check TWILIO_ACCOUNT_SID and ` +
                    'that the API key belongs to that account with read access');
  }
  if (!res.ok) throw new Error(`${res.status} from ${u.pathname}`);
  return res.json();
}

export async function listCalls(auth, account, since, limit) {
  let url = `${BASE}/Accounts/${account}/Calls.json`;
  let params = { 'StartTime>=': since, PageSize: 1000 };
  const out = [];
  while (url && out.length < limit) {
    const body = await get(auth, url, params);
    out.push(...(body.calls ?? []));
    url = body.next_page_uri ? HOST + body.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

/** Both log levels, merged on sid: 32001 is an error and 32012 is a warning. */
export async function sweepAlerts(auth, since, limit, levels) {
  const seen = new Map();
  for (const level of levels) {
    let url = `${MONITOR}/Alerts`;
    let params = { LogLevel: level, StartDate: since, PageSize: 1000 };
    let got = 0;
    while (url && got < limit) {
      const page = await get(auth, url, params);
      for (const a of page.alerts ?? []) {
        if (!seen.has(a.sid)) seen.set(a.sid, a);
        got += 1;
      }
      url = page.meta?.next_page_url ?? null;
      params = {};
    }
  }
  return [...seen.values()];
}

async function countTrunks(auth) {
  let url = `${TRUNKING}/Trunks`;
  let params = { PageSize: 100 };
  let total = 0;
  while (url) {
    const page = await get(auth, url, params);
    total += (page.trunks ?? []).length;
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return total;
}

function flagValue(name, fallback) {
  const i = process.argv.indexOf(name);
  return i === -1 ? fallback : Number(process.argv[i + 1]);
}

async function main() {
  const account = process.env.TWILIO_ACCOUNT_SID;
  const key = process.env.TWILIO_API_KEY;
  const secret = process.env.TWILIO_API_SECRET;
  if (!account || !key || !secret) {
    console.error('set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET ' +
                  '(an API Key with read access, not the auth token)');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);
  const days = Math.min(flagValue('--days', 1), 30);
  const ceiling = flagValue('--cps', 1);
  const maxCalls = flagValue('--max-calls', 20000);
  const levels = process.argv.includes('--errors-only') ? ['error'] : ['error', 'warning'];
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);

  const alerts = await sweepAlerts(auth, since, 10000, levels);
  const code = (a) => String(a.error_code ?? '').trim();
  const exceeded = alerts.filter((a) => code(a) === String(CPS_EXCEEDED));
  const warned = alerts.filter((a) => code(a) === String(CPS_WARNING));

  const calls = await listCalls(auth, account, since, maxCalls);
  const profile = burstProfile(calls.map((c) => c.start_time));
  const [state, detail] = verdict(profile, ceiling, exceeded.length, warned.length);

  console.log(`${calls.length} call(s) over ${days} day(s) across ` +
              `${await countTrunks(auth)} trunk(s)`);
  if (state === 'within-ceiling' || state === 'no-calls') {
    console.log(`${state.padEnd(15)} ${detail}`);
    return;
  }

  console.warn(`${state.padEnd(15)} ${detail}`);
  console.warn(`  repair: rate-limit the dialer below ${ceiling} call(s) per ` +
               'second, spread the campaign across additional trunks, or ask ' +
               "Twilio Support to raise the trunk's CPS");
  console.warn('  measure again over a window containing a real campaign: a peak ' +
               'taken on a quiet afternoon confirms nothing');
  process.exitCode = 1;
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing credentials and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from twilio_trunk_cps_audit import burst_profile, second_bucket, verdict

# Six starts inside one second, one in the next. The peak is six.
BURST = ["Tue, 31 Aug 2010 20:36:28 +0000"] * 6 + ["Tue, 31 Aug 2010 20:36:29 +0000"]


def test_rfc_2822_start_time_is_floored_to_the_second():
    assert second_bucket("Tue, 31 Aug 2010 20:36:28 +0000") == "2010-08-31T20:36:28Z"


def test_iso_timestamps_and_offsets_normalise_to_utc():
    assert second_bucket("2010-08-31T21:36:28+01:00") == "2010-08-31T20:36:28Z"
    assert second_bucket("2010-08-31T20:36:28Z") == "2010-08-31T20:36:28Z"


def test_an_unparseable_timestamp_is_dropped_rather_than_guessed():
    # Bucketed to the epoch it would stretch the span and flatten the peak.
    assert second_bucket("last tuesday") == ""
    assert second_bucket(None) == ""


def test_the_peak_is_the_busiest_single_second():
    p = burst_profile(BURST)
    assert p["calls"] == 7
    assert p["peak"] == 6
    assert p["at"] == "2010-08-31T20:36:28Z"
    assert p["active_seconds"] == 2
    assert p["span_seconds"] == 2


def test_an_empty_window_has_no_peak_and_no_span():
    p = burst_profile([])
    assert p == {"calls": 0, "peak": 0, "at": "", "active_seconds": 0,
                 "span_seconds": 0}
    assert verdict(p, 10)[0] == "no-calls"


def test_alerts_outrank_everything_and_quote_the_hiding_mean():
    state, detail = verdict(burst_profile(BURST), 5, alerts=44)
    assert state == "shedding"
    assert "44 call(s) rejected" in detail
    assert "3.50 per second" in detail


def test_a_peak_on_the_ceiling_is_its_own_state():
    state, detail = verdict(burst_profile(BURST), 6)
    assert state == "at-ceiling"
    assert "one call larger" in detail


def test_a_peak_above_the_ceiling_with_no_alert_says_so():
    state, detail = verdict(burst_profile(BURST), 4)
    assert state == "over-ceiling"
    assert "spread across trunks" in detail


def test_the_warning_level_code_is_reported_before_anything_is_lost():
    state, detail = verdict(burst_profile(BURST), 20, warnings=3)
    assert state == "warned"
    assert "error-only sweep" in detail


def test_a_burst_well_under_the_ceiling_is_still_the_finding():
    # 6 in one second against a mean of 3.5 is not four times the mean, so
    # stretch the window: the same six calls over a quieter minute are.
    quiet = BURST + ["Tue, 31 Aug 2010 20:37:%02d +0000" % s for s in range(30, 50)]
    state, detail = verdict(burst_profile(quiet), 50)
    assert state == "bursty"
    assert "no hourly average" in detail


def test_a_flat_stream_under_the_ceiling_is_clean():
    flat = ["Tue, 31 Aug 2010 20:36:%02d +0000" % s for s in range(10, 40)]
    state, _ = verdict(burst_profile(flat), 5)
    assert state == "within-ceiling"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { burstProfile, secondBucket, verdict } from './twilio-trunk-cps-audit.mjs';

// Six starts inside one second, one in the next. The peak is six.
const BURST = [
  ...Array(6).fill('Tue, 31 Aug 2010 20:36:28 +0000'),
  'Tue, 31 Aug 2010 20:36:29 +0000',
];

const pad = (n) => String(n).padStart(2, '0');

test('rfc 2822 start_time is floored to the second', () => {
  assert.equal(secondBucket('Tue, 31 Aug 2010 20:36:28 +0000'), '2010-08-31T20:36:28Z');
});

test('iso timestamps and offsets normalise to utc', () => {
  assert.equal(secondBucket('2010-08-31T21:36:28+01:00'), '2010-08-31T20:36:28Z');
  assert.equal(secondBucket('2010-08-31T20:36:28Z'), '2010-08-31T20:36:28Z');
});

test('an unparseable timestamp is dropped rather than guessed', () => {
  assert.equal(secondBucket('last tuesday'), '');
  assert.equal(secondBucket(null), '');
});

test('the peak is the busiest single second', () => {
  const p = burstProfile(BURST);
  assert.equal(p.calls, 7);
  assert.equal(p.peak, 6);
  assert.equal(p.at, '2010-08-31T20:36:28Z');
  assert.equal(p.active_seconds, 2);
  assert.equal(p.span_seconds, 2);
});

test('an empty window has no peak and no span', () => {
  const p = burstProfile([]);
  assert.deepEqual(p, { calls: 0, peak: 0, at: '', active_seconds: 0, span_seconds: 0 });
  assert.equal(verdict(p, 10)[0], 'no-calls');
});

test('alerts outrank everything and quote the hiding mean', () => {
  const [state, detail] = verdict(burstProfile(BURST), 5, 44);
  assert.equal(state, 'shedding');
  assert.match(detail, /44 call\(s\) rejected/);
  assert.match(detail, /3.50 per second/);
});

test('a peak on the ceiling is its own state', () => {
  const [state, detail] = verdict(burstProfile(BURST), 6);
  assert.equal(state, 'at-ceiling');
  assert.match(detail, /one call larger/);
});

test('a peak above the ceiling with no alert says so', () => {
  const [state, detail] = verdict(burstProfile(BURST), 4);
  assert.equal(state, 'over-ceiling');
  assert.match(detail, /spread across trunks/);
});

test('the warning level code is reported before anything is lost', () => {
  const [state, detail] = verdict(burstProfile(BURST), 20, 0, 3);
  assert.equal(state, 'warned');
  assert.match(detail, /error-only sweep/);
});

test('a burst well under the ceiling is still the finding', () => {
  const quiet = [...BURST];
  for (let s = 30; s < 50; s += 1) quiet.push(`Tue, 31 Aug 2010 20:37:${pad(s)} +0000`);
  const [state, detail] = verdict(burstProfile(quiet), 50);
  assert.equal(state, 'bursty');
  assert.match(detail, /no hourly average/);
});

test('a flat stream under the ceiling is clean', () => {
  const flat = [];
  for (let s = 10; s < 40; s += 1) flat.push(`Tue, 31 Aug 2010 20:36:${pad(s)} +0000`);
  assert.equal(verdict(burstProfile(flat), 5)[0], 'within-ceiling');
});