Setup and full context: https://lopes.id/log/log-health-monitoring

derive-seeds.py

#!/usr/bin/env python3
"""Seed + reassessment tool for log_health.source_config / excluded_log_types.

Setup and full context: https://lopes.id/log/log-health-monitoring

Before using, adjust the --project and --dataset flags on `emit` to match your
environment (defaults: chronicle-self / log_health, per the blog's placeholders).

Three subcommands:

  plan    — bootstrap. Reads the estate + silence-gap CSVs from the blog's
            Assessment section and writes an editable plan CSV (one row per
            source). UDM + UNSPECIFIED_LOG_TYPE are filtered (handled
            automatically as auto-excludes by emit).

  emit    — render. Reads the (edited) plan CSV and writes SQL:
              - default mode (bootstrap): INSERT statements for
                excluded_log_types + source_config.
              - --merge mode (reassessment): transaction-wrapped MERGE
                statements that update existing rows, insert new ones, and
                delete rows absent from the plan. enabled flag deliberately
                not touched.

  replan  — reassessment. Reads current source_config + excluded_log_types
            exports from BigQuery plus fresh estate + gaps CSVs, and writes
            a delta plan CSV. Current tracks/thresholds are PRESERVED; the
            `note` column flags PROMOTE/DEMOTE/NEW/GONE/EXCLUDED rows for
            human review.

Stdlib only. Workflow lives in the Assessment and Quarterly Reassessment
sections of the blog post: https://lopes.id/log/log-health-monitoring
"""

import argparse
import csv
import math
import sys

# LT_0..LT_20 are Chronicle backend placeholder log_types, not real sources.
AUTO_EXCLUDE = [
    ("", "Null/empty log_type — data quality bug"),
    ("UDM", "Chronicle Unified Data Model schema, not a source"),
    ("UNSPECIFIED_LOG_TYPE", "Chronicle catch-all; tracked separately"),
    *((f"LT_{i}", "Chronicle backend placeholder (LT_0..LT_20)") for i in range(21)),
]

SILENCE_FLOOR = 6
SILENCE_CEIL = 336
DEFAULT_THRESHOLD = 0.99

PLAN_COLS = [
    "source_key",
    "suggested_track",
    "arima_threshold",
    "max_silence_hours",
    "note",
    "pct_covered",
    "cv",
    "p99_gap_hours",
]


def to_float(val, default=None):
    if val in (None, "", "null", "NULL"):
        return default
    try:
        return float(val)
    except ValueError:
        return default


def to_int(val, default=None):
    f = to_float(val)
    return int(f) if f is not None else default


def silence_hours(p99):
    if p99 is None:
        return SILENCE_FLOOR
    return max(SILENCE_FLOOR, min(SILENCE_CEIL, math.ceil(p99 * 1.5)))


def sql_str(s):
    return "'" + (s or "").replace("'", "''") + "'"


# ----- plan (bootstrap) -----


def cmd_plan(args):
    auto = {lt for lt, _ in AUTO_EXCLUDE}

    with open(args.estate_csv, newline="") as fh:
        estate = list(csv.DictReader(fh))
    with open(args.gaps_csv, newline="") as fh:
        gaps = {r["source_key"]: r for r in csv.DictReader(fh)}

    rows = []
    for r in estate:
        key = (r.get("source_key") or "").strip()
        if not key or key in auto:
            continue
        track = (r.get("suggested_track") or "").strip() or "HEARTBEAT"
        thresh = to_float(r.get("suggested_threshold"), DEFAULT_THRESHOLD)
        p99 = to_float((gaps.get(key) or {}).get("p99_gap"))
        # ARIMA: flat 3h — SPIKE/DROP catches volume anomalies, FULL_STOP is
        # just the tight last-resort tripwire. Override per source in plan.csv
        # for genuinely bursty sources (e.g. HASHCAST at 72h).
        # HEARTBEAT: CEIL(p99_gap * 1.5) clamped to [6, 336].
        hours = 3 if track == "ARIMA" else silence_hours(p99)
        rows.append(
            {
                "source_key": key,
                "suggested_track": track,
                "arima_threshold": f"{thresh:g}",
                "max_silence_hours": hours,
                "note": "",
                "pct_covered": r.get("pct_covered", ""),
                "cv": r.get("cv", ""),
                "p99_gap_hours": "" if p99 is None else int(p99),
            }
        )

    rows.sort(key=lambda r: (r["suggested_track"], r["source_key"]))

    w = csv.DictWriter(sys.stdout, fieldnames=PLAN_COLS)
    w.writeheader()
    w.writerows(rows)

    n_arima = sum(1 for r in rows if r["suggested_track"] == "ARIMA")
    n_hb = sum(1 for r in rows if r["suggested_track"] == "HEARTBEAT")
    sparse = [
        r
        for r in rows
        if r["suggested_track"] == "HEARTBEAT"
        and isinstance(r["max_silence_hours"], int)
        and r["max_silence_hours"] >= 168
    ]
    print(
        f"-- {n_arima} ARIMA + {n_hb} HEARTBEAT = {len(rows)} sources "
        f"(auto-excluded: {', '.join(auto)})",
        file=sys.stderr,
    )
    if sparse:
        print(
            "-- Sparse-source review (max_silence_hours ≥ 168):", file=sys.stderr
        )
        for r in sparse:
            print(f"--   {r['source_key']}: {r['max_silence_hours']}h", file=sys.stderr)


# ----- replan (reassessment) -----


def cmd_replan(args):
    with open(args.config_csv, newline="") as fh:
        current = {r["source_key"]: r for r in csv.DictReader(fh)}
    with open(args.excluded_csv, newline="") as fh:
        excluded_now = {r["log_type"]: r for r in csv.DictReader(fh)}
    with open(args.estate_csv, newline="") as fh:
        estate = list(csv.DictReader(fh))
    with open(args.gaps_csv, newline="") as fh:
        gaps = {r["source_key"]: r for r in csv.DictReader(fh)}

    estate_keys = {(r.get("source_key") or "").strip() for r in estate}
    rows = []
    promotes = demotes = news = gones = 0

    # Carry forward currently-excluded sources.
    for lt, exc in excluded_now.items():
        rows.append(
            {
                "source_key": lt,
                "suggested_track": "EXCLUDE",
                "arima_threshold": f"{DEFAULT_THRESHOLD:g}",
                "max_silence_hours": SILENCE_FLOOR,
                "note": f"EXCLUDED — {exc.get('reason', '')}",
                "pct_covered": "",
                "cv": "",
                "p99_gap_hours": "",
            }
        )

    # Walk fresh estate.
    for r in estate:
        key = (r.get("source_key") or "").strip()
        if not key or key in excluded_now:
            continue
        track_suggested = (r.get("suggested_track") or "").strip()
        thresh_suggested = to_float(r.get("suggested_threshold"), DEFAULT_THRESHOLD)
        p99 = to_float((gaps.get(key) or {}).get("p99_gap"))
        hours_suggested = silence_hours(p99)
        pct = r.get("pct_covered", "")
        cv = r.get("cv", "")

        conf = current.get(key)
        if conf:
            track_current = (conf.get("detection_method") or "HEARTBEAT").strip()
            thresh_current = to_float(
                conf.get("anomaly_prob_threshold"), DEFAULT_THRESHOLD
            )
            hours_current = to_int(conf.get("max_silence_hours"), SILENCE_FLOOR)
            note = ""
            if track_suggested == "ARIMA" and track_current == "HEARTBEAT":
                note = f"PROMOTE candidate (pct={pct}, cv={cv})"
                promotes += 1
            elif track_suggested == "HEARTBEAT" and track_current == "ARIMA":
                note = f"DEMOTE candidate (pct={pct}, cv={cv})"
                demotes += 1
            rows.append(
                {
                    "source_key": key,
                    "suggested_track": track_current,
                    "arima_threshold": f"{thresh_current:g}",
                    "max_silence_hours": hours_current,
                    "note": note,
                    "pct_covered": pct,
                    "cv": cv,
                    "p99_gap_hours": "" if p99 is None else int(p99),
                }
            )
        else:
            # New source: always HEARTBEAT at first contact (see NEW_SOURCE
            # runbook in the blog's Operations section — ARIMA requires 90d of
            # training data that a new source hasn't accumulated yet).
            note = "NEW source — added as HEARTBEAT"
            if track_suggested and track_suggested != "HEARTBEAT":
                note += (
                    f" (fresh suggestion was {track_suggested}; consider at next cycle)"
                )
            news += 1
            rows.append(
                {
                    "source_key": key,
                    "suggested_track": "HEARTBEAT",
                    "arima_threshold": f"{thresh_suggested:g}",
                    "max_silence_hours": hours_suggested,
                    "note": note,
                    "pct_covered": pct,
                    "cv": cv,
                    "p99_gap_hours": "" if p99 is None else int(p99),
                }
            )

    # GONE: in current config but absent from fresh data.
    for key, conf in current.items():
        if key in estate_keys or key in excluded_now:
            continue
        track_current = (conf.get("detection_method") or "HEARTBEAT").strip()
        thresh_current = to_float(conf.get("anomaly_prob_threshold"), DEFAULT_THRESHOLD)
        hours_current = to_int(conf.get("max_silence_hours"), SILENCE_FLOOR)
        gones += 1
        rows.append(
            {
                "source_key": key,
                "suggested_track": track_current,
                "arima_threshold": f"{thresh_current:g}",
                "max_silence_hours": hours_current,
                "note": "GONE — was in config, no data in 90d. Keep, EXCLUDE, or delete row.",
                "pct_covered": "",
                "cv": "",
                "p99_gap_hours": "",
            }
        )

    rows.sort(key=lambda r: (r["suggested_track"], r["source_key"]))

    w = csv.DictWriter(sys.stdout, fieldnames=PLAN_COLS)
    w.writeheader()
    w.writerows(rows)

    print(f"-- replan summary ({len(rows)} rows):", file=sys.stderr)
    print(f"--   {promotes} PROMOTE candidate(s) (HEARTBEAT → ARIMA)", file=sys.stderr)
    print(f"--   {demotes} DEMOTE candidate(s)  (ARIMA → HEARTBEAT)", file=sys.stderr)
    print(f"--   {news} NEW source(s)         (added as HEARTBEAT)", file=sys.stderr)
    print(f"--   {gones} GONE source(s)       (no data in 90d)", file=sys.stderr)
    print(f"--   {len(excluded_now)} carried-forward EXCLUDED", file=sys.stderr)


# ----- emit -----


def _parse_plan(plan):
    excluded, arima, heartbeat = list(AUTO_EXCLUDE), [], []
    for r in plan:
        key = (r.get("source_key") or "").strip()
        if not key:
            continue
        track = (r.get("suggested_track") or "").strip().upper()
        note = (r.get("note") or "").strip()
        # Strip the "EXCLUDED — " prefix from carried-forward exclusions so
        # the reason in the table stays clean.
        if note.startswith("EXCLUDED — "):
            reason = note[len("EXCLUDED — ") :]
        else:
            reason = note or "Excluded via plan"
        if track == "EXCLUDE":
            excluded.append((key, reason))
        elif track == "ARIMA":
            excluded_keys = {k for k, _ in excluded}
            if key not in excluded_keys:
                arima.append(
                    (
                        key,
                        to_float(r.get("arima_threshold"), DEFAULT_THRESHOLD),
                        to_int(r.get("max_silence_hours"), 3),
                    )
                )
        elif track == "HEARTBEAT":
            excluded_keys = {k for k, _ in excluded}
            if key not in excluded_keys:
                heartbeat.append(
                    (key, to_int(r.get("max_silence_hours"), SILENCE_FLOOR))
                )
        else:
            print(
                f"-- WARN unknown track {track!r} for {key}; skipping", file=sys.stderr
            )
    arima.sort()
    heartbeat.sort(key=lambda r: (-r[1], r[0]))
    return excluded, arima, heartbeat


def _emit_insert(fq, excluded, arima, heartbeat):
    print("-- Bootstrap seeds for log_health (excluded_log_types + source_config)")
    print("-- Prerequisites: log_health dataset + tables already created.")
    print("-- Paste-and-run in BQ Studio. Single transaction; all-or-nothing.")
    print()
    print("BEGIN TRANSACTION;")
    print()
    print(f"-- excluded_log_types ({len(excluded)} entries)")
    print(f"INSERT INTO {fq}.excluded_log_types` (log_type, reason) VALUES")
    print(
        ",\n".join(f"  ({sql_str(lt)}, {sql_str(reason)})" for lt, reason in excluded)
        + ";"
    )
    print()
    print(f"-- source_config: ARIMA cohort ({len(arima)} sources)")
    print(f"INSERT INTO {fq}.source_config`")
    print(
        "  (source_key, detection_method, anomaly_prob_threshold, max_silence_hours) VALUES"
    )
    print(
        ",\n".join(f"  ({sql_str(k)}, 'ARIMA', {t:g}, {h})" for k, t, h in arima) + ";"
    )
    print()
    print(f"-- source_config: HEARTBEAT cohort ({len(heartbeat)} sources)")
    print(f"INSERT INTO {fq}.source_config`")
    print(
        "  (source_key, detection_method, anomaly_prob_threshold, max_silence_hours) VALUES"
    )
    print(
        ",\n".join(f"  ({sql_str(k)}, 'HEARTBEAT', NULL, {h})" for k, h in heartbeat)
        + ";"
    )
    print()
    print("COMMIT TRANSACTION;")
    print()
    print("-- Verification (results panel shows row counts after commit)")
    print(
        f"SELECT 'excluded_log_types' AS table_name, COUNT(*) AS n "
        f"FROM {fq}.excluded_log_types`"
    )
    print("UNION ALL")
    print(f"SELECT 'source_config', COUNT(*) FROM {fq}.source_config`;")


def _emit_merge(fq, excluded, arima, heartbeat):
    print("-- Reassessment MERGE (transactional)")
    print("--   UPDATE existing rows, INSERT new ones, DELETE rows absent from plan.")
    print("--   enabled flag on source_config NOT touched (preserved across cycles).")
    print()
    print("BEGIN TRANSACTION;")
    print()

    # excluded_log_types MERGE
    print(f"-- excluded_log_types ({len(excluded)} entries)")
    print(f"MERGE INTO {fq}.excluded_log_types` t")
    print("USING (")
    for i, (lt, reason) in enumerate(excluded):
        if i == 0:
            print(
                f"  SELECT CAST({sql_str(lt)} AS STRING) AS log_type, "
                f"CAST({sql_str(reason)} AS STRING) AS reason"
            )
        else:
            print(f"  UNION ALL SELECT {sql_str(lt)}, {sql_str(reason)}")
    print(") s ON t.log_type = s.log_type")
    print("WHEN MATCHED THEN UPDATE SET reason = s.reason")
    print(
        "WHEN NOT MATCHED THEN INSERT (log_type, reason) VALUES (s.log_type, s.reason)"
    )
    print("WHEN NOT MATCHED BY SOURCE THEN DELETE;")
    print()

    # source_config MERGE — combine ARIMA + HEARTBEAT into one statement
    combined = [(k, "ARIMA", t, h) for k, t, h in arima] + [
        (k, "HEARTBEAT", None, h) for k, h in heartbeat
    ]
    print(f"-- source_config ({len(arima)} ARIMA + {len(heartbeat)} HEARTBEAT)")
    print(f"MERGE INTO {fq}.source_config` t")
    print("USING (")
    for i, (k, method, thresh, hours) in enumerate(combined):
        thresh_sql = f"{thresh:g}" if thresh is not None else "NULL"
        if i == 0:
            print(
                f"  SELECT CAST({sql_str(k)} AS STRING) AS source_key, "
                f"CAST('{method}' AS STRING) AS detection_method, "
                f"CAST({thresh_sql} AS FLOAT64) AS anomaly_prob_threshold, "
                f"CAST({hours} AS INT64) AS max_silence_hours"
            )
        else:
            print(f"  UNION ALL SELECT {sql_str(k)}, '{method}', {thresh_sql}, {hours}")
    print(") s ON t.source_key = s.source_key")
    print("WHEN MATCHED THEN UPDATE SET")
    print("  detection_method       = s.detection_method,")
    print("  anomaly_prob_threshold = s.anomaly_prob_threshold,")
    print("  max_silence_hours      = s.max_silence_hours")
    print("  -- enabled deliberately not touched; controlled via separate UPDATEs.")
    print("WHEN NOT MATCHED THEN INSERT")
    print("  (source_key, detection_method, anomaly_prob_threshold, max_silence_hours)")
    print(
        "  VALUES (s.source_key, s.detection_method, s.anomaly_prob_threshold, s.max_silence_hours)"
    )
    print("WHEN NOT MATCHED BY SOURCE THEN DELETE;")
    print()

    print("COMMIT TRANSACTION;")
    print()


def cmd_emit(args):
    fq = f"`{args.project}.{args.dataset}"
    with open(args.plan_csv, newline="") as fh:
        plan = list(csv.DictReader(fh))
    excluded, arima, heartbeat = _parse_plan(plan)
    if args.merge:
        _emit_merge(fq, excluded, arima, heartbeat)
    else:
        _emit_insert(fq, excluded, arima, heartbeat)
    print(
        f"-- {len(arima)} ARIMA + {len(heartbeat)} HEARTBEAT, {len(excluded)} excluded",
        file=sys.stderr,
    )


# ----- CLI -----


def main():
    ap = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    sub = ap.add_subparsers(dest="cmd", required=True)

    p1 = sub.add_parser("plan", help="bootstrap: estate + gaps CSVs → plan CSV")
    p1.add_argument("estate_csv", help="estate profile query output")
    p1.add_argument("gaps_csv", help="silence-gap query output")
    p1.set_defaults(func=cmd_plan)

    p2 = sub.add_parser("emit", help="render plan CSV → SQL")
    p2.add_argument("plan_csv")
    p2.add_argument(
        "--merge",
        action="store_true",
        help="emit transactional MERGE (reassessment mode) "
        "instead of bare INSERT (bootstrap mode)",
    )
    p2.add_argument("--project", default="chronicle-self")
    p2.add_argument("--dataset", default="log_health")
    p2.set_defaults(func=cmd_emit)

    p3 = sub.add_parser(
        "replan", help="reassessment: current state + fresh discovery → delta plan CSV"
    )
    p3.add_argument("config_csv", help="export of source_config from BigQuery")
    p3.add_argument("excluded_csv", help="export of excluded_log_types from BigQuery")
    p3.add_argument("estate_csv", help="fresh estate profile query output")
    p3.add_argument("gaps_csv", help="fresh silence-gap query output")
    p3.set_defaults(func=cmd_replan)

    args = ap.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()