478 lines
19 KiB
Python
478 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""seed_journal.py - backdate synthetic alarm activity into the alarm journal.
|
|
|
|
The alarm simulator (SimHarness/alarmsim) generates activity in real time, so
|
|
"give me 8 hours of alarms" would take 8 hours. This writes the same kind of
|
|
activity straight into the journal tables with backdated timestamps, shaped to
|
|
land on a target Alarm Health grade.
|
|
|
|
python3 tools/seed_journal.py --hours 8 --grade B --dry-run # plan only
|
|
python3 tools/seed_journal.py --hours 8 --grade B # insert
|
|
|
|
How the grade is hit (weights from PrimeControls.calc.DEFAULTS):
|
|
|
|
rate 0.30 activations/hr vs the 6/hr ISA target
|
|
flood 0.25 % of window inside a flood episode (>10 activations/10 min)
|
|
chatter 0.20 sources re-activating >10x/hr with a median gap <=120 s
|
|
standing 0.15 alarms active >24 h (LIVE queryStatus - not seeded, see below)
|
|
priority 0.10 deviation from the ISA 80/15/5 low/medium/high mix
|
|
|
|
Only `rate`, `flood` and `priority` are seedable. `standing` reads live gateway
|
|
alarm state via system.alarm.queryStatus, so whatever is actually active now is
|
|
measured as-is and the profile budgets for it (--standing-now).
|
|
|
|
The plan is verified offline against the real PrimeControls.calc before anything
|
|
is written - the same module the gateway runs - and the run aborts if the
|
|
predicted grade misses the target.
|
|
|
|
Journal conventions this mirrors (harvested from existing rows):
|
|
eventtype 0=active 1=clear 2=ack eventflags 0 for tag alarms
|
|
eventtime DATETIME in GATEWAY LOCAL time (America/Chicago), not UTC
|
|
event_data active/clear -> eventValue (dtype 0, intvalue 1/0)
|
|
ack -> ackUser + ackUserName (dtype 2, strvalue)
|
|
|
|
Flood episodes are stable under window drift: calc bins on `origin =
|
|
(start_ms // bin_ms) * bin_ms`, always a multiple of bin_ms, so bin edges sit on
|
|
absolute 10-minute epoch marks no matter when the dashboard re-anchors "now". A
|
|
burst placed inside one such cell stays inside one bin.
|
|
"""
|
|
import argparse
|
|
import collections
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import random
|
|
import subprocess
|
|
import sys
|
|
import types
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
except ImportError:
|
|
ZoneInfo = None
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
PROJECT = os.path.join(ROOT, "ignition", "gateway", "projects", "PrimeBAT",
|
|
"ignition", "script-python", "PrimeControls")
|
|
TAGS_FILE = os.path.join(ROOT, "test-data", "simulation_tags.json")
|
|
|
|
DB_CONTAINER = "buildathon-db"
|
|
DB_NAME = "ignition"
|
|
DB_USER = "ignition"
|
|
DB_PASS = "ignition"
|
|
EVENTS_TABLE = "PrimeControls_alarm_events"
|
|
DATA_TABLE = "PrimeControls_alarm_event_data"
|
|
|
|
GATEWAY_TZ = "America/Chicago"
|
|
TAG_PROVIDER = "default"
|
|
TAG_ROOT = "BuildathonSim"
|
|
|
|
PRIORITY_LEVEL = {"Diagnostic": 0, "Low": 1, "Medium": 2, "High": 3, "Critical": 4}
|
|
BUCKET_OF = {0: "low", 1: "low", 2: "medium", 3: "high", 4: "high"}
|
|
ACK_USERS = [("Prime", 953), ("operator", 284), ("admin", 122)]
|
|
|
|
# Sources whose names advertise a pathology - excluded from a healthy seed.
|
|
BAD_ACTOR_MARKERS = ("_Chatter", "_Standing", "_Fleeting")
|
|
|
|
BIN_MS = 600000 # calc DEFAULTS bin_ms (10 min)
|
|
HOUR_MS = 3600000
|
|
|
|
# Per-window profiles. `counts` is the exact low/medium/high activation split;
|
|
# `burst` is (hours_before_end, activation_count) placed inside one 10-min cell.
|
|
PROFILES = {
|
|
"B": {
|
|
"current": {"counts": {"low": 38, "medium": 9, "high": 3},
|
|
"burst": (2.2, 11), "burst_area": "Packaging"},
|
|
# Prior period only feeds the period-over-period delta chips; its grade
|
|
# is never scored. Made deliberately worse so deltas read as improving.
|
|
# Kept within the low-bucket capacity (13 sources x SOURCE_CAP).
|
|
"prior": {"counts": {"low": 46, "medium": 14, "high": 6},
|
|
"burst": (2.6, 13), "burst_area": "BoilerHouse"},
|
|
},
|
|
}
|
|
|
|
BASELINE_BIN_CAP = 4 # keeps non-burst cells under the flood-start (>10) and
|
|
# flood-continue (>=5) thresholds
|
|
SOURCE_CAP = 4 # < calc chatter_min_count (5) => structurally no chatter
|
|
MIN_DURATION_S = 45 # > calc fleeting_s (10) => structurally no fleeting
|
|
MAX_DURATION_S = 420
|
|
ACK_RATE = 0.85
|
|
CLEAR_RATE = 0.94
|
|
|
|
|
|
# ---------------- the gateway's own calc module ----------------
|
|
|
|
|
|
def load_calc():
|
|
"""Import PrimeControls.calc from the project exactly as tests/conftest.py
|
|
does, so the prediction uses the code the gateway actually runs."""
|
|
pkg = types.ModuleType("PrimeControls")
|
|
pkg.__path__ = [PROJECT]
|
|
sys.modules["PrimeControls"] = pkg
|
|
out = {}
|
|
for mod in ("calc", "fmt"):
|
|
spec = importlib.util.spec_from_file_location(
|
|
"PrimeControls." + mod, os.path.join(PROJECT, mod, "code.py"))
|
|
m = importlib.util.module_from_spec(spec)
|
|
sys.modules["PrimeControls." + mod] = m
|
|
spec.loader.exec_module(m)
|
|
setattr(pkg, mod, m)
|
|
out[mod] = m
|
|
return out["calc"]
|
|
|
|
|
|
# ---------------- sources ----------------
|
|
|
|
|
|
def load_sources():
|
|
"""[(rel_path, priority_level)] for every real (non-bad-actor) sim alarm."""
|
|
with open(TAGS_FILE) as fh:
|
|
doc = json.load(fh)
|
|
rows = []
|
|
|
|
def walk(node, path):
|
|
name = node.get("name")
|
|
p = path + [name] if name else path
|
|
for alarm in node.get("alarms") or []:
|
|
rel = "/".join(p[1:])
|
|
if any(m in rel for m in BAD_ACTOR_MARKERS):
|
|
continue
|
|
rows.append((rel, PRIORITY_LEVEL[alarm.get("priority")]))
|
|
for child in node.get("tags") or []:
|
|
walk(child, p)
|
|
|
|
walk(doc, [])
|
|
return sorted(set(rows))
|
|
|
|
|
|
def source_path(rel):
|
|
name = rel.split("/")[-1]
|
|
return "prov:%s:/tag:%s/%s:/alm:%s" % (TAG_PROVIDER, TAG_ROOT, rel, name)
|
|
|
|
|
|
# ---------------- planning ----------------
|
|
|
|
|
|
def cell_start(ms):
|
|
"""Absolute 10-minute epoch cell containing ms."""
|
|
return (ms // BIN_MS) * BIN_MS
|
|
|
|
|
|
def allocate(sources, counts, burst_count, burst_area, rng):
|
|
"""Pick (rel, level) for every activation, honouring the exact bucket split
|
|
and SOURCE_CAP. Returns (burst_picks, baseline_picks)."""
|
|
by_bucket = collections.defaultdict(list)
|
|
for rel, lvl in sources:
|
|
by_bucket[BUCKET_OF[lvl]].append((rel, lvl))
|
|
|
|
used = collections.Counter()
|
|
|
|
def take(bucket, area=None):
|
|
pool = [s for s in by_bucket[bucket] if used[s[0]] < SOURCE_CAP
|
|
and (area is None or s[0].split("/")[0] == area)]
|
|
if not pool:
|
|
if area is not None:
|
|
return take(bucket, None) # area can't serve it; go global
|
|
raise SystemExit("exhausted %s sources (raise SOURCE_CAP)" % bucket)
|
|
pool.sort(key=lambda s: (used[s[0]], s[0]))
|
|
least = [s for s in pool if used[s[0]] == used[pool[0][0]]]
|
|
pick = rng.choice(least)
|
|
used[pick[0]] += 1
|
|
return pick
|
|
|
|
slots = []
|
|
for bucket, n in counts.items():
|
|
slots += [bucket] * n
|
|
rng.shuffle(slots)
|
|
|
|
# The burst is an area cascade: draw its slots from one area where possible.
|
|
burst_slots, baseline_slots = slots[:burst_count], slots[burst_count:]
|
|
burst = [take(b, burst_area) for b in burst_slots]
|
|
baseline = [take(b) for b in baseline_slots]
|
|
return burst, baseline
|
|
|
|
|
|
def place_times(start_ms, end_ms, burst, baseline, burst_at_ms, rng):
|
|
"""Assign an activation timestamp to every pick.
|
|
|
|
Burst picks land inside the single 10-minute cell containing burst_at_ms
|
|
(with an edge margin so they cannot spill into the neighbouring cell).
|
|
Baseline picks spread over the remaining cells, capped at BASELINE_BIN_CAP.
|
|
"""
|
|
burst_cell = cell_start(burst_at_ms)
|
|
events = []
|
|
|
|
margin = 45000
|
|
for pick in burst:
|
|
t = rng.randint(burst_cell + margin, burst_cell + BIN_MS - margin - 1)
|
|
events.append((t, pick))
|
|
|
|
cells = []
|
|
c = cell_start(start_ms)
|
|
if c < start_ms:
|
|
c += BIN_MS # first whole cell inside the window
|
|
while c + BIN_MS <= end_ms:
|
|
if c != burst_cell:
|
|
cells.append(c)
|
|
c += BIN_MS
|
|
if not cells:
|
|
raise SystemExit("window too short to place baseline activations")
|
|
|
|
capacity = len(cells) * BASELINE_BIN_CAP
|
|
if len(baseline) > capacity:
|
|
raise SystemExit("baseline %d exceeds capacity %d (cap %d/cell over %d cells)"
|
|
% (len(baseline), capacity, BASELINE_BIN_CAP, len(cells)))
|
|
|
|
per_cell = collections.Counter()
|
|
for pick in baseline:
|
|
# Free choice among under-cap cells, so activity clumps the way real
|
|
# activity does. BASELINE_BIN_CAP (4) is below both the flood-start
|
|
# (>10) and flood-continue (>=5) thresholds, so no clump can start or
|
|
# extend a flood episode however the draws fall.
|
|
choices = [c for c in cells if per_cell[c] < BASELINE_BIN_CAP]
|
|
cell = rng.choice(choices)
|
|
per_cell[cell] += 1
|
|
t = rng.randint(cell + 5000, cell + BIN_MS - 5000)
|
|
events.append((t, pick))
|
|
|
|
events.sort(key=lambda e: e[0])
|
|
return events, burst_cell, per_cell
|
|
|
|
|
|
def build_instances(events, end_ms, rng):
|
|
"""Expand each activation into its active / ack / clear lifecycle."""
|
|
instances = []
|
|
for active_ms, (rel, lvl) in events:
|
|
dur_s = rng.randint(MIN_DURATION_S, MAX_DURATION_S)
|
|
clear_ms = active_ms + dur_s * 1000
|
|
# A clear past the window end would not be fetched; leave it open, which
|
|
# is what a still-active alarm looks like anyway.
|
|
if rng.random() > CLEAR_RATE or clear_ms > end_ms - 30000:
|
|
clear_ms = None
|
|
ack_ms = None
|
|
ack_user = None
|
|
if rng.random() < ACK_RATE:
|
|
ceiling = (clear_ms if clear_ms else end_ms) - 5000
|
|
floor = active_ms + 15000
|
|
if ceiling > floor:
|
|
ack_ms = rng.randint(floor, ceiling)
|
|
ack_user = rng.choices([u for u, _ in ACK_USERS],
|
|
weights=[w for _, w in ACK_USERS])[0]
|
|
instances.append({
|
|
"event_id": str(uuid.uuid4()), "rel": rel, "source": source_path(rel),
|
|
"priority": lvl, "active_ms": active_ms, "ack_ms": ack_ms,
|
|
"clear_ms": clear_ms, "ack_user": ack_user,
|
|
})
|
|
return instances
|
|
|
|
|
|
def plan_window(sources, spec, start_ms, end_ms, rng):
|
|
counts = spec["counts"]
|
|
burst_hours, burst_count = spec["burst"]
|
|
burst, baseline = allocate(sources, counts, burst_count,
|
|
spec.get("burst_area"), rng)
|
|
burst_at = end_ms - int(burst_hours * HOUR_MS)
|
|
events, burst_cell, per_cell = place_times(start_ms, end_ms, burst, baseline,
|
|
burst_at, rng)
|
|
instances = build_instances(events, end_ms, rng)
|
|
return instances, burst_cell, per_cell
|
|
|
|
|
|
# ---------------- verification against the real calc ----------------
|
|
|
|
|
|
def to_journal_rows(instances, calc):
|
|
"""Flatten instances into the row dicts alarms._norm_event would produce."""
|
|
rows = []
|
|
for inst in instances:
|
|
lvl, name = calc.normalize_priority(inst["priority"])
|
|
base = {"event_id": inst["event_id"], "source": inst["source"],
|
|
"display_path": "", "priority": lvl, "priority_name": name,
|
|
"is_system": False, "ack_user": None}
|
|
rows.append(dict(base, state="active", ts=inst["active_ms"]))
|
|
if inst["ack_ms"]:
|
|
rows.append(dict(base, state="ack", ts=inst["ack_ms"],
|
|
ack_user=inst["ack_user"]))
|
|
if inst["clear_ms"]:
|
|
rows.append(dict(base, state="clear", ts=inst["clear_ms"]))
|
|
return rows
|
|
|
|
|
|
def predict(calc, rows, active_now, start_ms, end_ms, now_ms):
|
|
bundle = calc.build_bundle(rows, active_now, 0, start_ms, end_ms, now_ms,
|
|
{"filters": {"priorities": [], "areas": [],
|
|
"states": [], "search": ""}})
|
|
return bundle
|
|
|
|
|
|
def synthetic_standing(calc, count, now_ms):
|
|
"""Stand-ins for the live queryStatus rows, so the offline prediction sees
|
|
the same standing-alarm penalty the gateway will apply."""
|
|
rows = []
|
|
for i in range(count):
|
|
lvl, name = calc.normalize_priority(1)
|
|
rows.append({"source": source_path("TankFarm/ManifoldLeakDetect"),
|
|
"display_path": "", "priority": lvl, "priority_name": name,
|
|
"active_ms": now_ms - (48 + i) * HOUR_MS, "unacked": True})
|
|
return rows
|
|
|
|
|
|
# ---------------- SQL ----------------
|
|
|
|
|
|
def local_dt(ms):
|
|
if ZoneInfo is not None:
|
|
return datetime.fromtimestamp(ms / 1000.0, ZoneInfo(GATEWAY_TZ)).strftime(
|
|
"%Y-%m-%d %H:%M:%S")
|
|
return datetime.fromtimestamp(ms / 1000.0).strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def sql_rows(instances, first_id):
|
|
"""(event_values, data_values) as SQL literal tuples."""
|
|
events, data = [], []
|
|
next_id = first_id
|
|
for inst in instances:
|
|
stages = [(0, inst["active_ms"])]
|
|
if inst["ack_ms"]:
|
|
stages.append((2, inst["ack_ms"]))
|
|
if inst["clear_ms"]:
|
|
stages.append((1, inst["clear_ms"]))
|
|
for etype, ts in stages:
|
|
rid = next_id
|
|
next_id += 1
|
|
events.append("(%d,'%s','%s','',%d,%d,0,'%s')" % (
|
|
rid, inst["event_id"], inst["source"], inst["priority"],
|
|
etype, local_dt(ts)))
|
|
if etype == 0:
|
|
data.append("(%d,'eventValue',0,1,NULL,NULL)" % rid)
|
|
elif etype == 1:
|
|
data.append("(%d,'eventValue',0,0,NULL,NULL)" % rid)
|
|
else:
|
|
user = inst["ack_user"]
|
|
data.append("(%d,'ackUser',2,NULL,NULL,'usr-prov:%s:/usr:%s')"
|
|
% (rid, TAG_PROVIDER, user))
|
|
data.append("(%d,'ackUserName',2,NULL,NULL,'%s')" % (rid, user))
|
|
return events, data, next_id
|
|
|
|
|
|
def mariadb(sql, capture=True):
|
|
cmd = ["docker", "exec", "-i", DB_CONTAINER, "mariadb",
|
|
"-u" + DB_USER, "-p" + DB_PASS, DB_NAME, "-N", "-B", "-e", sql]
|
|
res = subprocess.run(cmd, capture_output=capture, text=True)
|
|
if res.returncode != 0:
|
|
raise SystemExit("mariadb failed: %s" % (res.stderr or "").strip())
|
|
return (res.stdout or "").strip()
|
|
|
|
|
|
def max_id():
|
|
out = mariadb("SELECT COALESCE(MAX(id),0) FROM %s;" % EVENTS_TABLE)
|
|
return int(out.split("\n")[0])
|
|
|
|
|
|
# ---------------- reporting ----------------
|
|
|
|
|
|
def show_health(label, bundle):
|
|
h = bundle["health"]
|
|
print(" %s: grade %s score %.2f" % (label, h["grade"], h["score"]))
|
|
for sub in h["subs"]:
|
|
score = "none" if sub["score"] is None else "%6.2f" % sub["score"]
|
|
print(" %-9s w=%.2f score=%s %s"
|
|
% (sub["key"], sub["weight"], score, sub["detail"]))
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("--hours", type=float, default=8.0,
|
|
help="window length; must match the dashboard preset (default 8)")
|
|
ap.add_argument("--grade", default="B", choices=sorted(PROFILES),
|
|
help="target Alarm Health grade (default B)")
|
|
ap.add_argument("--standing-now", type=int, default=1,
|
|
help="live alarms active >24h, which cost 20 pts each on the "
|
|
"standing sub-score. NOT seedable - standing() reads "
|
|
"system.alarm.queryStatus. Confirm the real number with "
|
|
"the SimHarness ProbeTick probe (default 1)")
|
|
ap.add_argument("--no-prior", action="store_true",
|
|
help="skip the preceding comparison window (delta chips go 'new')")
|
|
ap.add_argument("--seed", type=int, default=20260731)
|
|
ap.add_argument("--dry-run", action="store_true",
|
|
help="plan and predict only; write nothing")
|
|
args = ap.parse_args()
|
|
|
|
calc = load_calc()
|
|
sources = load_sources()
|
|
rng = random.Random(args.seed)
|
|
|
|
now_ms = int(datetime.now().timestamp() * 1000)
|
|
span = int(args.hours * HOUR_MS)
|
|
end_ms, start_ms = now_ms, now_ms - span
|
|
prior_start = start_ms - span
|
|
|
|
profile = PROFILES[args.grade]
|
|
|
|
print("window %s -> %s (%.1f h)"
|
|
% (local_dt(start_ms), local_dt(end_ms), args.hours))
|
|
print("sources %d real sim alarms (bad-actor tags excluded)" % len(sources))
|
|
|
|
cur, burst_cell, per_cell = plan_window(sources, profile["current"],
|
|
start_ms, end_ms, rng)
|
|
instances = list(cur)
|
|
if not args.no_prior:
|
|
pri, _, _ = plan_window(sources, profile["prior"], prior_start, start_ms, rng)
|
|
instances = pri + instances
|
|
|
|
standing_n = args.standing_now
|
|
print("standing %d live alarm(s) >24h (measured, not seeded)" % standing_n)
|
|
|
|
rows = to_journal_rows(instances, calc)
|
|
active_now = synthetic_standing(calc, standing_n, now_ms)
|
|
bundle = predict(calc, rows, active_now, start_ms, end_ms, now_ms)
|
|
|
|
m = bundle["meta"]
|
|
print()
|
|
print("predicted (offline, via the gateway's own PrimeControls.calc):")
|
|
print(" activations %d over %.1f h = %.2f/hr"
|
|
% (m["activation_count"], m["window_hours"], bundle["kpis"]["rate_per_hr"]["value"]))
|
|
print(" floods %d episode(s), %.3f%% of window"
|
|
% (len(bundle["floods"]["episodes"]), bundle["floods"]["pct_time_in_flood"]))
|
|
print(" chattering %d fleeting %d standing %d"
|
|
% (len(bundle["chattering"]), bundle["fleeting"]["total"],
|
|
bundle["standing"]["count"]))
|
|
pct = bundle["priority"]["pct"]
|
|
print(" priority %.1f/%.1f/%.1f vs 80/15/5 (dev %.1f)"
|
|
% (pct["low"], pct["medium"], pct["high"], bundle["priority"]["sum_abs_dev"]))
|
|
print(" busiest 10-min cell (non-burst): %d burst cell: %d"
|
|
% (max(per_cell.values()) if per_cell else 0,
|
|
profile["current"]["burst"][1]))
|
|
show_health("health", bundle)
|
|
|
|
grade = bundle["health"]["grade"]
|
|
if grade != args.grade:
|
|
raise SystemExit("\nABORT: predicted grade %s != target %s; nothing written."
|
|
% (grade, args.grade))
|
|
print("\nprediction matches target grade %s" % args.grade)
|
|
|
|
event_rows = sum(1 for _ in rows)
|
|
if args.dry_run:
|
|
print("dry run: would insert %d event rows for %d activations "
|
|
"(%d in the scored window)"
|
|
% (event_rows, len(instances), len(cur)))
|
|
return
|
|
|
|
first = max_id() + 1
|
|
events, data, last = sql_rows(instances, first)
|
|
stmt = ("INSERT INTO %s (id,eventid,source,displaypath,priority,eventtype,"
|
|
"eventflags,eventtime) VALUES %s;\nINSERT INTO %s "
|
|
"(id,propname,dtype,intvalue,floatvalue,strvalue) VALUES %s;"
|
|
% (EVENTS_TABLE, ",".join(events), DATA_TABLE, ",".join(data)))
|
|
mariadb(stmt)
|
|
print("inserted %d event rows + %d data rows (ids %d..%d)"
|
|
% (len(events), len(data), first, last - 1))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|