a failed Event Streams sink drops events and nothing says so

"""Report Twilio Event Streams sinks that are not delivering events.

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 send messages and
spend money.
"""
import argparse
import datetime as dt
import logging
import os
import sys

import requests

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

EVENTS = "https://events.twilio.com/v1"
MONITOR = "https://monitor.twilio.com/v1"

HEALTHY = "active"
NEVER_RAN = ("initialized", "validating")


def subscribers(subscriptions):
    """Map every sink SID to the subscriptions feeding it. Pure.

    Sinks and subscriptions are separate resources joined only by sink_sid, and
    that join is the whole diagnosis: a failed sink with three subscriptions is
    an outage, and a failed sink with none is litter.
    """
    out = {}
    for sub in subscriptions or []:
        sink = str(sub.get("sink_sid") or "").strip()
        if not sink:
            continue
        out.setdefault(sink, []).append(str(sub.get("sid") or "?"))
    return out


def verdict(sink, subs=None):
    """Classify one sink. Pure, so the difference between a sink that stopped
    working and one that never worked is written down once.

    Returns (state, detail).
    """
    subs = list(subs or [])
    status = str(sink.get("status") or "").lower()
    kind = str(sink.get("sink_type") or "unknown")
    feeding = ("%d subscription(s): %s" % (len(subs), ", ".join(subs)) if subs
               else "no subscription points at it")

    if status == HEALTHY:
        if subs:
            return ("active", "%s sink, delivering, %s." % (kind, feeding))
        return ("unused",
                "%s sink is active but %s, so it delivers nothing. Healthy in the "
                "list and carrying no events." % (kind, feeding))

    if status == "failed":
        if subs:
            return ("failed",
                    "%s sink is failed and %s. Every event those subscriptions "
                    "carry is being dropped, and nothing in the message or call "
                    "logs changed." % (kind, feeding))
        return ("failed-detached",
                "%s sink is failed and %s. Nothing is being lost through it; it "
                "is a dead resource somebody left behind." % (kind, feeding))

    if status in NEVER_RAN:
        return ("unvalidated",
                "%s sink is %s, which means validation was never completed: it "
                "has never delivered a single event. %s."
                % (kind, status, feeding[0].upper() + feeding[1:]))

    return ("unknown-status",
            "%s sink reports status %r, which this check does not recognise. Read "
            "the sink resource by hand." % (kind, status or "empty"))


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 paged(session, url, key, limit, **first):
    params = dict(first)
    params.setdefault("PageSize", 50)
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get(key, []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def alert_dates(session, days, sids):
    """Earliest error alert per sink SID, to date the outage. Alerts are kept
    for 30 days, so an older failure has nothing here and is still a failure."""
    if not sids:
        return {}
    start = (dt.date.today() - dt.timedelta(days=days)).isoformat()
    alerts = paged(session, "%s/Alerts" % MONITOR, "alerts", 10000,
                   LogLevel="error", StartDate=start)
    out = {}
    for a in alerts:
        sid = str(a.get("resource_sid") or "")
        if sid not in sids:
            continue
        when = str(a.get("date_generated") or a.get("date_created") or "")
        if when and (sid not in out or when < out[sid]):
            out[sid] = when
    return out


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=7,
                    help="how far back to sweep alerts to date the outage")
    ap.add_argument("--max-sinks", type=int, default=200,
                    help="stop paging after this many sinks")
    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)

    sinks = paged(session, "%s/Sinks" % EVENTS, "sinks", args.max_sinks)
    if not sinks:
        log.info("no Event Streams sinks on this account")
        return 0

    feeds = subscribers(paged(session, "%s/Subscriptions" % EVENTS,
                              "subscriptions", 500))

    broken = set(str(s.get("sid")) for s in sinks
                 if str(s.get("status") or "").lower() != HEALTHY)
    dated = alert_dates(session, args.days, broken)

    bad = 0
    for sink in sinks:
        sid = str(sink.get("sid"))
        state, detail = verdict(sink, feeds.get(sid))
        line = "%-16s %s (%s)  %s" % (state, sid, sink.get("description", "?"), detail)
        if state == "active":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        if sid in dated:
            log.warning("  first error alert in the window: %s", dated[sid])
        if state == "unused":
            log.warning("  repair: point a subscription at this sink, or delete it "
                        "so it stops looking like observability.")
            continue
        log.warning("  repair: fix the destination or its credentials, validate the "
                    "sink at %s/Sinks/%s/Validate with a TestId, then re-attach it "
                    "at %s/Subscriptions/{SubscriptionSid} with SinkSid=%s. Fixing "
                    "the endpoint alone does not restart delivery.",
                    EVENTS, sid, EVENTS, sid)

    log.info("%d sink(s), %d dropping events", len(sinks), bad)
    return 1 if bad else 0


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

/**
 * Report Twilio Event Streams sinks that are not delivering events.
 *
 * 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 EVENTS = 'https://events.twilio.com/v1';
const MONITOR = 'https://monitor.twilio.com/v1';

const HEALTHY = 'active';
const NEVER_RAN = ['initialized', 'validating'];

/**
 * Map every sink SID to the subscriptions feeding it. Pure.
 *
 * Sinks and subscriptions are separate resources joined only by sink_sid, and
 * that join is the whole diagnosis: a failed sink with three subscriptions is
 * an outage, and a failed sink with none is litter.
 */
export function subscribers(subscriptions) {
  const out = new Map();
  for (const sub of subscriptions ?? []) {
    const sink = String(sub.sink_sid ?? '').trim();
    if (!sink) continue;
    if (!out.has(sink)) out.set(sink, []);
    out.get(sink).push(String(sub.sid ?? '?'));
  }
  return out;
}

/**
 * Classify one sink. Pure, so the difference between a sink that stopped
 * working and one that never worked is written down once.
 * Returns [state, detail].
 */
export function verdict(sink, subs = []) {
  const list = [...(subs ?? [])];
  const status = String(sink.status ?? '').toLowerCase();
  const kind = String(sink.sink_type ?? 'unknown');
  const feeding = list.length
    ? `${list.length} subscription(s): ${list.join(', ')}`
    : 'no subscription points at it';

  if (status === HEALTHY) {
    if (list.length) return ['active', `${kind} sink, delivering, ${feeding}.`];
    return ['unused',
      `${kind} sink is active but ${feeding}, so it delivers nothing. Healthy in ` +
      'the list and carrying no events.'];
  }

  if (status === 'failed') {
    if (list.length) {
      return ['failed',
        `${kind} sink is failed and ${feeding}. Every event those subscriptions ` +
        'carry is being dropped, and nothing in the message or call logs changed.'];
    }
    return ['failed-detached',
      `${kind} sink is failed and ${feeding}. Nothing is being lost through it; it ` +
      'is a dead resource somebody left behind.'];
  }

  if (NEVER_RAN.includes(status)) {
    return ['unvalidated',
      `${kind} sink is ${status}, which means validation was never completed: it ` +
      `has never delivered a single event. ${feeding[0].toUpperCase()}${feeding.slice(1)}.`];
  }

  return ['unknown-status',
    `${kind} sink reports status "${status || 'empty'}", which this check does not ` +
    'recognise. Read the sink resource by hand.'];
}

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 paged(auth, url, key, limit = 200, first = {}) {
  let next = url;
  let params = { PageSize: 50, ...first };
  const out = [];
  while (next && out.length < limit) {
    const page = await get(auth, next, params);
    out.push(...(page[key] ?? []));
    next = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

async function alertDates(auth, days, sids) {
  if (sids.size === 0) return new Map();
  const start = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);
  const alerts = await paged(auth, `${MONITOR}/Alerts`, 'alerts', 10000,
                             { LogLevel: 'error', StartDate: start });
  const out = new Map();
  for (const a of alerts) {
    const sid = String(a.resource_sid ?? '');
    if (!sids.has(sid)) continue;
    const when = String(a.date_generated ?? a.date_created ?? '');
    if (when && (!out.has(sid) || when < out.get(sid))) out.set(sid, when);
  }
  return out;
}

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 = Number(process.argv.includes('--days')
    ? process.argv[process.argv.indexOf('--days') + 1] : 7) || 7;

  const sinks = await paged(auth, `${EVENTS}/Sinks`, 'sinks');
  if (sinks.length === 0) {
    console.log('no Event Streams sinks on this account');
    return;
  }

  const feeds = subscribers(await paged(auth, `${EVENTS}/Subscriptions`,
                                        'subscriptions', 500));

  const broken = new Set(sinks
    .filter((s) => String(s.status ?? '').toLowerCase() !== HEALTHY)
    .map((s) => String(s.sid)));
  const dated = await alertDates(auth, days, broken);

  let bad = 0;
  for (const sink of sinks) {
    const sid = String(sink.sid);
    const [state, detail] = verdict(sink, feeds.get(sid));
    const line = `${state.padEnd(16)} ${sid} (${sink.description ?? '?'})  ${detail}`;
    if (state === 'active') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    if (dated.has(sid)) {
      console.warn(`  first error alert in the window: ${dated.get(sid)}`);
    }
    if (state === 'unused') {
      console.warn('  repair: point a subscription at this sink, or delete it so it ' +
                   'stops looking like observability.');
      continue;
    }
    console.warn('  repair: fix the destination or its credentials, validate the sink ' +
                 `at ${EVENTS}/Sinks/${sid}/Validate with a TestId, then re-attach it ` +
                 `at ${EVENTS}/Subscriptions/{SubscriptionSid} with SinkSid=${sid}. ` +
                 'Fixing the endpoint alone does not restart delivery.');
  }

  console.log(`${sinks.length} sink(s), ${bad} dropping events`);
  process.exitCode = bad ? 1 : 0;
}

// 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_event_sink_audit import subscribers, verdict

SINK = "DG11111111111111111111111111111111"


def sink(status="active", kind="webhook", sid=SINK):
    return {"sid": sid, "status": status, "sink_type": kind,
            "description": "warehouse"}


def subscription(sid="DF1", sink_sid=SINK):
    return {"sid": sid, "sink_sid": sink_sid}


def test_subscribers_joins_on_sink_sid():
    feeds = subscribers([subscription("DF1"), subscription("DF2"),
                         subscription("DF3", "DG99")])
    assert feeds[SINK] == ["DF1", "DF2"]
    assert feeds["DG99"] == ["DF3"]


def test_a_subscription_with_no_sink_is_skipped_not_crashed_on():
    assert subscribers([{"sid": "DF1"}, {"sid": "DF2", "sink_sid": ""}]) == {}
    assert subscribers(None) == {}


def test_a_failed_sink_with_subscriptions_is_the_outage():
    state, detail = verdict(sink("failed"), ["DF1", "DF2"])
    assert state == "failed"
    assert "2 subscription(s)" in detail
    assert "being dropped" in detail


def test_a_failed_sink_nothing_feeds_is_litter_not_an_outage():
    state, detail = verdict(sink("failed"), [])
    assert state == "failed-detached"
    assert "left behind" in detail


def test_initialized_and_validating_never_delivered_anything():
    for status in ("initialized", "validating"):
        state, detail = verdict(sink(status), ["DF1"])
        assert state == "unvalidated"
        assert "never delivered a single event" in detail


def test_an_active_sink_with_no_subscription_delivers_nothing():
    # The failure mode people create while fixing the other one: the sink is
    # green in the list and carries no events at all.
    state, detail = verdict(sink("active"), [])
    assert state == "unused"
    assert "delivers nothing" in detail


def test_an_active_sink_with_a_subscription_is_healthy():
    state, detail = verdict(sink("active"), ["DF1"])
    assert state == "active"
    assert "DF1" in detail


def test_an_unrecognised_status_is_reported_rather_than_assumed_healthy():
    assert verdict(sink("paused"), ["DF1"])[0] == "unknown-status"
    assert verdict(sink(""), ["DF1"])[0] == "unknown-status"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { subscribers, verdict } from './twilio-event-sink-audit.mjs';

const SINK = 'DG11111111111111111111111111111111';

const sink = (status = 'active', kind = 'webhook', sid = SINK) => ({
  sid, status, sink_type: kind, description: 'warehouse',
});

const subscription = (sid = 'DF1', sinkSid = SINK) => ({ sid, sink_sid: sinkSid });

test('subscribers joins on sink_sid', () => {
  const feeds = subscribers([subscription('DF1'), subscription('DF2'),
                             subscription('DF3', 'DG99')]);
  assert.deepEqual(feeds.get(SINK), ['DF1', 'DF2']);
  assert.deepEqual(feeds.get('DG99'), ['DF3']);
});

test('a subscription with no sink is skipped, not crashed on', () => {
  assert.equal(subscribers([{ sid: 'DF1' }, { sid: 'DF2', sink_sid: '' }]).size, 0);
  assert.equal(subscribers(null).size, 0);
});

test('a failed sink with subscriptions is the outage', () => {
  const [state, detail] = verdict(sink('failed'), ['DF1', 'DF2']);
  assert.equal(state, 'failed');
  assert.match(detail, /2 subscription\(s\)/);
  assert.match(detail, /being dropped/);
});

test('a failed sink nothing feeds is litter, not an outage', () => {
  const [state, detail] = verdict(sink('failed'), []);
  assert.equal(state, 'failed-detached');
  assert.match(detail, /left behind/);
});

test('initialized and validating never delivered anything', () => {
  for (const status of ['initialized', 'validating']) {
    const [state, detail] = verdict(sink(status), ['DF1']);
    assert.equal(state, 'unvalidated');
    assert.match(detail, /never delivered a single event/);
  }
});

test('an active sink with no subscription delivers nothing', () => {
  const [state, detail] = verdict(sink('active'), []);
  assert.equal(state, 'unused');
  assert.match(detail, /delivers nothing/);
});

test('an active sink with a subscription is healthy', () => {
  const [state, detail] = verdict(sink('active'), ['DF1']);
  assert.equal(state, 'active');
  assert.match(detail, /DF1/);
});

test('an unrecognised status is reported rather than assumed healthy', () => {
  assert.equal(verdict(sink('paused'), ['DF1'])[0], 'unknown-status');
  assert.equal(verdict(sink(''), ['DF1'])[0], 'unknown-status');
});