jaredandClaude Opus 5.5
c262e5ecc8
Lint / JS (eslint) (push) Successful in 7s
Lint / Python (flake8) (push) Successful in 54s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 1m1s
Lint / JS (eslint) (pull_request) Successful in 19s
Test / Python Tests (pytest) (push) Successful in 2m6s
Security / Python Security (bandit) (pull_request) Successful in 1m18s
Lint / Python (flake8) (pull_request) Successful in 53s
Test / Python Tests (pytest) (pull_request) Successful in 1m1s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Device pages get an Inventory button: it opens the device's HomeBox record
once an admin pastes its URL (stored in the new device_links table; only
URLs on inventory.base_url are accepted), and searches the inventory for
the model until then. Tickets link the record too.
Faceplate ports with a documented cable get a corner dot: green matches,
red mismatch, grey documented but unconfirmed. The port's tooltip names
the cable, and a legend toggle (remembered per browser) hides them.
Page titles no longer repeat GANDALF ("Alert History - GANDALF - GANDALF").
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1000 lines
40 KiB
Python
1000 lines
40 KiB
Python
"""Database operations for Gandalf network monitor."""
|
|
import json
|
|
import logging
|
|
import threading
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
import pymysql
|
|
import pymysql.cursors
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_config_cache = None
|
|
_local = threading.local()
|
|
|
|
|
|
def _config() -> dict:
|
|
global _config_cache
|
|
if _config_cache is None:
|
|
with open('config.json') as f:
|
|
_config_cache = json.load(f)['database']
|
|
return _config_cache
|
|
|
|
|
|
def _new_conn(cfg: dict):
|
|
return pymysql.connect(
|
|
host=cfg['host'],
|
|
port=cfg.get('port', 3306),
|
|
user=cfg['user'],
|
|
password=cfg['password'],
|
|
database=cfg['name'],
|
|
autocommit=True,
|
|
cursorclass=pymysql.cursors.DictCursor,
|
|
connect_timeout=10,
|
|
charset='utf8mb4',
|
|
# All app code treats database times as UTC. Without this, a server in another
|
|
# time zone shifts displayed ages and timed suppression expiries by its offset.
|
|
init_command="SET time_zone='+00:00'",
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def get_conn():
|
|
"""Yield a per-thread cached database connection, reconnecting as needed."""
|
|
cfg = _config()
|
|
conn = getattr(_local, 'conn', None)
|
|
if conn is None:
|
|
conn = _new_conn(cfg)
|
|
_local.conn = conn
|
|
else:
|
|
try:
|
|
conn.ping(reconnect=True)
|
|
except Exception:
|
|
try:
|
|
conn.close()
|
|
except Exception:
|
|
pass
|
|
_local.conn = None
|
|
conn = _new_conn(cfg)
|
|
_local.conn = conn
|
|
yield conn
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Monitor state (key/value store)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def set_state(key: str, value) -> None:
|
|
if not isinstance(value, str):
|
|
value = json.dumps(value, default=str)
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""INSERT INTO monitor_state (key_name, value)
|
|
VALUES (%s, %s)
|
|
ON DUPLICATE KEY UPDATE value=VALUES(value), updated_at=NOW()""",
|
|
(key, value),
|
|
)
|
|
|
|
|
|
def get_state(key: str, default=None):
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT value FROM monitor_state WHERE key_name=%s', (key,))
|
|
row = cur.fetchone()
|
|
return row['value'] if row else default
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Interface baseline tracking
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_baseline() -> dict:
|
|
raw = get_state('interface_baseline')
|
|
if raw:
|
|
try:
|
|
return json.loads(raw)
|
|
except Exception:
|
|
logger.error('Failed to parse interface_baseline JSON; resetting baseline')
|
|
return {}
|
|
|
|
|
|
def set_baseline(baseline: dict) -> None:
|
|
set_state('interface_baseline', json.dumps(baseline))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Network events
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def upsert_event(
|
|
event_type: str,
|
|
severity: str,
|
|
source_type: str,
|
|
target_name: str,
|
|
target_detail: str,
|
|
description: str,
|
|
) -> tuple:
|
|
"""Insert or update a network event. Returns (id, is_new, consecutive_failures, ticket_id)."""
|
|
detail = target_detail or ''
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT id, consecutive_failures, ticket_id FROM network_events
|
|
WHERE event_type=%s AND target_name=%s AND target_detail=%s
|
|
AND resolved_at IS NULL LIMIT 1""",
|
|
(event_type, target_name, detail),
|
|
)
|
|
existing = cur.fetchone()
|
|
|
|
if existing:
|
|
new_count = existing['consecutive_failures'] + 1
|
|
cur.execute(
|
|
"""UPDATE network_events
|
|
SET last_seen=NOW(), consecutive_failures=%s, description=%s
|
|
WHERE id=%s""",
|
|
(new_count, description, existing['id']),
|
|
)
|
|
return existing['id'], False, new_count, existing.get('ticket_id')
|
|
else:
|
|
cur.execute(
|
|
"""INSERT INTO network_events
|
|
(event_type, severity, source_type, target_name, target_detail, description)
|
|
VALUES (%s, %s, %s, %s, %s, %s)""",
|
|
(event_type, severity, source_type, target_name, detail, description),
|
|
)
|
|
return cur.lastrowid, True, 1, None
|
|
|
|
|
|
def get_events_for_target(target_name: str, limit: int = 25) -> list:
|
|
"""Return a device's events, unresolved first, then most recent."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT * FROM network_events WHERE target_name=%s
|
|
ORDER BY resolved_at IS NOT NULL, COALESCE(resolved_at, last_seen) DESC
|
|
LIMIT %s""",
|
|
(target_name, limit),
|
|
)
|
|
rows = cur.fetchall()
|
|
for r in rows:
|
|
for k in ('first_seen', 'last_seen', 'resolved_at'):
|
|
if r.get(k) and hasattr(r[k], 'isoformat'):
|
|
r[k] = r[k].isoformat() + 'Z'
|
|
return rows
|
|
|
|
|
|
def resolve_event(event_type: str, target_name: str, target_detail: str = '') -> None:
|
|
detail = target_detail or ''
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""UPDATE network_events SET resolved_at=NOW()
|
|
WHERE event_type=%s AND target_name=%s AND target_detail=%s
|
|
AND resolved_at IS NULL""",
|
|
(event_type, target_name, detail),
|
|
)
|
|
|
|
|
|
def get_open_event_targets(event_type: str) -> list:
|
|
"""(target_name, target_detail) of each unresolved event of one type."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'SELECT target_name, target_detail FROM network_events WHERE event_type=%s AND resolved_at IS NULL',
|
|
(event_type,),
|
|
)
|
|
return [(r['target_name'], r['target_detail']) for r in cur.fetchall()]
|
|
|
|
|
|
def set_ticket_id(event_id: int, ticket_id: str) -> None:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'UPDATE network_events SET ticket_id=%s WHERE id=%s',
|
|
(ticket_id, event_id),
|
|
)
|
|
|
|
|
|
def get_active_events(limit: int = 200, offset: int = 0) -> list:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT * FROM network_events
|
|
WHERE resolved_at IS NULL
|
|
ORDER BY
|
|
FIELD(severity,'critical','warning','info'),
|
|
first_seen DESC
|
|
LIMIT %s OFFSET %s""",
|
|
(limit, offset),
|
|
)
|
|
rows = cur.fetchall()
|
|
for r in rows:
|
|
for k in ('first_seen', 'last_seen'):
|
|
if r.get(k) and hasattr(r[k], 'isoformat'):
|
|
r[k] = r[k].isoformat() + 'Z'
|
|
return rows
|
|
|
|
|
|
def count_active_events() -> int:
|
|
"""Return count of all unresolved events (for pagination)."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT COUNT(*) AS n FROM network_events WHERE resolved_at IS NULL"
|
|
)
|
|
return cur.fetchone()['n']
|
|
|
|
|
|
def get_recent_resolved(hours: int = 24, limit: int = 50) -> list:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT * FROM network_events
|
|
WHERE resolved_at IS NOT NULL
|
|
AND resolved_at > DATE_SUB(NOW(), INTERVAL %s HOUR)
|
|
ORDER BY resolved_at DESC LIMIT %s""",
|
|
(hours, limit),
|
|
)
|
|
rows = cur.fetchall()
|
|
for r in rows:
|
|
for k in ('first_seen', 'last_seen', 'resolved_at'):
|
|
if r.get(k) and hasattr(r[k], 'isoformat'):
|
|
r[k] = r[k].isoformat() + 'Z'
|
|
return rows
|
|
|
|
|
|
def get_status_summary() -> dict:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT severity, COUNT(*) as cnt FROM network_events
|
|
WHERE resolved_at IS NULL GROUP BY severity"""
|
|
)
|
|
counts = {r['severity']: r['cnt'] for r in cur.fetchall()}
|
|
cur.execute(
|
|
"""SELECT COUNT(*) as cnt FROM network_events
|
|
WHERE resolved_at IS NOT NULL
|
|
AND resolved_at > DATE_SUB(NOW(), INTERVAL 24 HOUR)"""
|
|
)
|
|
resolved_24h = cur.fetchone()['cnt']
|
|
return {
|
|
'critical': counts.get('critical', 0),
|
|
'warning': counts.get('warning', 0),
|
|
'info': counts.get('info', 0),
|
|
'resolved_24h': resolved_24h,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Suppression rules
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def get_active_suppressions() -> list:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT * FROM suppression_rules
|
|
WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
|
|
ORDER BY created_at DESC"""
|
|
)
|
|
rows = cur.fetchall()
|
|
for r in rows:
|
|
for k in ('created_at', 'expires_at'):
|
|
if r.get(k) and hasattr(r[k], 'isoformat'):
|
|
r[k] = r[k].isoformat() + 'Z'
|
|
return rows
|
|
|
|
|
|
def get_suppression_history(limit: int = 50) -> list:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'SELECT * FROM suppression_rules ORDER BY created_at DESC LIMIT %s',
|
|
(limit,),
|
|
)
|
|
rows = cur.fetchall()
|
|
for r in rows:
|
|
for k in ('created_at', 'expires_at'):
|
|
if r.get(k) and hasattr(r[k], 'isoformat'):
|
|
r[k] = r[k].isoformat() + 'Z'
|
|
return rows
|
|
|
|
|
|
def create_suppression(
|
|
target_type: str,
|
|
target_name: str,
|
|
target_detail: str,
|
|
reason: str,
|
|
suppressed_by: str,
|
|
expires_minutes: Optional[int] = None,
|
|
) -> int:
|
|
expires_at = None
|
|
if expires_minutes:
|
|
expires_at = datetime.now(timezone.utc) + timedelta(minutes=int(expires_minutes))
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""INSERT INTO suppression_rules
|
|
(target_type, target_name, target_detail, reason, suppressed_by, expires_at, active)
|
|
VALUES (%s, %s, %s, %s, %s, %s, TRUE)""",
|
|
(target_type, target_name or '', target_detail or '', reason, suppressed_by, expires_at),
|
|
)
|
|
return cur.lastrowid
|
|
|
|
|
|
def deactivate_suppression(sup_id: int) -> None:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'UPDATE suppression_rules SET active=FALSE WHERE id=%s', (sup_id,)
|
|
)
|
|
|
|
|
|
def cleanup_expired_suppressions() -> int:
|
|
"""Mark expired time-limited suppressions as inactive. Returns count deactivated."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""UPDATE suppression_rules
|
|
SET active=FALSE
|
|
WHERE active=TRUE AND expires_at IS NOT NULL AND expires_at <= NOW()"""
|
|
)
|
|
n = cur.rowcount
|
|
if n:
|
|
logger.info(f'Deactivated {n} expired suppression(s)')
|
|
return n
|
|
|
|
|
|
def ensure_events_schema() -> None:
|
|
"""Add columns newer than the original events table, so deploys need no manual migration."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute("SHOW COLUMNS FROM network_events LIKE 'recovery_notified'")
|
|
if cur.fetchone():
|
|
return
|
|
cur.execute(
|
|
'ALTER TABLE network_events ADD COLUMN recovery_notified TINYINT(1) NOT NULL DEFAULT 0'
|
|
)
|
|
# Events that cleared before this existed are history, not owed a comment
|
|
cur.execute(
|
|
'UPDATE network_events SET recovery_notified=1, last_seen=last_seen WHERE resolved_at IS NOT NULL'
|
|
)
|
|
|
|
|
|
def get_pending_recovery_notes(hours: int = 24, exclude_types: tuple = ()) -> list:
|
|
"""Resolved events with a ticket whose recovery comment hasn't been posted yet."""
|
|
sql = """SELECT id, event_type, target_name, target_detail, ticket_id,
|
|
first_seen, resolved_at, consecutive_failures
|
|
FROM network_events
|
|
WHERE resolved_at IS NOT NULL AND ticket_id IS NOT NULL AND recovery_notified=0
|
|
AND resolved_at > DATE_SUB(NOW(), INTERVAL %s HOUR)"""
|
|
params: list = [hours]
|
|
if exclude_types:
|
|
sql += f' AND event_type NOT IN ({", ".join(["%s"] * len(exclude_types))})' # nosec B608 - placeholders only
|
|
params.extend(exclude_types)
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql + ' ORDER BY resolved_at', params)
|
|
return list(cur.fetchall())
|
|
|
|
|
|
def mark_recovery_notified(event_id: int) -> None:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
# last_seen updates itself on any write; keep it as the last failing check
|
|
cur.execute(
|
|
'UPDATE network_events SET recovery_notified=1, last_seen=last_seen WHERE id=%s',
|
|
(event_id,),
|
|
)
|
|
|
|
|
|
def event_stats(event_type: str, target_name: str, target_detail: str = '') -> dict:
|
|
"""How often one alert has fired before, for a ticket's history section.
|
|
|
|
Returns {'last_7d', 'last_30d', 'last_first_seen', 'last_resolved_at'} over resolved events.
|
|
"""
|
|
key = (event_type, target_name, target_detail or '')
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT COALESCE(SUM(first_seen > DATE_SUB(NOW(), INTERVAL 7 DAY)), 0) AS last_7d,
|
|
COUNT(*) AS last_30d
|
|
FROM network_events
|
|
WHERE event_type=%s AND target_name=%s AND target_detail=%s
|
|
AND resolved_at IS NOT NULL AND first_seen > DATE_SUB(NOW(), INTERVAL 30 DAY)""",
|
|
key,
|
|
)
|
|
counts = cur.fetchone() or {}
|
|
cur.execute(
|
|
"""SELECT first_seen, resolved_at FROM network_events
|
|
WHERE event_type=%s AND target_name=%s AND target_detail=%s
|
|
AND resolved_at IS NOT NULL ORDER BY resolved_at DESC LIMIT 1""",
|
|
key,
|
|
)
|
|
last = cur.fetchone() or {}
|
|
return {
|
|
'last_7d': int(counts.get('last_7d') or 0),
|
|
'last_30d': int(counts.get('last_30d') or 0),
|
|
'last_first_seen': last.get('first_seen'),
|
|
'last_resolved_at': last.get('resolved_at'),
|
|
}
|
|
|
|
|
|
def _history_where(since_hours: int, event_type: str = '', target: str = '',
|
|
severity: str = '', status: str = 'all') -> tuple:
|
|
"""WHERE clause shared by the history list and its summaries.
|
|
|
|
An event is in range if it was open at any point in the window: it started
|
|
in it, or it was still open when the window began.
|
|
"""
|
|
clauses = ['(first_seen > DATE_SUB(NOW(), INTERVAL %s HOUR) OR '
|
|
'resolved_at IS NULL OR resolved_at > DATE_SUB(NOW(), INTERVAL %s HOUR))']
|
|
params: list = [since_hours, since_hours]
|
|
if event_type:
|
|
clauses.append('event_type=%s')
|
|
params.append(event_type)
|
|
if severity:
|
|
clauses.append('severity=%s')
|
|
params.append(severity)
|
|
if target:
|
|
like = '%' + target.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_') + '%'
|
|
clauses.append('(target_name LIKE %s OR target_detail LIKE %s)')
|
|
params.extend([like, like])
|
|
if status == 'active':
|
|
clauses.append('resolved_at IS NULL')
|
|
elif status == 'resolved':
|
|
clauses.append('resolved_at IS NOT NULL')
|
|
return ' AND '.join(clauses), params
|
|
|
|
|
|
def _iso(rows: list, keys=('first_seen', 'last_seen', 'resolved_at')) -> list:
|
|
for r in rows:
|
|
for k in keys:
|
|
if r.get(k) and hasattr(r[k], 'isoformat'):
|
|
r[k] = r[k].isoformat() + 'Z'
|
|
return rows
|
|
|
|
|
|
def get_events_history(since_hours: int, event_type: str = '', target: str = '', severity: str = '',
|
|
status: str = 'all', limit: int = 100, offset: int = 0) -> dict:
|
|
"""Events open at any time in the last `since_hours`, newest first, with totals.
|
|
|
|
Returns {'events', 'total', 'by_type': {type: count}}.
|
|
"""
|
|
where, params = _history_where(since_hours, event_type, target, severity, status)
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
f"""SELECT id, event_type, severity, source_type, target_name, target_detail, description,
|
|
first_seen, last_seen, resolved_at, consecutive_failures, ticket_id,
|
|
TIMESTAMPDIFF(SECOND, first_seen, COALESCE(resolved_at, NOW())) AS duration_s
|
|
FROM network_events WHERE {where}
|
|
ORDER BY first_seen DESC, id DESC LIMIT %s OFFSET %s""", # nosec B608 - fixed clauses
|
|
params + [limit, offset],
|
|
)
|
|
events = _iso(list(cur.fetchall()))
|
|
cur.execute(
|
|
f"""SELECT event_type, COUNT(*) AS n
|
|
FROM network_events WHERE {where} GROUP BY event_type ORDER BY n DESC""", # nosec B608
|
|
params,
|
|
)
|
|
groups = cur.fetchall()
|
|
return {
|
|
'events': events,
|
|
'total': sum(int(g['n']) for g in groups),
|
|
'by_type': {g['event_type']: int(g['n']) for g in groups},
|
|
}
|
|
|
|
|
|
def get_flappers(since_hours: int, event_type: str = '', target: str = '', severity: str = '',
|
|
min_count: int = 2, limit: int = 10) -> list:
|
|
"""The alerts that fired most often in the window: the ones worth a root cause."""
|
|
where, params = _history_where(since_hours, event_type, target, severity)
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
f"""SELECT event_type, target_name, target_detail, COUNT(*) AS n,
|
|
SUM(TIMESTAMPDIFF(SECOND, first_seen, COALESCE(resolved_at, NOW()))) AS total_s,
|
|
MAX(first_seen) AS last_seen, SUM(resolved_at IS NULL) AS open_now
|
|
FROM network_events WHERE {where}
|
|
GROUP BY event_type, target_name, target_detail
|
|
HAVING n >= %s ORDER BY n DESC, last_seen DESC LIMIT %s""", # nosec B608 - fixed clauses
|
|
params + [min_count, limit],
|
|
)
|
|
rows = list(cur.fetchall())
|
|
for r in rows:
|
|
r['n'], r['total_s'], r['open_now'] = int(r['n']), int(r['total_s'] or 0), int(r['open_now'] or 0)
|
|
return _iso(rows, ('last_seen',))
|
|
|
|
|
|
def get_event_types() -> list:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT DISTINCT event_type FROM network_events ORDER BY event_type')
|
|
return [r['event_type'] for r in cur.fetchall()]
|
|
|
|
|
|
def purge_old_resolved_events(days: int = 90) -> int:
|
|
"""Delete resolved events older than `days` days. Returns count deleted."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""DELETE FROM network_events
|
|
WHERE resolved_at IS NOT NULL
|
|
AND resolved_at < DATE_SUB(NOW(), INTERVAL %s DAY)""",
|
|
(days,),
|
|
)
|
|
n = cur.rowcount
|
|
if n:
|
|
logger.info(f'Purged {n} old resolved event(s) (>{days}d)')
|
|
return n
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WAN modem optical history
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def ensure_modem_schema() -> None:
|
|
"""Create the modem history table so deploys need no manual migration."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""CREATE TABLE IF NOT EXISTS modem_samples (
|
|
sampled_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP PRIMARY KEY,
|
|
rx_power_dbm DECIMAL(5,1) NULL,
|
|
tx_power_dbm DECIMAL(5,1) NULL,
|
|
temp_c DECIMAL(5,1) NULL
|
|
) ENGINE=InnoDB"""
|
|
)
|
|
|
|
|
|
def record_modem_sample(rx_power_dbm, tx_power_dbm, temp_c) -> None:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""INSERT INTO modem_samples (sampled_at, rx_power_dbm, tx_power_dbm, temp_c)
|
|
VALUES (NOW(), %s, %s, %s)
|
|
ON DUPLICATE KEY UPDATE rx_power_dbm=VALUES(rx_power_dbm),
|
|
tx_power_dbm=VALUES(tx_power_dbm), temp_c=VALUES(temp_c)""",
|
|
(rx_power_dbm, tx_power_dbm, temp_c),
|
|
)
|
|
|
|
|
|
def get_modem_rx_values(days: int = 7) -> list:
|
|
"""Return recent Rx power readings (dBm) for computing a drift baseline."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT rx_power_dbm FROM modem_samples
|
|
WHERE rx_power_dbm IS NOT NULL
|
|
AND sampled_at > DATE_SUB(NOW(), INTERVAL %s DAY)""",
|
|
(days,),
|
|
)
|
|
return [float(r['rx_power_dbm']) for r in cur.fetchall()]
|
|
|
|
|
|
def get_modem_history(hours: int = 168, bucket_seconds: int = 3600) -> list:
|
|
"""Return optical history averaged into time buckets, oldest first."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT FLOOR(UNIX_TIMESTAMP(sampled_at) / %s) * %s AS ts,
|
|
ROUND(AVG(rx_power_dbm), 1) AS rx_avg,
|
|
MIN(rx_power_dbm) AS rx_min,
|
|
ROUND(AVG(tx_power_dbm), 1) AS tx_avg,
|
|
ROUND(AVG(temp_c), 1) AS temp_avg
|
|
FROM modem_samples
|
|
WHERE sampled_at > DATE_SUB(NOW(), INTERVAL %s HOUR)
|
|
GROUP BY ts ORDER BY ts""",
|
|
(bucket_seconds, bucket_seconds, hours),
|
|
)
|
|
return [
|
|
{k: (float(v) if v is not None else None) for k, v in row.items()}
|
|
for row in cur.fetchall()
|
|
]
|
|
|
|
|
|
def purge_old_modem_samples(days: int = 90) -> int:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'DELETE FROM modem_samples WHERE sampled_at < DATE_SUB(NOW(), INTERVAL %s DAY)',
|
|
(days,),
|
|
)
|
|
return cur.rowcount
|
|
|
|
|
|
def check_suppressed(suppressions: list, target_type: str, target_name: str, target_detail: str = '') -> bool:
|
|
"""Check suppression against a pre-loaded list (avoids per-call DB queries)."""
|
|
for s in suppressions:
|
|
if s['target_type'] == 'all':
|
|
return True
|
|
if s['target_type'] == target_type and s['target_name'] == target_name:
|
|
if not (s.get('target_detail') or ''):
|
|
return True
|
|
if target_detail and s.get('target_detail') == target_detail:
|
|
return True
|
|
return False
|
|
|
|
|
|
def is_suppressed(target_type: str, target_name: str, target_detail: str = '') -> bool:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
# Global suppression (all)
|
|
cur.execute(
|
|
"""SELECT id FROM suppression_rules
|
|
WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
|
|
AND target_type='all' LIMIT 1"""
|
|
)
|
|
if cur.fetchone():
|
|
return True
|
|
|
|
if not target_name:
|
|
return False
|
|
|
|
# Host-level suppression (covers all interfaces on that host)
|
|
cur.execute(
|
|
"""SELECT id FROM suppression_rules
|
|
WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
|
|
AND target_type=%s AND target_name=%s
|
|
AND target_detail='' LIMIT 1""",
|
|
(target_type, target_name),
|
|
)
|
|
if cur.fetchone():
|
|
return True
|
|
|
|
# Interface/device-specific suppression
|
|
if target_detail:
|
|
cur.execute(
|
|
"""SELECT id FROM suppression_rules
|
|
WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
|
|
AND target_type=%s AND target_name=%s AND target_detail=%s LIMIT 1""",
|
|
(target_type, target_name, target_detail),
|
|
)
|
|
if cur.fetchone():
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cutsheets (see cutsheet.py). Every change writes a cutsheet_history row in
|
|
# the same transaction, so the log can't miss an edit.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_CUTSHEET_SCHEMA = (
|
|
"""CREATE TABLE IF NOT EXISTS cutsheet_items (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
name VARCHAR(100) NOT NULL,
|
|
kind VARCHAR(20) NOT NULL,
|
|
port_count SMALLINT NULL,
|
|
location VARCHAR(255) NOT NULL DEFAULT '',
|
|
rack_u SMALLINT NULL,
|
|
rack_height SMALLINT NULL,
|
|
notes TEXT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
UNIQUE KEY uq_item_name (name)
|
|
) ENGINE=InnoDB""",
|
|
"""CREATE TABLE IF NOT EXISTS cutsheet_cables (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
label VARCHAR(32) NULL,
|
|
a_device VARCHAR(100) NOT NULL,
|
|
a_port VARCHAR(64) NOT NULL DEFAULT '',
|
|
a_side VARCHAR(5) NOT NULL DEFAULT '',
|
|
b_device VARCHAR(100) NOT NULL DEFAULT '',
|
|
b_port VARCHAR(64) NOT NULL DEFAULT '',
|
|
b_side VARCHAR(5) NOT NULL DEFAULT '',
|
|
cable_type VARCHAR(32) NOT NULL DEFAULT '',
|
|
length_m DECIMAL(5,2) NULL,
|
|
color VARCHAR(32) NOT NULL DEFAULT '',
|
|
status VARCHAR(10) NOT NULL DEFAULT 'installed',
|
|
label_applied BOOLEAN NOT NULL DEFAULT FALSE,
|
|
notes TEXT NULL,
|
|
created_by VARCHAR(255) NOT NULL DEFAULT '',
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_by VARCHAR(255) NOT NULL DEFAULT '',
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
UNIQUE KEY uq_cable_label (label),
|
|
INDEX idx_cable_a (a_device, a_port),
|
|
INDEX idx_cable_b (b_device, b_port)
|
|
) ENGINE=InnoDB""",
|
|
"""CREATE TABLE IF NOT EXISTS cutsheet_port_labels (
|
|
item_id INT NOT NULL,
|
|
port VARCHAR(64) NOT NULL,
|
|
label VARCHAR(64) NOT NULL DEFAULT '',
|
|
applied BOOLEAN NOT NULL DEFAULT FALSE,
|
|
PRIMARY KEY (item_id, port)
|
|
) ENGINE=InnoDB""",
|
|
"""CREATE TABLE IF NOT EXISTS cutsheet_history (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
changed_by VARCHAR(255) NOT NULL,
|
|
entity VARCHAR(12) NOT NULL,
|
|
entity_id INT NOT NULL,
|
|
action VARCHAR(10) NOT NULL,
|
|
summary VARCHAR(500) NOT NULL,
|
|
before_json TEXT NULL,
|
|
after_json TEXT NULL,
|
|
INDEX idx_hist_time (changed_at),
|
|
INDEX idx_hist_entity (entity, entity_id)
|
|
) ENGINE=InnoDB""",
|
|
# Where a device lives in the asset inventory (HomeBox), set by hand on its Gandalf page
|
|
"""CREATE TABLE IF NOT EXISTS device_links (
|
|
name VARCHAR(255) NOT NULL PRIMARY KEY,
|
|
inventory_url VARCHAR(512) NOT NULL,
|
|
updated_by VARCHAR(255) NOT NULL,
|
|
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
|
|
) ENGINE=InnoDB""",
|
|
)
|
|
|
|
|
|
def get_device_link(name: str) -> str:
|
|
"""The device's inventory URL, or ''."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT inventory_url FROM device_links WHERE name=%s', (name,))
|
|
row = cur.fetchone()
|
|
return row['inventory_url'] if row else ''
|
|
|
|
|
|
def set_device_link(name: str, inventory_url: str, user: str) -> None:
|
|
"""Save a device's inventory URL; an empty URL removes it."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
if not inventory_url:
|
|
cur.execute('DELETE FROM device_links WHERE name=%s', (name,))
|
|
return
|
|
cur.execute(
|
|
"""INSERT INTO device_links (name, inventory_url, updated_by) VALUES (%s, %s, %s)
|
|
ON DUPLICATE KEY UPDATE inventory_url=VALUES(inventory_url), updated_by=VALUES(updated_by)""",
|
|
(name, inventory_url, user),
|
|
)
|
|
|
|
_ITEM_FIELDS = ('name', 'kind', 'port_count', 'location', 'rack_u', 'rack_height', 'notes')
|
|
_CABLE_FIELDS = ('label', 'a_device', 'a_port', 'a_side', 'b_device', 'b_port', 'b_side',
|
|
'cable_type', 'length_m', 'color', 'status', 'label_applied', 'notes')
|
|
|
|
|
|
class CutsheetConflict(Exception):
|
|
"""A change that would clash with existing records (duplicate name or label, item in use)."""
|
|
|
|
|
|
def ensure_cutsheet_schema() -> None:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
for statement in _CUTSHEET_SCHEMA:
|
|
cur.execute(statement)
|
|
|
|
|
|
def _cs_jsonable(row: Optional[dict]) -> Optional[dict]:
|
|
if row is None:
|
|
return None
|
|
out = {}
|
|
for k, v in row.items():
|
|
if hasattr(v, 'isoformat'):
|
|
v = v.isoformat() + 'Z'
|
|
elif v is not None and k == 'length_m':
|
|
v = float(v)
|
|
elif k in ('label_applied', 'applied'):
|
|
v = bool(v)
|
|
out[k] = v
|
|
return out
|
|
|
|
|
|
def _cs_log(cur, user: str, entity: str, entity_id: int, action: str, summary: str,
|
|
before: Optional[dict], after: Optional[dict]) -> None:
|
|
cur.execute(
|
|
"""INSERT INTO cutsheet_history (changed_by, entity, entity_id, action, summary, before_json, after_json)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s)""",
|
|
(user, entity, entity_id, action, summary[:500],
|
|
json.dumps(_cs_jsonable(before)) if before else None,
|
|
json.dumps(_cs_jsonable(after)) if after else None),
|
|
)
|
|
|
|
|
|
@contextmanager
|
|
def _cs_transaction():
|
|
with get_conn() as conn:
|
|
conn.begin()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
yield cur
|
|
conn.commit()
|
|
except pymysql.err.IntegrityError as e:
|
|
conn.rollback()
|
|
raise CutsheetConflict(_cs_integrity_message(e)) from e
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
|
|
|
|
def _cs_integrity_message(e: Exception) -> str:
|
|
text = str(e)
|
|
if 'uq_cable_label' in text:
|
|
return 'Another cable already has that label'
|
|
if 'uq_item_name' in text:
|
|
return 'Another item already has that name'
|
|
return 'That change conflicts with an existing record'
|
|
|
|
|
|
def cutsheet_load() -> dict:
|
|
"""Every cutsheet record, for one page render or API response."""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT * FROM cutsheet_items ORDER BY name')
|
|
items = [_cs_jsonable(r) for r in cur.fetchall()]
|
|
cur.execute('SELECT * FROM cutsheet_cables ORDER BY id')
|
|
cables = [_cs_jsonable(r) for r in cur.fetchall()]
|
|
cur.execute(
|
|
"""SELECT l.item_id, i.name AS item_name, l.port, l.label, l.applied
|
|
FROM cutsheet_port_labels l JOIN cutsheet_items i ON i.id = l.item_id"""
|
|
)
|
|
port_labels = [_cs_jsonable(r) for r in cur.fetchall()]
|
|
return {'items': items, 'cables': cables, 'port_labels': port_labels}
|
|
|
|
|
|
def cutsheet_history(limit: int = 100) -> list:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""SELECT id, changed_at, changed_by, entity, entity_id, action, summary
|
|
FROM cutsheet_history ORDER BY id DESC LIMIT %s""",
|
|
(limit,),
|
|
)
|
|
return [_cs_jsonable(r) for r in cur.fetchall()]
|
|
|
|
|
|
def _cs_get(cur, table: str, row_id: int) -> Optional[dict]:
|
|
cur.execute(f'SELECT * FROM {table} WHERE id=%s FOR UPDATE', (row_id,)) # nosec B608 (fixed names)
|
|
return cur.fetchone()
|
|
|
|
|
|
def cutsheet_create_item(item: dict, user: str) -> int:
|
|
with _cs_transaction() as cur:
|
|
cur.execute(
|
|
f'INSERT INTO cutsheet_items ({", ".join(_ITEM_FIELDS)}) ' # nosec B608 (fixed names)
|
|
f'VALUES ({", ".join(["%s"] * len(_ITEM_FIELDS))})',
|
|
tuple(item[f] for f in _ITEM_FIELDS),
|
|
)
|
|
item_id = cur.lastrowid
|
|
_cs_log(cur, user, 'item', item_id, 'create', f'Added {item["kind"]} {item["name"]}', None, item)
|
|
return item_id
|
|
|
|
|
|
def cutsheet_update_item(item_id: int, item: dict, user: str) -> bool:
|
|
"""Update an item. A rename carries over to the cables that reference it."""
|
|
with _cs_transaction() as cur:
|
|
before = _cs_get(cur, 'cutsheet_items', item_id)
|
|
if before is None:
|
|
return False
|
|
cur.execute(
|
|
f'UPDATE cutsheet_items SET {", ".join(f"{f}=%s" for f in _ITEM_FIELDS)} WHERE id=%s', # nosec B608
|
|
tuple(item[f] for f in _ITEM_FIELDS) + (item_id,),
|
|
)
|
|
summary = f'Edited {item["name"]}'
|
|
if before['name'] != item['name']:
|
|
for end in ('a', 'b'):
|
|
cur.execute(
|
|
f'UPDATE cutsheet_cables SET {end}_device=%s, updated_by=%s WHERE {end}_device=%s', # nosec B608
|
|
(item['name'], user, before['name']),
|
|
)
|
|
summary = f'Renamed {before["name"]} to {item["name"]}'
|
|
_cs_log(cur, user, 'item', item_id, 'update', summary, before, item)
|
|
return True
|
|
|
|
|
|
def cutsheet_delete_item(item_id: int, user: str) -> bool:
|
|
with _cs_transaction() as cur:
|
|
before = _cs_get(cur, 'cutsheet_items', item_id)
|
|
if before is None:
|
|
return False
|
|
cur.execute(
|
|
'SELECT COUNT(*) AS n FROM cutsheet_cables WHERE a_device=%s OR b_device=%s',
|
|
(before['name'], before['name']),
|
|
)
|
|
used = cur.fetchone()['n']
|
|
if used:
|
|
raise CutsheetConflict(f'{before["name"]} still has {used} cable(s); move or delete them first')
|
|
cur.execute('DELETE FROM cutsheet_port_labels WHERE item_id=%s', (item_id,))
|
|
cur.execute('DELETE FROM cutsheet_items WHERE id=%s', (item_id,))
|
|
_cs_log(cur, user, 'item', item_id, 'delete', f'Removed {before["name"]}', before, None)
|
|
return True
|
|
|
|
|
|
def _cable_summary(c: dict) -> str:
|
|
def end(device: str, port: str) -> str:
|
|
return ' '.join(filter(None, [device or 'unterminated', port]))
|
|
ends = f'{end(c["a_device"], c["a_port"])} ↔ {end(c["b_device"], c["b_port"])}'
|
|
return f'{c.get("label") or "unlabeled cable"}: {ends}'
|
|
|
|
|
|
def cutsheet_create_cable(cable: dict, user: str) -> int:
|
|
with _cs_transaction() as cur:
|
|
cur.execute(
|
|
f'INSERT INTO cutsheet_cables ({", ".join(_CABLE_FIELDS)}, created_by, updated_by) ' # nosec B608
|
|
f'VALUES ({", ".join(["%s"] * len(_CABLE_FIELDS))}, %s, %s)',
|
|
tuple(cable[f] for f in _CABLE_FIELDS) + (user, user),
|
|
)
|
|
cable_id = cur.lastrowid
|
|
_cs_log(cur, user, 'cable', cable_id, 'create', f'Added {_cable_summary(cable)}', None, cable)
|
|
return cable_id
|
|
|
|
|
|
def cutsheet_update_cable(cable_id: int, cable: dict, user: str) -> bool:
|
|
with _cs_transaction() as cur:
|
|
before = _cs_get(cur, 'cutsheet_cables', cable_id)
|
|
if before is None:
|
|
return False
|
|
cur.execute(
|
|
f'UPDATE cutsheet_cables SET {", ".join(f"{f}=%s" for f in _CABLE_FIELDS)}, updated_by=%s ' # nosec B608
|
|
'WHERE id=%s',
|
|
tuple(cable[f] for f in _CABLE_FIELDS) + (user, cable_id),
|
|
)
|
|
_cs_log(cur, user, 'cable', cable_id, 'update', f'Edited {_cable_summary(cable)}', before, cable)
|
|
return True
|
|
|
|
|
|
def cutsheet_delete_cable(cable_id: int, user: str) -> bool:
|
|
with _cs_transaction() as cur:
|
|
before = _cs_get(cur, 'cutsheet_cables', cable_id)
|
|
if before is None:
|
|
return False
|
|
cur.execute('DELETE FROM cutsheet_cables WHERE id=%s', (cable_id,))
|
|
_cs_log(cur, user, 'cable', cable_id, 'delete', f'Deleted {_cable_summary(before)}', before, None)
|
|
return True
|
|
|
|
|
|
def cutsheet_set_port_label(item_id: int, port: str, label: str, applied: bool, user: str) -> bool:
|
|
with _cs_transaction() as cur:
|
|
item = _cs_get(cur, 'cutsheet_items', item_id)
|
|
if item is None:
|
|
return False
|
|
cur.execute('SELECT * FROM cutsheet_port_labels WHERE item_id=%s AND port=%s', (item_id, port))
|
|
before = cur.fetchone()
|
|
if label:
|
|
cur.execute(
|
|
"""INSERT INTO cutsheet_port_labels (item_id, port, label, applied) VALUES (%s, %s, %s, %s)
|
|
ON DUPLICATE KEY UPDATE label=VALUES(label), applied=VALUES(applied)""",
|
|
(item_id, port, label, applied),
|
|
)
|
|
else:
|
|
cur.execute('DELETE FROM cutsheet_port_labels WHERE item_id=%s AND port=%s', (item_id, port))
|
|
after = {'item_id': item_id, 'port': port, 'label': label, 'applied': applied} if label else None
|
|
_cs_log(cur, user, 'port_label', item_id, 'update',
|
|
f'{item["name"]} port {port} label: {label or "(cleared)"}', before, after)
|
|
return True
|
|
|
|
|
|
def cutsheet_import_cables(cables: list, user: str) -> dict:
|
|
"""Add or update cables from a CSV import, matching on label. All or nothing."""
|
|
added = updated = 0
|
|
with _cs_transaction() as cur:
|
|
for cable in cables:
|
|
before = None
|
|
if cable['label']:
|
|
cur.execute('SELECT * FROM cutsheet_cables WHERE label=%s FOR UPDATE', (cable['label'],))
|
|
before = cur.fetchone()
|
|
if before:
|
|
cur.execute(
|
|
f'UPDATE cutsheet_cables SET {", ".join(f"{f}=%s" for f in _CABLE_FIELDS)}, ' # nosec B608
|
|
'updated_by=%s WHERE id=%s',
|
|
tuple(cable[f] for f in _CABLE_FIELDS) + (user, before['id']),
|
|
)
|
|
_cs_log(cur, user, 'cable', before['id'], 'update', f'Imported {_cable_summary(cable)}',
|
|
before, cable)
|
|
updated += 1
|
|
else:
|
|
cur.execute(
|
|
f'INSERT INTO cutsheet_cables ({", ".join(_CABLE_FIELDS)}, created_by, updated_by) ' # nosec
|
|
f'VALUES ({", ".join(["%s"] * len(_CABLE_FIELDS))}, %s, %s)',
|
|
tuple(cable[f] for f in _CABLE_FIELDS) + (user, user),
|
|
)
|
|
_cs_log(cur, user, 'cable', cur.lastrowid, 'create', f'Imported {_cable_summary(cable)}',
|
|
None, cable)
|
|
added += 1
|
|
return {'added': added, 'updated': updated}
|