132 lines
4.4 KiB
Python
132 lines
4.4 KiB
Python
"""Mark VIII reliability event journal. JSONL on /mmc, rotated."""
|
|
import json, os, threading, time
|
|
from collections import deque
|
|
|
|
MK8_DIR = '/mmc/mk8'
|
|
EVENTS_PATH = os.path.join(MK8_DIR, 'events.log')
|
|
MAX_BYTES = 5 * 1024 * 1024
|
|
KEEP = 4
|
|
_LOCK = threading.Lock()
|
|
|
|
_COUNTER_KEYS = {'boot': 'boots', 'unexpected_boot': 'unexpected_boots',
|
|
'rollback': 'rollbacks', 'restart': 'restarts',
|
|
'guard_fix': 'guard_fixes'}
|
|
COUNTER_KINDS = tuple(_COUNTER_KEYS)
|
|
|
|
|
|
def _ensure_dir():
|
|
try:
|
|
os.makedirs(MK8_DIR, exist_ok=True)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def log_event(kind, sev='info', msg='', meta=None):
|
|
"""Append one journal entry. Never raises: a reliability journal that
|
|
can crash its caller would defeat its purpose. Single-writer per
|
|
process is assumed; there is no inter-process lock."""
|
|
try:
|
|
entry = {'ts': int(time.time()), 'kind': str(kind), 'sev': sev,
|
|
'msg': str(msg)[:500]}
|
|
if meta is not None:
|
|
json.dumps(meta)
|
|
entry['meta'] = meta
|
|
line = json.dumps(entry) + '\n'
|
|
except Exception:
|
|
try:
|
|
line = json.dumps({'ts': int(time.time()), 'kind': str(kind),
|
|
'sev': sev, 'msg': str(msg)[:500],
|
|
'meta_repr': repr(meta)[:500]}) + '\n'
|
|
except Exception:
|
|
return
|
|
with _LOCK:
|
|
_ensure_dir()
|
|
try:
|
|
if os.path.exists(EVENTS_PATH) and \
|
|
os.path.getsize(EVENTS_PATH) > MAX_BYTES:
|
|
for i in range(KEEP - 1, 0, -1):
|
|
src = '%s.%d' % (EVENTS_PATH, i)
|
|
dst = '%s.%d' % (EVENTS_PATH, i + 1)
|
|
if os.path.exists(src):
|
|
os.replace(src, dst)
|
|
if os.path.exists(EVENTS_PATH):
|
|
os.replace(EVENTS_PATH, EVENTS_PATH + '.1')
|
|
with open(EVENTS_PATH, 'a') as f:
|
|
f.write(line)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def read_events(limit=100):
|
|
out = []
|
|
paths = [EVENTS_PATH + '.%d' % i for i in range(KEEP, 0, -1)]
|
|
paths.append(EVENTS_PATH)
|
|
for path in paths:
|
|
try:
|
|
with open(path) as f:
|
|
for l in f:
|
|
if not l.strip():
|
|
continue
|
|
try:
|
|
row = json.loads(l)
|
|
except ValueError:
|
|
continue
|
|
if isinstance(row, dict):
|
|
out.append(row)
|
|
except OSError:
|
|
continue
|
|
out.sort(key=lambda r: r.get('ts', 0))
|
|
return out[-limit:][::-1]
|
|
|
|
|
|
def counters():
|
|
counts = {v: 0 for v in _COUNTER_KEYS.values()}
|
|
for row in read_events(limit=5000):
|
|
k = row.get('kind')
|
|
if k in _COUNTER_KEYS:
|
|
counts[_COUNTER_KEYS[k]] += 1
|
|
return counts
|
|
|
|
|
|
def snapshot(event_limit=20, scan=2000):
|
|
"""Newest-first events (up to event_limit) plus kind counters computed
|
|
over at most `scan` most-recent entries, in ONE parse pass. Bounded so a
|
|
large rotated journal cannot spike memory/CPU on every health poll."""
|
|
events = []
|
|
counts = {v: 0 for v in _COUNTER_KEYS.values()}
|
|
scanned = 0
|
|
paths = [EVENTS_PATH]
|
|
paths.extend(EVENTS_PATH + '.%d' % i for i in range(KEEP, 0, -1))
|
|
tail_len = max(scan, event_limit)
|
|
for path in paths:
|
|
if len(events) >= event_limit and scanned >= scan:
|
|
break
|
|
try:
|
|
with open(path) as f:
|
|
tail = deque((l for l in f if l.strip()), maxlen=tail_len)
|
|
except OSError:
|
|
continue
|
|
for line in reversed(tail):
|
|
try:
|
|
row = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if not isinstance(row, dict):
|
|
continue
|
|
scanned += 1
|
|
kind = row.get('kind')
|
|
if kind in _COUNTER_KEYS:
|
|
counts[_COUNTER_KEYS[kind]] += 1
|
|
if len(events) < event_limit:
|
|
events.append(row)
|
|
if len(events) >= event_limit and scanned >= scan:
|
|
break
|
|
events.sort(key=lambda r: r.get('ts', 0), reverse=True)
|
|
return {'events': events[:event_limit], 'reliability': counts}
|
|
|
|
|
|
def mark_boot(unexpected=False):
|
|
log_event('unexpected_boot' if unexpected else 'boot', sev='warn'
|
|
if unexpected else 'info',
|
|
msg='service started' + ('' if unexpected else ' cleanly'))
|