jaredandClaude Opus 5.5
437f7a415f
Lint / JS (eslint) (push) Successful in 7s
Lint / Python (flake8) (push) Successful in 1m26s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 1m30s
Lint / Python (flake8) (pull_request) Successful in 52s
Test / Python Tests (pytest) (push) Successful in 1m4s
Lint / JS (eslint) (pull_request) Successful in 7s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 1m28s
Test / Python Tests (pytest) (pull_request) Successful in 2m0s
When an alert that opened a ticket clears, the monitor now comments on the ticket (when it cleared, how long it lasted, failed checks, a link back). It never closes or changes a ticket's status. Failed notes retry each cycle for 24 hours; a 401/403 from Tinker stops them until restart. Tickets are now Markdown with a "What to check" list and, where known: the switch port the device plugs into (kept from the last time it was seen, so a dead host still shows it), cutsheet cables and their path, how often the alert fired in the last 7/30 days, other open alerts on the device, and links into Gandalf (gandalf.public_url). Titles are unchanged so Tinker's duplicate detection still works. network_events gains recovery_notified, added at monitor startup; rows that cleared before the upgrade count as already notified. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1916 lines
89 KiB
Python
1916 lines
89 KiB
Python
#!/usr/bin/env python3
|
||
"""Gandalf network monitor daemon.
|
||
|
||
Polls Prometheus (node_exporter) and the UniFi controller for network
|
||
interface and device state. Creates tickets in Tinker Tickets when issues
|
||
are detected, with deduplication and suppression support.
|
||
|
||
Run as a separate systemd service alongside the Flask web app.
|
||
"""
|
||
import json
|
||
import logging
|
||
import re
|
||
import shlex
|
||
import statistics
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from typing import Dict, List, Optional
|
||
from urllib.parse import urljoin, urlparse
|
||
|
||
import requests
|
||
from urllib3.exceptions import InsecureRequestWarning
|
||
|
||
import db
|
||
import cutsheet
|
||
import devices
|
||
import modem as modem_api
|
||
import ticket_text
|
||
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s %(levelname)s %(name)s %(message)s',
|
||
)
|
||
logger = logging.getLogger('gandalf.monitor')
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Interface filtering
|
||
# --------------------------------------------------------------------------
|
||
_SKIP_PREFIXES = (
|
||
'lo', 'veth', 'tap', 'fwbr', 'fwln', 'fwpr',
|
||
'docker', 'dummy', 'br-', 'virbr', 'vmbr',
|
||
)
|
||
_VLAN_SUFFIX = re.compile(r'\.\d+$')
|
||
|
||
|
||
def is_physical_interface(name: str) -> bool:
|
||
"""Return True for physical/bond interfaces worth monitoring."""
|
||
if any(name.startswith(p) for p in _SKIP_PREFIXES):
|
||
return False
|
||
if _VLAN_SUFFIX.search(name):
|
||
return False
|
||
return True
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Prometheus client
|
||
# --------------------------------------------------------------------------
|
||
class PrometheusClient:
|
||
def __init__(self, url: str):
|
||
self.url = url.rstrip('/')
|
||
|
||
def _fetch(self, promql: str) -> Optional[list]:
|
||
"""Return query results, or None if the query itself failed."""
|
||
try:
|
||
resp = requests.get(
|
||
f'{self.url}/api/v1/query',
|
||
params={'query': promql},
|
||
timeout=15,
|
||
)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
if data.get('status') == 'success':
|
||
return data['data']['result']
|
||
logger.error(f'Prometheus query unsuccessful ({promql!r}): {data}')
|
||
except Exception as e:
|
||
logger.error(f'Prometheus query failed ({promql!r}): {e}')
|
||
return None
|
||
|
||
def query(self, promql: str) -> list:
|
||
return self._fetch(promql) or []
|
||
|
||
def get_interface_states(self) -> Optional[Dict[str, Dict[str, bool]]]:
|
||
"""Return {instance: {device: is_up}} for physical interfaces.
|
||
|
||
Returns None when Prometheus can't answer or has no series at all, so
|
||
callers can tell "we can't see anything" apart from "this host is gone".
|
||
"""
|
||
results = self._fetch('node_network_up')
|
||
if not results:
|
||
return None
|
||
hosts: Dict[str, Dict[str, bool]] = {}
|
||
for r in results:
|
||
instance = r['metric'].get('instance', '')
|
||
device = r['metric'].get('device', '')
|
||
if not is_physical_interface(device):
|
||
continue
|
||
hosts.setdefault(instance, {})[device] = (r['value'][1] == '1')
|
||
return hosts
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# UniFi client
|
||
# --------------------------------------------------------------------------
|
||
class UnifiClient:
|
||
# Devices whose port tables are collected: switches and the gateway/router
|
||
_PORT_DEVICE_TYPES = ('usw', 'udm', 'ugw', 'uxg')
|
||
_SYSTEM_LOG_MAX_PAGES = 10
|
||
|
||
def __init__(self, cfg: dict):
|
||
self.base_url = cfg['controller']
|
||
self.site_id = cfg.get('site_id', 'default')
|
||
self.session = requests.Session()
|
||
self.session.verify = cfg.get('verify_ssl', True)
|
||
if not self.session.verify:
|
||
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
|
||
self.headers = {
|
||
'X-API-KEY': cfg['api_key'],
|
||
'Accept': 'application/json',
|
||
}
|
||
|
||
def get_device_details(self) -> Optional[Dict[str, dict]]:
|
||
"""Return firmware, load, WAN, and outlet details per device name.
|
||
|
||
Uses the v1 stat API, the only one that includes these fields.
|
||
"""
|
||
try:
|
||
url = f'{self.base_url}/proxy/network/api/s/{self.site_id}/stat/device'
|
||
resp = self.session.get(url, headers=self.headers, timeout=15)
|
||
resp.raise_for_status()
|
||
return {
|
||
d.get('name') or d.get('mac', 'unknown'): self._device_detail(d)
|
||
for d in resp.json().get('data', [])
|
||
}
|
||
except Exception as e:
|
||
logger.error(f'UniFi device detail error: {e}')
|
||
return None
|
||
|
||
@staticmethod
|
||
def _device_detail(d: dict) -> dict:
|
||
def number(raw) -> Optional[float]:
|
||
try:
|
||
return float(raw)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
stats = d.get('system-stats') or {}
|
||
wan = {}
|
||
for key in ('wan1', 'wan2'):
|
||
w = d.get(key) or {}
|
||
if w:
|
||
wan[key] = {
|
||
'up': w.get('up'), 'speed_mbps': w.get('speed'),
|
||
'latency_ms': w.get('latency'), 'availability': w.get('availability'),
|
||
}
|
||
return {
|
||
'mac': d.get('mac', ''),
|
||
'firmware': d.get('version', ''),
|
||
'uptime_s': d.get('uptime'),
|
||
'cpu_pct': number(stats.get('cpu')),
|
||
'mem_pct': number(stats.get('mem')),
|
||
'clients': d.get('num_sta'),
|
||
'temps': [
|
||
{'name': t.get('name', ''), 'value_c': number(t.get('value'))}
|
||
for t in d.get('temperatures') or []
|
||
],
|
||
'wan': wan,
|
||
'power_w': number(d.get('outlet_ac_power_consumption')),
|
||
'outlets': [
|
||
{
|
||
'index': o.get('index'),
|
||
'name': o.get('name', ''),
|
||
'on': o.get('relay_state'),
|
||
'auto_power_cycle': o.get('cycle_enabled'),
|
||
'power_w': number(o.get('outlet_power')),
|
||
'current_a': number(o.get('outlet_current')),
|
||
}
|
||
for o in d.get('outlet_table') or []
|
||
],
|
||
}
|
||
|
||
def get_wired_clients(self) -> List[dict]:
|
||
"""Return wired clients as {mac, sw_mac, sw_port}; empty if unavailable."""
|
||
try:
|
||
url = f'{self.base_url}/proxy/network/api/s/{self.site_id}/stat/sta'
|
||
resp = self.session.get(url, headers=self.headers, timeout=15)
|
||
resp.raise_for_status()
|
||
return [
|
||
{'mac': c.get('mac', ''), 'sw_mac': c.get('sw_mac', ''), 'sw_port': c.get('sw_port')}
|
||
for c in resp.json().get('data', []) if c.get('is_wired')
|
||
]
|
||
except Exception as e:
|
||
logger.error(f'UniFi client list error: {e}')
|
||
return []
|
||
|
||
def get_system_events(self, since_ms: int, keys: tuple) -> Optional[List[dict]]:
|
||
"""Return system log entries with the given keys, or None if the log is unavailable."""
|
||
url = f'{self.base_url}/proxy/network/v2/api/site/{self.site_id}/system-log/all'
|
||
found: List[dict] = []
|
||
try:
|
||
for page in range(self._SYSTEM_LOG_MAX_PAGES):
|
||
resp = self.session.post(
|
||
url, headers=self.headers, timeout=20,
|
||
json={
|
||
'timestampFrom': since_ms, 'timestampTo': int(time.time() * 1000),
|
||
'pageNumber': page, 'pageSize': 200,
|
||
},
|
||
)
|
||
resp.raise_for_status()
|
||
items = resp.json().get('data') or []
|
||
if not items:
|
||
break
|
||
for e in items:
|
||
if e.get('key') in keys and e.get('timestamp', 0) >= since_ms:
|
||
params = e.get('parameters') or {}
|
||
found.append({
|
||
'key': e['key'], 'ts': e['timestamp'] / 1000,
|
||
'device': (params.get('DEVICE') or {}).get('name', ''),
|
||
'wan': (params.get('WAN_ID') or {}).get('name', '')
|
||
or str((params.get('WAN_ID') or {}).get('id', '')),
|
||
})
|
||
return found
|
||
except Exception as e:
|
||
logger.error(f'UniFi system log error: {e}')
|
||
return None
|
||
|
||
def get_devices(self) -> Optional[List[dict]]:
|
||
"""Return list of UniFi devices, or None if the controller is unreachable."""
|
||
try:
|
||
url = f'{self.base_url}/proxy/network/v2/api/site/{self.site_id}/device'
|
||
resp = self.session.get(url, headers=self.headers, timeout=15)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
devices = []
|
||
for d in data.get('network_devices', []):
|
||
state = d.get('state', 1)
|
||
devices.append({
|
||
'name': d.get('name') or d.get('mac', 'unknown'),
|
||
'mac': d.get('mac', ''),
|
||
'ip': d.get('ip', ''),
|
||
'type': d.get('type', 'unknown'),
|
||
'model': d.get('model', ''),
|
||
'state': state,
|
||
'connected': state == 1,
|
||
})
|
||
return devices
|
||
except Exception as e:
|
||
logger.error(f'UniFi API error: {e}')
|
||
return None
|
||
|
||
@staticmethod
|
||
def _sfp_detail(port: dict) -> Optional[dict]:
|
||
"""The module in an SFP cage, or None for a port that has no cage.
|
||
|
||
Every cage reports whether a module is present and what it is. Optical
|
||
readings (DOM) only come from modules that report them, such as fibre
|
||
optics; copper DACs have none. Switches and the UDM spell the fault flags
|
||
differently.
|
||
"""
|
||
if 'sfp_found' not in port:
|
||
return None
|
||
if not port.get('sfp_found'):
|
||
return {'found': False}
|
||
|
||
def number(key: str) -> Optional[float]:
|
||
try:
|
||
return float(port[key])
|
||
except (KeyError, TypeError, ValueError):
|
||
return None
|
||
|
||
part = str(port.get('sfp_part') or '').strip()
|
||
compliance = str(port.get('sfp_compliance') or '').strip()
|
||
if compliance.lower() == 'unknown':
|
||
compliance = ''
|
||
rx_power = number('sfp_rxpower')
|
||
if 'DAC' in compliance.upper() or re.search(r'(^|[-_])CU\d', part, re.IGNORECASE):
|
||
kind = 'dac'
|
||
elif rx_power is not None or compliance:
|
||
kind = 'optic'
|
||
else:
|
||
kind = ''
|
||
return {
|
||
'found': True,
|
||
'kind': kind,
|
||
'vendor': str(port.get('sfp_vendor') or '').strip(),
|
||
'part': part,
|
||
'serial': str(port.get('sfp_serial') or '').strip(),
|
||
'compliance': compliance,
|
||
'rx_power_dbm': rx_power,
|
||
'tx_power_dbm': number('sfp_txpower'),
|
||
'temperature_c': number('sfp_temperature'),
|
||
'voltage_v': number('sfp_voltage'),
|
||
'current_ma': number('sfp_current'),
|
||
'rx_fault': bool(port.get('sfp_rxfault', port.get('sfp_rx_los', False))),
|
||
'tx_fault': bool(port.get('sfp_txfault', port.get('sfp_tx_fault', False))),
|
||
}
|
||
|
||
@staticmethod
|
||
def _pair_dac_ends(switches: Dict[str, dict]) -> None:
|
||
"""Link the two ends of each DAC cable: both report the cable's serial."""
|
||
ends: Dict[str, List[tuple]] = {}
|
||
for sw_name, sw in switches.items():
|
||
for port in sw['ports'].values():
|
||
sfp = port.get('sfp') or {}
|
||
if sfp.get('kind') == 'dac' and sfp.get('serial'):
|
||
ends.setdefault(sfp['serial'], []).append((sw_name, port))
|
||
for pair in ends.values():
|
||
if len(pair) != 2:
|
||
continue # one end not seen, or a serial that isn't unique
|
||
for (_, port), (peer_name, peer) in (pair, pair[::-1]):
|
||
port['sfp']['cable_peer'] = {'switch': peer_name, 'port_idx': peer['port_idx']}
|
||
|
||
def get_switch_ports(self) -> Optional[Dict[str, dict]]:
|
||
"""Return per-port stats for UniFi switches and routers, keyed by device name.
|
||
|
||
Uses the v1 stat API which includes full port_table data.
|
||
Returns {switch_name: {'ip': str, 'model': str, 'ports': {port_name: {...}}}}.
|
||
"""
|
||
try:
|
||
url = f'{self.base_url}/proxy/network/api/s/{self.site_id}/stat/device'
|
||
resp = self.session.get(url, headers=self.headers, timeout=15)
|
||
resp.raise_for_status()
|
||
devices = resp.json().get('data', [])
|
||
result: Dict[str, dict] = {}
|
||
for dev in devices:
|
||
if dev.get('type', '').lower() not in self._PORT_DEVICE_TYPES:
|
||
continue
|
||
sw_name = dev.get('name') or dev.get('mac', 'unknown')
|
||
sw_ip = dev.get('ip', '')
|
||
sw_model = dev.get('model', '')
|
||
ports: Dict[str, dict] = {}
|
||
# Build LLDP neighbor map (keyed by port_idx)
|
||
lldp_map: Dict[int, dict] = {}
|
||
for entry in dev.get('lldp_table', []):
|
||
# Current UniFi Network versions use local_port_idx
|
||
pidx = entry.get('local_port_idx', entry.get('lldp_port_idx'))
|
||
if pidx is not None:
|
||
lldp_map[int(pidx)] = {
|
||
'chassis_id': entry.get('chassis_id', ''),
|
||
'system_name': entry.get('system_name', ''),
|
||
'port_id': entry.get('port_id', ''),
|
||
'port_desc': entry.get('port_desc', ''),
|
||
'mgmt_ips': entry.get('management_ips', []),
|
||
}
|
||
for port in dev.get('port_table', []):
|
||
idx = port.get('port_idx', 0)
|
||
pname = port.get('name') or f'Port {idx}'
|
||
# Ports are keyed by name, and UniFi lets two ports share one
|
||
# (e.g. a stale label copied to the next port). Without this the
|
||
# later port silently replaced the earlier one.
|
||
if pname in ports:
|
||
pname = f'{pname} (Port {idx})'
|
||
raw_poe = port.get('poe_power')
|
||
raw_poe_max = port.get('poe_max_power')
|
||
ports[pname] = {
|
||
'port_idx': idx,
|
||
'switch_ip': sw_ip,
|
||
'up': port.get('up', False),
|
||
'speed_mbps': port.get('speed', 0),
|
||
'full_duplex': port.get('full_duplex', False),
|
||
'autoneg': port.get('autoneg', False),
|
||
'is_uplink': port.get('is_uplink', False),
|
||
'media': port.get('media', ''),
|
||
'poe_power': float(raw_poe) if raw_poe is not None else None,
|
||
'poe_class': port.get('poe_class'),
|
||
'poe_max_power': float(raw_poe_max) if raw_poe_max is not None else None,
|
||
'poe_mode': port.get('poe_mode', ''),
|
||
'lldp': lldp_map.get(idx),
|
||
'tx_bytes': port.get('tx_bytes', 0),
|
||
'rx_bytes': port.get('rx_bytes', 0),
|
||
'tx_errors': port.get('tx_errors', 0),
|
||
'rx_errors': port.get('rx_errors', 0),
|
||
'tx_dropped': port.get('tx_dropped', 0),
|
||
'rx_dropped': port.get('rx_dropped', 0),
|
||
'sfp': self._sfp_detail(port),
|
||
}
|
||
if ports:
|
||
result[sw_name] = {'ip': sw_ip, 'model': sw_model, 'ports': ports}
|
||
self._pair_dac_ends(result)
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f'UniFi switch port stats error: {e}')
|
||
return None
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Ticket client
|
||
# --------------------------------------------------------------------------
|
||
class TicketClient:
|
||
def __init__(self, cfg: dict):
|
||
self.url = cfg.get('url', '')
|
||
self.api_key = cfg.get('api_key', '')
|
||
# Tinker serves its other API scripts under api/ beside create_ticket_api.php
|
||
self.comment_url = cfg.get('comment_url') or (
|
||
urljoin(self.url, 'api/ticket_comment_api.php') if self.url else '')
|
||
self.comments_denied = False
|
||
|
||
def comment(self, ticket_id: str, text: str) -> bool:
|
||
"""Add a Markdown comment to a ticket. Never raises; False means try again later."""
|
||
if not self.api_key or not self.comment_url or self.comments_denied:
|
||
return False
|
||
try:
|
||
resp = requests.post(
|
||
self.comment_url,
|
||
json={'ticket_id': str(ticket_id), 'comment_text': text, 'markdown_enabled': True},
|
||
headers={'Authorization': f'Bearer {self.api_key}'},
|
||
timeout=15,
|
||
)
|
||
if resp.status_code in (401, 403):
|
||
# Retrying every cycle can't fix a key without write access, so stop until restart
|
||
self.comments_denied = True
|
||
logger.error(
|
||
f'Ticket API refused a comment on #{ticket_id} (HTTP {resp.status_code}). '
|
||
f'The API key needs read_write scope; recovery notes are off until the monitor restarts.'
|
||
)
|
||
return False
|
||
if resp.status_code == 404:
|
||
logger.warning(f'Ticket #{ticket_id} no longer exists; skipping its recovery note')
|
||
return True
|
||
resp.raise_for_status()
|
||
if resp.json().get('success'):
|
||
logger.info(f'Commented on ticket #{ticket_id}')
|
||
return True
|
||
logger.warning(f'Unexpected comment API response for #{ticket_id}: {resp.text[:200]}')
|
||
except Exception as e:
|
||
logger.error(f'Ticket comment failed for #{ticket_id}: {e}')
|
||
return False
|
||
|
||
def create(self, title: str, description: str, priority: str = '2') -> Optional[str]:
|
||
if not self.api_key or not self.url:
|
||
logger.warning('Ticket API not configured – skipping ticket creation')
|
||
return None
|
||
try:
|
||
resp = requests.post(
|
||
self.url,
|
||
json={
|
||
'title': title,
|
||
'description': description,
|
||
'status': 'Open',
|
||
'priority': priority,
|
||
'category': 'Network',
|
||
'type': 'Issue',
|
||
},
|
||
headers={'Authorization': f'Bearer {self.api_key}'},
|
||
timeout=15,
|
||
)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
if data.get('success'):
|
||
tid = data.get('ticket_id')
|
||
if not tid:
|
||
logger.warning(f'Ticket API success but no ticket_id in response: {data}')
|
||
return None
|
||
logger.info(f'Created ticket #{tid}: {title}')
|
||
return tid
|
||
if data.get('existing_ticket_id'):
|
||
logger.info(f'Duplicate suppressed by API – existing #{data["existing_ticket_id"]}')
|
||
return data['existing_ticket_id']
|
||
logger.warning(f'Unexpected ticket API response: {data}')
|
||
except Exception as e:
|
||
logger.error(f'Ticket creation failed: {e}')
|
||
return None
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Pulse HTTP client (delegates SSH commands to Pulse worker)
|
||
# --------------------------------------------------------------------------
|
||
class PulseClient:
|
||
"""Submit a command to a Pulse worker via the internal M2M API and poll for result."""
|
||
|
||
def __init__(self, cfg: dict):
|
||
p = cfg.get('pulse', {})
|
||
self.url = p.get('url', '').rstrip('/')
|
||
self.api_key = p.get('api_key', '')
|
||
self.worker_id = p.get('worker_id', '')
|
||
self.timeout = p.get('timeout', 45)
|
||
self.last_execution_id: Optional[str] = None
|
||
self.session = requests.Session()
|
||
self.session.headers.update({
|
||
'X-Gandalf-API-Key': self.api_key,
|
||
'Content-Type': 'application/json',
|
||
})
|
||
|
||
def run_command(self, command: str, _retry: bool = True) -> Optional[str]:
|
||
"""Submit *command* to Pulse, poll until done, return stdout or None.
|
||
|
||
Retries once automatically on transient submit failures or timeouts.
|
||
"""
|
||
self.last_execution_id = None
|
||
if not self.url or not self.api_key or not self.worker_id:
|
||
return None
|
||
|
||
# Submit
|
||
try:
|
||
resp = self.session.post(
|
||
f'{self.url}/api/internal/command',
|
||
json={'worker_id': self.worker_id, 'command': command},
|
||
timeout=10,
|
||
)
|
||
resp.raise_for_status()
|
||
execution_id = resp.json().get('execution_id')
|
||
if not execution_id:
|
||
logger.error('Pulse submit response missing execution_id')
|
||
return None
|
||
self.last_execution_id = execution_id
|
||
except Exception as e:
|
||
logger.error(f'Pulse command submit failed: {e}')
|
||
if _retry:
|
||
logger.info('Retrying Pulse command submit in 5s...')
|
||
time.sleep(5)
|
||
return self.run_command(command, _retry=False)
|
||
return None
|
||
|
||
# Poll
|
||
deadline = time.time() + self.timeout
|
||
while time.time() < deadline:
|
||
time.sleep(2)
|
||
try:
|
||
r = self.session.get(
|
||
f'{self.url}/api/internal/executions/{execution_id}',
|
||
timeout=10,
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
status = data.get('status')
|
||
if status == 'completed':
|
||
logs = data.get('logs', [])
|
||
for entry in logs:
|
||
if entry.get('action') == 'command_result':
|
||
return entry.get('stdout', '')
|
||
return ''
|
||
if status in ('failed', 'timed_out', 'cancelled'):
|
||
logger.error(
|
||
f'Pulse execution {execution_id} ended with status={status!r}; '
|
||
f'view at {self.url}/executions/{execution_id}'
|
||
)
|
||
if _retry and status != 'cancelled':
|
||
logger.info('Retrying failed Pulse command in 5s...')
|
||
time.sleep(5)
|
||
return self.run_command(command, _retry=False)
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f'Pulse poll failed for {execution_id}: {e}')
|
||
|
||
logger.warning(
|
||
f'Pulse command timed out after {self.timeout}s '
|
||
f'(execution_id={execution_id}); '
|
||
f'view at {self.url}/executions/{execution_id}'
|
||
)
|
||
if _retry:
|
||
logger.info('Retrying timed-out Pulse command in 5s...')
|
||
time.sleep(5)
|
||
return self.run_command(command, _retry=False)
|
||
return None
|
||
|
||
def ping(self, ip: str, count: int = 3, timeout: int = 2) -> Optional[bool]:
|
||
"""Ping *ip* via the Pulse worker.
|
||
|
||
Returns True/False for the host's reachability, or None if Pulse itself
|
||
couldn't run the check, which says nothing about the host.
|
||
"""
|
||
ip_q = shlex.quote(ip)
|
||
output = self.run_command(
|
||
f'ping -c {count} -W {timeout} {ip_q} >/dev/null 2>&1 && echo REACHABLE || echo UNREACHABLE'
|
||
)
|
||
if output is None:
|
||
return None
|
||
return output.strip() == 'REACHABLE'
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Link stats collector (ethtool + Prometheus traffic metrics)
|
||
# --------------------------------------------------------------------------
|
||
class LinkStatsCollector:
|
||
"""Collects detailed per-interface statistics via SSH (ethtool) and Prometheus,
|
||
plus per-port stats from UniFi switches."""
|
||
|
||
def __init__(self, cfg: dict, prom: 'PrometheusClient',
|
||
unifi: Optional['UnifiClient'] = None):
|
||
self.cfg = cfg
|
||
self.prom = prom
|
||
self.pulse = PulseClient(cfg)
|
||
self.unifi = unifi
|
||
# State for UniFi rate calculation (previous snapshot + timestamp)
|
||
self._prev_unifi: Dict[str, dict] = {}
|
||
self._prev_unifi_time: float = 0.0
|
||
|
||
# ------------------------------------------------------------------
|
||
# SSH collection (via Pulse worker)
|
||
# ------------------------------------------------------------------
|
||
def _ssh_batch(self, ip: str, ifaces: List[str]) -> Dict[str, dict]:
|
||
"""
|
||
Delegate one SSH session to the Pulse worker to collect ethtool + SFP DOM
|
||
data for all *ifaces*. Returns {iface: {speed_mbps, duplex, ..., sfp: {...}}}.
|
||
"""
|
||
if not ifaces or not self.pulse.url:
|
||
return {}
|
||
|
||
# Validate interface names (kernel names: [a-zA-Z0-9_.-], max 15 chars per IFNAMSIZ)
|
||
safe_ifaces = [i for i in ifaces if re.match(r'^[a-zA-Z0-9_.-]{1,15}$', i)]
|
||
if not safe_ifaces:
|
||
return {}
|
||
|
||
# Build a single shell command: for each iface output ethtool + -m with sentinels
|
||
parts = []
|
||
for iface in safe_ifaces:
|
||
q = shlex.quote(iface)
|
||
parts.append(
|
||
f'echo "___IFACE:{iface}___";'
|
||
f' ethtool {q} 2>/dev/null;'
|
||
f' echo "___DOM:{iface}___";'
|
||
f' ethtool -m {q} 2>/dev/null;'
|
||
f' echo "___END___"'
|
||
)
|
||
shell_cmd = ' '.join(parts)
|
||
|
||
ssh_cmd = (
|
||
f'ssh -o StrictHostKeyChecking=accept-new -o ConnectTimeout=5 '
|
||
f'-o BatchMode=yes -o LogLevel=ERROR '
|
||
f'-o ServerAliveInterval=10 -o ServerAliveCountMax=2 '
|
||
f'root@{ip} {shlex.quote(shell_cmd)}'
|
||
)
|
||
output = self.pulse.run_command(ssh_cmd)
|
||
if output is None:
|
||
logger.error(f'Pulse ethtool collection returned None for {ip}')
|
||
return {}
|
||
|
||
return self._parse_ssh_output(output)
|
||
|
||
@staticmethod
|
||
def _parse_ssh_output(output: str) -> Dict[str, dict]:
|
||
result: Dict[str, dict] = {}
|
||
current_iface: Optional[str] = None
|
||
current_section: Optional[str] = None
|
||
buf: List[str] = []
|
||
|
||
def flush(iface, section, lines):
|
||
if not iface or not lines:
|
||
return
|
||
text = '\n'.join(lines)
|
||
if section == 'ethtool':
|
||
result.setdefault(iface, {}).update(
|
||
LinkStatsCollector._parse_ethtool(text)
|
||
)
|
||
elif section == 'dom':
|
||
sfp = LinkStatsCollector._parse_ethtool_m(text)
|
||
if sfp:
|
||
result.setdefault(iface, {})['sfp'] = sfp
|
||
|
||
for line in output.splitlines():
|
||
if line.startswith('___IFACE:') and line.endswith('___'):
|
||
flush(current_iface, current_section, buf)
|
||
current_iface = line[9:-3]
|
||
current_section = 'ethtool'
|
||
buf = []
|
||
elif line.startswith('___DOM:') and line.endswith('___'):
|
||
flush(current_iface, current_section, buf)
|
||
current_iface = line[7:-3]
|
||
current_section = 'dom'
|
||
buf = []
|
||
elif line == '___END___':
|
||
flush(current_iface, current_section, buf)
|
||
current_iface = None
|
||
current_section = None
|
||
buf = []
|
||
else:
|
||
buf.append(line)
|
||
|
||
flush(current_iface, current_section, buf)
|
||
return result
|
||
|
||
@staticmethod
|
||
def _parse_ethtool(output: str) -> dict:
|
||
data: dict = {}
|
||
for line in output.splitlines():
|
||
if ':' not in line:
|
||
continue
|
||
key, _, val = line.partition(':')
|
||
key = key.strip()
|
||
val = val.strip()
|
||
if key == 'Speed':
|
||
m = re.match(r'(\d+)Mb/s', val)
|
||
if m:
|
||
data['speed_mbps'] = int(m.group(1))
|
||
elif key == 'Duplex':
|
||
data['duplex'] = val.lower()
|
||
elif key == 'Port':
|
||
data['port_type'] = val
|
||
elif key == 'Auto-negotiation':
|
||
data['auto_neg'] = (val.lower() == 'on')
|
||
elif key == 'Link detected':
|
||
data['link_detected'] = (val.lower() == 'yes')
|
||
return data
|
||
|
||
@staticmethod
|
||
def _parse_ethtool_m(output: str) -> dict:
|
||
"""Parse ethtool -m (SFP DOM / digital optical monitoring) output."""
|
||
if not output:
|
||
return {}
|
||
# Skip if module diagnostics unsupported
|
||
lower = output.lower()
|
||
if 'cannot get' in lower or 'not supported' in lower or 'no sfp' in lower:
|
||
return {}
|
||
|
||
data: dict = {}
|
||
for line in output.splitlines():
|
||
if ':' not in line:
|
||
continue
|
||
key, _, val = line.partition(':')
|
||
key = key.strip()
|
||
val = val.strip()
|
||
|
||
if key == 'Vendor name':
|
||
data['vendor'] = val
|
||
elif key == 'Vendor PN':
|
||
data['part_no'] = val
|
||
elif key == 'Identifier':
|
||
m = re.search(r'\((.+?)\)', val)
|
||
if m:
|
||
data['sfp_type'] = m.group(1)
|
||
elif key == 'Connector':
|
||
m = re.search(r'\((.+?)\)', val)
|
||
if m:
|
||
data['connector'] = m.group(1)
|
||
elif key == 'Laser wavelength':
|
||
m = re.match(r'(\d+)', val)
|
||
if m:
|
||
data['wavelength_nm'] = int(m.group(1))
|
||
elif key == 'Laser bias current':
|
||
# e.g. "4.340 mA"
|
||
m = re.match(r'([\d.]+)\s+mA', val)
|
||
if m:
|
||
data['bias_ma'] = float(m.group(1))
|
||
elif key == 'Laser output power':
|
||
# e.g. "0.1234 mW / -9.09 dBm"
|
||
m = re.search(r'/\s*([-\d.]+)\s*dBm', val)
|
||
if m:
|
||
try:
|
||
data['tx_power_dbm'] = float(m.group(1))
|
||
except ValueError:
|
||
pass
|
||
elif 'receiver' in key.lower() and ('power' in key.lower() or 'optical' in key.lower()):
|
||
m = re.search(r'/\s*([-\d.]+)\s*dBm', val)
|
||
if m:
|
||
try:
|
||
data['rx_power_dbm'] = float(m.group(1))
|
||
except ValueError:
|
||
pass
|
||
elif key == 'Module temperature':
|
||
# e.g. "36.00 degrees C / 96.80 degrees F"
|
||
m = re.match(r'([\d.]+)\s+degrees', val)
|
||
if m:
|
||
data['temp_c'] = float(m.group(1))
|
||
elif key == 'Module voltage':
|
||
# e.g. "3.3180 V"
|
||
m = re.match(r'([\d.]+)\s+V', val)
|
||
if m:
|
||
data['voltage_v'] = float(m.group(1))
|
||
|
||
return data
|
||
|
||
# ------------------------------------------------------------------
|
||
# Prometheus traffic / error metrics
|
||
# ------------------------------------------------------------------
|
||
def _collect_prom_metrics(self) -> Dict[str, Dict[str, dict]]:
|
||
"""Return {instance: {device: {tx_bytes_rate, rx_bytes_rate, ...}}}."""
|
||
metrics: Dict[str, Dict[str, dict]] = {}
|
||
|
||
queries = [
|
||
('tx_bytes_rate', 'rate(node_network_transmit_bytes_total[5m])'),
|
||
('rx_bytes_rate', 'rate(node_network_receive_bytes_total[5m])'),
|
||
('tx_errs_rate', 'rate(node_network_transmit_errs_total[5m])'),
|
||
('rx_errs_rate', 'rate(node_network_receive_errs_total[5m])'),
|
||
('tx_drops_rate', 'rate(node_network_transmit_drop_total[5m])'),
|
||
('rx_drops_rate', 'rate(node_network_receive_drop_total[5m])'),
|
||
('carrier_changes', 'node_network_carrier_changes_total'),
|
||
]
|
||
|
||
for field, promql in queries:
|
||
for r in self.prom.query(promql):
|
||
instance = r['metric'].get('instance', '')
|
||
device = r['metric'].get('device', '')
|
||
if not is_physical_interface(device):
|
||
continue
|
||
raw = r['value'][1]
|
||
try:
|
||
val: Optional[float] = float(raw)
|
||
if val != val: # NaN
|
||
val = None
|
||
except (ValueError, TypeError):
|
||
val = None
|
||
metrics.setdefault(instance, {}).setdefault(device, {})[field] = val
|
||
|
||
return metrics
|
||
|
||
# ------------------------------------------------------------------
|
||
# Main collection entry point
|
||
# ------------------------------------------------------------------
|
||
def collect(self, instance_map: Dict[str, str]) -> dict:
|
||
"""
|
||
Collect full link stats for all Prometheus-monitored hosts.
|
||
|
||
*instance_map*: ``{'10.10.10.2:9100': 'large1', ...}``
|
||
|
||
Returns a dict suitable for ``db.set_state('link_stats', ...)``.
|
||
"""
|
||
prom_metrics = self._collect_prom_metrics()
|
||
result_hosts: Dict[str, Dict[str, dict]] = {}
|
||
exclude_ips = set(self.cfg.get('monitor', {}).get('links_exclude_ips', []))
|
||
|
||
for instance, iface_metrics in prom_metrics.items():
|
||
host_ip = instance.split(':')[0]
|
||
if host_ip in exclude_ips:
|
||
continue
|
||
host = instance_map.get(instance, host_ip)
|
||
ifaces = list(iface_metrics.keys())
|
||
|
||
# SSH ethtool collection via Pulse worker — only for explicitly configured
|
||
# hosts (instance_map keys). Hosts like postgresql/matrix may report
|
||
# node_exporter metrics to Prometheus but don't need link diagnostics.
|
||
ethtool_data: Dict[str, dict] = {}
|
||
if self.pulse.url and ifaces and instance in instance_map:
|
||
try:
|
||
ethtool_data = self._ssh_batch(host_ip, ifaces)
|
||
except Exception as e:
|
||
logger.warning(f'ethtool collection failed for {host} ({host_ip}): {e}')
|
||
|
||
# Merge Prometheus metrics + ethtool data per interface
|
||
merged: Dict[str, dict] = {}
|
||
for iface in ifaces:
|
||
d: dict = {'host_ip': host_ip}
|
||
d.update(iface_metrics.get(iface, {}))
|
||
eth = ethtool_data.get(iface, {})
|
||
for k, v in eth.items():
|
||
if k != 'sfp':
|
||
d[k] = v
|
||
if 'sfp' in eth:
|
||
d['sfp'] = eth['sfp']
|
||
merged[iface] = d
|
||
|
||
result_hosts[host] = merged
|
||
|
||
# Collect UniFi switch port stats
|
||
unifi_switches: dict = {}
|
||
if self.unifi:
|
||
try:
|
||
raw = self.unifi.get_switch_ports()
|
||
if raw is not None:
|
||
now = time.time()
|
||
unifi_switches = self._compute_unifi_rates(raw, now)
|
||
self._prev_unifi = raw
|
||
self._prev_unifi_time = now
|
||
except Exception as e:
|
||
logger.warning(f'UniFi switch port collection failed: {e}')
|
||
|
||
unifi_devices = (self.unifi.get_device_details() or {}) if self.unifi else {}
|
||
self._resolve_neighbors(unifi_switches, unifi_devices, instance_map)
|
||
|
||
return {
|
||
'hosts': result_hosts,
|
||
'unifi_switches': unifi_switches,
|
||
'unifi_devices': unifi_devices,
|
||
'updated': datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC'),
|
||
}
|
||
|
||
def _resolve_neighbors(self, switches: dict, unifi_devices: dict, instance_map: Dict[str, str]) -> None:
|
||
"""Name port neighbors: UniFi's LLDP data has only MACs, and many servers send none."""
|
||
host_macs: Dict[str, tuple] = {}
|
||
host_ips: Dict[str, str] = {}
|
||
# Bonds share their MAC with their member NICs, and LLDP runs on the NIC.
|
||
results = sorted(
|
||
self.prom.query('node_network_info'),
|
||
key=lambda r: r['metric'].get('device', '').startswith('bond'),
|
||
)
|
||
for r in results:
|
||
metric = r['metric']
|
||
device, mac = metric.get('device', ''), metric.get('address', '').lower()
|
||
instance = metric.get('instance', '')
|
||
if not is_physical_interface(device) or not mac or mac == '00:00:00:00:00:00':
|
||
continue
|
||
ip = instance.split(':')[0]
|
||
host = instance_map.get(instance, ip)
|
||
host_macs.setdefault(mac, (host, device))
|
||
host_ips[ip] = host
|
||
device_macs = {
|
||
info['mac'].lower(): name for name, info in unifi_devices.items() if info.get('mac')
|
||
}
|
||
devices.resolve_lldp_neighbors(switches, host_macs, device_macs, host_ips)
|
||
devices.apply_client_locations(switches, self.unifi.get_wired_clients(), host_macs, device_macs)
|
||
|
||
def _compute_unifi_rates(self, raw: Dict[str, dict], now: float) -> Dict[str, dict]:
|
||
"""Compute per-port byte/error rates from delta against previous snapshot."""
|
||
dt = now - self._prev_unifi_time if self._prev_unifi_time > 0 else 0
|
||
|
||
def rate(new_val: int, old_val: Optional[int]) -> Optional[float]:
|
||
# No earlier reading (first poll, or a newly reported port) means no rate,
|
||
# not "all bytes since boot arrived in this one interval".
|
||
if dt <= 0 or old_val is None:
|
||
return None
|
||
return max(0.0, (new_val - old_val) / dt)
|
||
|
||
result: Dict[str, dict] = {}
|
||
for sw_name, sw_data in raw.items():
|
||
# Match by port number: names are re-read from UniFi every poll, so a
|
||
# port renamed there keeps its rates instead of losing them for a cycle.
|
||
prev_ports = {
|
||
p.get('port_idx'): p
|
||
for p in self._prev_unifi.get(sw_name, {}).get('ports', {}).values()
|
||
}
|
||
merged_ports: Dict[str, dict] = {}
|
||
for pname, d in sw_data['ports'].items():
|
||
entry = dict(d)
|
||
prev = prev_ports.get(d.get('port_idx'), {})
|
||
entry['tx_bytes_rate'] = rate(d['tx_bytes'], prev.get('tx_bytes'))
|
||
entry['rx_bytes_rate'] = rate(d['rx_bytes'], prev.get('rx_bytes'))
|
||
entry['tx_errs_rate'] = rate(d['tx_errors'], prev.get('tx_errors'))
|
||
entry['rx_errs_rate'] = rate(d['rx_errors'], prev.get('rx_errors'))
|
||
entry['tx_drops_rate'] = rate(d['tx_dropped'], prev.get('tx_dropped'))
|
||
entry['rx_drops_rate'] = rate(d['rx_dropped'], prev.get('rx_dropped'))
|
||
merged_ports[pname] = entry
|
||
result[sw_name] = {'ip': sw_data['ip'], 'model': sw_data['model'],
|
||
'ports': merged_ports}
|
||
return result
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Helpers
|
||
# --------------------------------------------------------------------------
|
||
def _now_utc() -> str:
|
||
return datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')
|
||
|
||
|
||
def _utcnow() -> datetime:
|
||
"""Naive UTC, to compare with database times."""
|
||
return datetime.now(timezone.utc).replace(tzinfo=None)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Monitor
|
||
# --------------------------------------------------------------------------
|
||
CLUSTER_NAME = 'proxmox-cluster'
|
||
DEFAULT_PUBLIC_URL = 'https://gandalf.lotusguild.org'
|
||
# Reports of something already over: the ticket is the whole story, no recovery note
|
||
ONE_SHOT_EVENTS = ('modem_power_cycled', 'wan_unstable')
|
||
# Ports a host or device was last seen on, kept so a ticket for a dead host
|
||
# (whose link, and so its LLDP entry, is gone) can still say where it plugs in
|
||
KNOWN_UPLINKS_STATE = 'known_uplinks'
|
||
|
||
|
||
class NetworkMonitor:
|
||
def __init__(self):
|
||
with open('config.json') as f:
|
||
self.cfg = json.load(f)
|
||
|
||
prom_url = self.cfg['prometheus']['url']
|
||
self.prom = PrometheusClient(prom_url)
|
||
self.unifi = UnifiClient(self.cfg['unifi'])
|
||
self.tickets = TicketClient(self.cfg.get('ticket_api', {}))
|
||
self.link_stats = LinkStatsCollector(self.cfg, self.prom, self.unifi)
|
||
self.pulse = self.link_stats.pulse # convenience alias
|
||
|
||
mon = self.cfg.get('monitor', {})
|
||
self.poll_interval = mon.get('poll_interval', 120)
|
||
self.fail_thresh = mon.get('failure_threshold', 2)
|
||
# UniFi gear (switches/APs) can take several minutes to rejoin after a
|
||
# firmware-update reboot, much longer than a Proxmox host link flap.
|
||
# Use a separate, more forgiving threshold so those momentary reboots
|
||
# don't cut tickets, while still catching genuine device failures.
|
||
self.unifi_fail_thresh = mon.get('unifi_failure_threshold', self.fail_thresh)
|
||
# A host's exporter stops as soon as a shutdown begins and a normal reboot
|
||
# takes about 3-5 minutes, so wait longer than that before ticketing a host
|
||
# that has disappeared. A NIC going down on a running host stays on fail_thresh.
|
||
self.host_fail_thresh = mon.get('host_failure_threshold', 5)
|
||
self.cluster_thresh = mon.get('cluster_threshold', 3)
|
||
self.cluster_name = mon.get('cluster_name', CLUSTER_NAME)
|
||
|
||
# Optional WAN modem (AT&T BGW-series); see modem.py
|
||
modem_cfg = self.cfg.get('modem') or {}
|
||
self.modem = modem_api.ModemClient(modem_cfg) if modem_cfg.get('url') else None
|
||
self.modem_name = modem_cfg.get('name', 'att-modem')
|
||
# Modem reboots (including AT&T's automatic firmware updates) take a few
|
||
# minutes, so this can be raised independently of failure_threshold.
|
||
self.modem_fail_thresh = modem_cfg.get('failure_threshold', self.fail_thresh)
|
||
self.rx_drift_db = modem_cfg.get('rx_drift_db', 3.0)
|
||
self.rx_drift_min_samples = modem_cfg.get('rx_drift_min_samples', 288)
|
||
self.modem_lan_ip = urlparse(modem_cfg['url']).hostname if modem_cfg.get('url') else ''
|
||
self._prev_modem_counters: Optional[dict] = None
|
||
self._prev_modem_time = 0.0
|
||
if self.modem:
|
||
try:
|
||
db.ensure_modem_schema()
|
||
except Exception as e:
|
||
logger.error(f'Could not create modem history table: {e}')
|
||
|
||
# Cutsheet mismatches open a P4 ticket once they've lasted this long (0 = never ticket)
|
||
self.cutsheet_ticket_hours = (self.cfg.get('cutsheet') or {}).get('mismatch_ticket_hours', 24)
|
||
try:
|
||
db.ensure_cutsheet_schema()
|
||
except Exception as e:
|
||
logger.error(f'Could not create cutsheet tables: {e}')
|
||
try:
|
||
db.ensure_events_schema()
|
||
except Exception as e:
|
||
logger.error(f'Could not update the events table: {e}')
|
||
|
||
# Absolute links in tickets point here
|
||
self.public_url = ((self.cfg.get('gandalf') or {}).get('public_url')
|
||
or DEFAULT_PUBLIC_URL).rstrip('/')
|
||
self._link_data: Optional[dict] = None
|
||
self._known_uplinks: Optional[dict] = None
|
||
|
||
# Last good Prometheus result, shown (as unknown) during a Prometheus outage
|
||
self._last_iface_states: Dict[str, Dict[str, bool]] = {}
|
||
|
||
# Build Prometheus instance → hostname lookup
|
||
self._instance_map: Dict[str, str] = {
|
||
h['prometheus_instance']: h['name']
|
||
for h in self.cfg.get('hosts', [])
|
||
if 'prometheus_instance' in h
|
||
}
|
||
|
||
def _hostname(self, instance: str) -> str:
|
||
return self._instance_map.get(instance, instance.split(':')[0])
|
||
|
||
# ------------------------------------------------------------------
|
||
# Tickets: context, and notes when an alert clears
|
||
# ------------------------------------------------------------------
|
||
def _base_url(self) -> str:
|
||
return getattr(self, 'public_url', DEFAULT_PUBLIC_URL)
|
||
|
||
def _open_ticket(
|
||
self, event_id: int, event_type: str, target: str, detail: str, *,
|
||
title: str, heading: str, facts: list, summary: str, checks: list, priority: str,
|
||
host: Optional[str] = None, port: Optional[tuple] = None,
|
||
) -> None:
|
||
"""Open a ticket with the alert's facts plus everything Gandalf knows around it.
|
||
|
||
host: a device whose switch ports to list. port: (switch, port_idx) the alert is about.
|
||
"""
|
||
sections = self._ticket_sections(event_id, event_type, target, detail, host=host, port=port)
|
||
desc = ticket_text.description(heading, facts, summary, checks, sections)
|
||
tid = self.tickets.create(title, desc, priority=priority)
|
||
if tid:
|
||
db.set_ticket_id(event_id, tid)
|
||
|
||
def _ticket_sections(self, event_id: int, event_type: str, target: str, detail: str,
|
||
host: Optional[str] = None, port: Optional[tuple] = None) -> list:
|
||
"""Each section is optional: a failure loses that section, never the ticket."""
|
||
sections = []
|
||
for title, build in (
|
||
('Where it is connected', lambda: self._where_lines(event_type, detail, host, port)),
|
||
('History', lambda: ticket_text.history_lines(db.event_stats(event_type, target, detail), _utcnow())),
|
||
('Other active alerts here', lambda: self._other_alert_lines(event_id, {target, host})),
|
||
('Links', lambda: self._link_lines(host or (port[0] if port else None))),
|
||
):
|
||
try:
|
||
sections.append((title, build()))
|
||
except Exception as e:
|
||
logger.error(f'Ticket section "{title}" for {target} failed: {e}', exc_info=True)
|
||
return sections
|
||
|
||
def _switches(self) -> dict:
|
||
return (getattr(self, '_link_data', None) or {}).get('unifi_switches') or {}
|
||
|
||
def _remember_uplinks(self, link_data: dict) -> None:
|
||
"""Record where every LLDP-visible device is plugged in, for when it disappears."""
|
||
known = self._load_known_uplinks()
|
||
switches = link_data.get('unifi_switches') or {}
|
||
names = {(p.get('lldp') or {}).get('system_name')
|
||
for sw in switches.values() for p in (sw.get('ports') or {}).values()}
|
||
now = _now_utc()
|
||
changed = False
|
||
for name in sorted(n for n in names if n):
|
||
ups = devices.uplinks_for_host(name, switches)
|
||
entry = {'seen': now, 'uplinks': [
|
||
{k: u[k] for k in ('switch', 'port', 'port_idx', 'host_interface')} for u in ups]}
|
||
if (known.get(name) or {}).get('uplinks') != entry['uplinks']:
|
||
changed = True
|
||
known[name] = entry
|
||
self._known_uplinks = known
|
||
# Write when wiring changes, and hourly so "last seen" stays roughly right
|
||
last_saved = getattr(self, '_known_uplinks_saved', 0.0)
|
||
if changed or time.time() - last_saved > 3600:
|
||
db.set_state(KNOWN_UPLINKS_STATE, known)
|
||
self._known_uplinks_saved = time.time()
|
||
|
||
def _load_known_uplinks(self) -> dict:
|
||
if getattr(self, '_known_uplinks', None) is None:
|
||
self._known_uplinks = {}
|
||
raw = db.get_state(KNOWN_UPLINKS_STATE)
|
||
if raw:
|
||
try:
|
||
self._known_uplinks = json.loads(raw)
|
||
except ValueError:
|
||
logger.error('Failed to parse known_uplinks JSON; starting over')
|
||
return self._known_uplinks
|
||
|
||
def _uplinks_for(self, name: str, switches: dict) -> tuple:
|
||
"""(uplinks, None) if the device is linked now, else (last known uplinks, when seen)."""
|
||
current = devices.uplinks_for_host(name, switches)
|
||
if current:
|
||
return current, None
|
||
entry = self._load_known_uplinks().get(name) or {}
|
||
return entry.get('uplinks') or [], entry.get('seen')
|
||
|
||
def _where_lines(self, event_type: str, detail: str, host: Optional[str], port: Optional[tuple]) -> list:
|
||
base, switches = self._base_url(), self._switches()
|
||
lines = []
|
||
if host:
|
||
ups, seen = self._uplinks_for(host, switches)
|
||
if event_type == 'interface_down':
|
||
ups = [u for u in ups if u.get('host_interface') == detail] or ups
|
||
for u in ups:
|
||
live = (switches.get(u['switch'], {}).get('ports') or {}).get(u['port'])
|
||
text = ticket_text.port_text(u['switch'], u['port'], live or dict(u, up=False))
|
||
iface = f'`{u["host_interface"]}` → ' if u.get('host_interface') else ''
|
||
lines.append(f'- {iface}' + ticket_text.link(
|
||
text, ticket_text.device_url(base, u['switch'], u.get('port_idx'))))
|
||
if seen:
|
||
lines[-1] += f' _(last seen linked there {seen})_'
|
||
if not ups:
|
||
lines.append('- Gandalf has not seen this device on a switch port (no LLDP or client entry).')
|
||
if port:
|
||
switch, idx = port
|
||
live = next((dict(p, name=n) for n, p in (switches.get(switch, {}).get('ports') or {}).items()
|
||
if str(p.get('port_idx')) == str(idx)), None)
|
||
if live:
|
||
lines.append('- ' + ticket_text.link(ticket_text.port_text(switch, live['name'], live),
|
||
ticket_text.device_url(base, switch, idx)))
|
||
device = host or (port[0] if port else None)
|
||
if device:
|
||
data = db.cutsheet_load()
|
||
items = {i['name']: i for i in data['items']}
|
||
runs = cutsheet.runs_for(device, data['cables'], items, switches,
|
||
port=str(port[1]) if port else None)
|
||
for r in runs:
|
||
label = ticket_text.link(f'**{r["label"]}**', ticket_text.cutsheet_url(base, r['cable_id']))
|
||
lines.append(f'- Cable {label}: {r["path"]}. Cutsheet check: '
|
||
f'{cutsheet.VERDICTS.get(r["verdict"], r["verdict"]).lower()}.')
|
||
if not runs:
|
||
lines.append('- ' + ticket_text.link('Not on the cutsheet yet', f'{base}/cutsheet' if base else ''))
|
||
return lines
|
||
|
||
def _other_alert_lines(self, event_id: int, names: set) -> list:
|
||
lines = []
|
||
for name in sorted(n for n in names if n):
|
||
for e in db.get_events_for_target(name, limit=25):
|
||
if e.get('resolved_at') or e.get('id') == event_id:
|
||
continue
|
||
what = e['event_type'].replace('_', ' ')
|
||
where = f'{e["target_name"]} {e.get("target_detail") or ""}'.strip()
|
||
ticket = f' (ticket #{e["ticket_id"]})' if e.get('ticket_id') else ''
|
||
lines.append(f'- {what} on {where} since {e.get("first_seen", "?")}{ticket}')
|
||
return lines[:10]
|
||
|
||
def _link_lines(self, device: Optional[str]) -> list:
|
||
base = self._base_url()
|
||
if not base:
|
||
return []
|
||
lines = []
|
||
if device:
|
||
lines.append(f'- {ticket_text.link(f"{device} in Gandalf", ticket_text.device_url(base, device))}')
|
||
lines.append(f'- {ticket_text.link("Gandalf dashboard", base + "/")}')
|
||
return lines
|
||
|
||
def _event_url(self, event: dict) -> str:
|
||
base = self._base_url()
|
||
if not base:
|
||
return ''
|
||
if event['event_type'] == 'cable_mismatch':
|
||
cable_id = str(event.get('target_detail') or '').replace('cable ', '')
|
||
return ticket_text.cutsheet_url(base, cable_id)
|
||
if event['event_type'] in ('cluster_network_issue', 'monitor_source_unavailable'):
|
||
return base + '/'
|
||
return ticket_text.device_url(base, event['target_name'])
|
||
|
||
def _post_recovery_notes(self) -> None:
|
||
"""Comment on the tickets of alerts that have cleared. Failures retry next cycle (for 24 h)."""
|
||
if self.tickets.comments_denied:
|
||
return
|
||
now = _utcnow()
|
||
for event in db.get_pending_recovery_notes(hours=24, exclude_types=ONE_SHOT_EVENTS):
|
||
text = ticket_text.recovery_comment(event, now, self._event_url(event))
|
||
if self.tickets.comment(event['ticket_id'], text):
|
||
db.mark_recovery_notified(event['id'])
|
||
elif self.tickets.comments_denied:
|
||
break
|
||
|
||
# ------------------------------------------------------------------
|
||
# Interface monitoring (Prometheus)
|
||
# ------------------------------------------------------------------
|
||
def _process_interfaces(self, states: Dict[str, Dict[str, bool]], suppressions: list) -> None:
|
||
baseline = db.get_baseline()
|
||
new_baseline = {k: dict(v) for k, v in baseline.items()}
|
||
# Only count hosts with genuine regressions (UP→DOWN) toward cluster threshold
|
||
hosts_with_regression: List[str] = []
|
||
|
||
for instance, ifaces in states.items():
|
||
if instance not in self._instance_map:
|
||
continue # skip unconfigured Prometheus instances
|
||
host = self._hostname(instance)
|
||
new_baseline.setdefault(host, {})
|
||
host_has_regression = False
|
||
|
||
for iface, is_up in ifaces.items():
|
||
prev = baseline.get(host, {}).get(iface) # 'up', 'initial_down', or None
|
||
|
||
if is_up:
|
||
new_baseline[host][iface] = 'up'
|
||
db.resolve_event('interface_down', host, iface)
|
||
else:
|
||
if prev is None:
|
||
# First observation is down – could be unused port, don't alert
|
||
new_baseline[host][iface] = 'initial_down'
|
||
|
||
elif prev == 'initial_down':
|
||
# Persistently down since first observation – no alert
|
||
pass
|
||
|
||
else: # prev == 'up'
|
||
# Regression: was UP, now DOWN
|
||
host_has_regression = True
|
||
sup = (
|
||
db.check_suppressed(suppressions, 'interface', host, iface) or
|
||
db.check_suppressed(suppressions, 'host', host)
|
||
)
|
||
event_id, is_new, consec, ticket_id = db.upsert_event(
|
||
'interface_down', 'critical', 'prometheus',
|
||
host, iface,
|
||
f'Interface {iface} on {host} went link-down ({_now_utc()})',
|
||
)
|
||
if not sup and consec >= self.fail_thresh and not ticket_id:
|
||
self._ticket_interface(event_id, host, iface, consec)
|
||
|
||
if host_has_regression:
|
||
hosts_with_regression.append(host)
|
||
|
||
# A dead host doesn't report its interfaces as down: its series just
|
||
# disappear, so a configured host that's absent needs its own check.
|
||
for instance, host in self._instance_map.items():
|
||
if instance in states:
|
||
db.resolve_event('host_metrics_missing', host, instance)
|
||
continue
|
||
hosts_with_regression.append(host)
|
||
sup = db.check_suppressed(suppressions, 'host', host)
|
||
event_id, is_new, consec, ticket_id = db.upsert_event(
|
||
'host_metrics_missing', 'critical', 'prometheus',
|
||
host, instance,
|
||
f'No metrics from {host} ({instance}): host or node_exporter is down ({_now_utc()})',
|
||
)
|
||
if not sup and consec >= self.host_fail_thresh and not ticket_id:
|
||
self._ticket_metrics_missing(event_id, host, instance, consec)
|
||
|
||
db.set_baseline(new_baseline)
|
||
|
||
# Cluster-wide check – only genuine regressions count
|
||
if len(hosts_with_regression) >= self.cluster_thresh:
|
||
sup = db.check_suppressed(suppressions, 'all', '')
|
||
event_id, is_new, consec, ticket_id = db.upsert_event(
|
||
'cluster_network_issue', 'critical', 'prometheus',
|
||
self.cluster_name, '',
|
||
f'{len(hosts_with_regression)} hosts with simultaneous interface or metrics failures: '
|
||
f'{", ".join(hosts_with_regression)}',
|
||
)
|
||
if not sup and (is_new or not ticket_id):
|
||
self._ticket_cluster(event_id, hosts_with_regression)
|
||
else:
|
||
db.resolve_event('cluster_network_issue', self.cluster_name, '')
|
||
|
||
def _ticket_cluster(self, event_id: int, hosts: List[str]) -> None:
|
||
switches = self._switches()
|
||
# The switches these hosts hang off are the likeliest shared cause
|
||
shared = sorted({u['switch'] for h in hosts for u in self._uplinks_for(h, switches)[0]})
|
||
self._open_ticket(
|
||
event_id, 'cluster_network_issue', self.cluster_name, '',
|
||
title=(f'[{self.cluster_name}][auto][production][issue][network][cluster-wide] '
|
||
f'Multiple hosts reporting interface failures'),
|
||
heading='Cluster Network Alert',
|
||
facts=[('Affected hosts', ', '.join(hosts)),
|
||
('Switches they connect through', ', '.join(shared)),
|
||
('Detected', _now_utc())],
|
||
summary=(f'{len(hosts)} Proxmox hosts reported interface regressions (link-down on '
|
||
f'interfaces previously known up) or stopped reporting metrics at the same time. '
|
||
f'This usually means a switch or upstream network failure rather than the hosts.'),
|
||
checks=['Check the core and management switches (power, uplinks, recent firmware updates).',
|
||
'Check whether the hosts share a PDU or circuit that lost power.',
|
||
'Look at the per-host tickets and the Gandalf dashboard for which links went down first.'],
|
||
priority='1',
|
||
)
|
||
|
||
def _ticket_metrics_missing(
|
||
self, event_id: int, host: str, instance: str, consec: int
|
||
) -> None:
|
||
self._open_ticket(
|
||
event_id, 'host_metrics_missing', host, instance,
|
||
title=f'[{host}][auto][production][issue][network][single-node] No metrics from host',
|
||
heading='Host Metrics Alert',
|
||
facts=[('Host', host), ('Prometheus instance', instance), ('Detected', _now_utc()),
|
||
('Consecutive check failures', str(consec)),
|
||
('Down for at least', ticket_text.duration(consec * self._poll_seconds()))],
|
||
summary=(f'Prometheus is working but has no node_network_up series for {host}. '
|
||
f'Either the host is down or node_exporter has stopped on it. '
|
||
f'Gandalf waits {self.host_fail_thresh} checks before opening this, so a normal '
|
||
f'reboot should not cause it.'),
|
||
checks=[f'Is {host} powered on and reachable (Proxmox UI, ping, console)?',
|
||
'If it is up, check `systemctl status prometheus-node-exporter` on the host.',
|
||
'If it is down, check its switch port below: a dark port with no PoE or link '
|
||
'points to power or cabling.'],
|
||
priority='2', host=host,
|
||
)
|
||
|
||
def _poll_seconds(self) -> int:
|
||
return int(getattr(self, 'poll_interval', 120) or 120)
|
||
|
||
def _process_prometheus_outage(self, suppressions: list) -> None:
|
||
"""Interface and host alerts can't fire without Prometheus, so say so."""
|
||
sup = db.check_suppressed(suppressions, 'host', 'prometheus')
|
||
event_id, is_new, consec, ticket_id = db.upsert_event(
|
||
'monitor_source_unavailable', 'critical', 'prometheus',
|
||
'prometheus', '',
|
||
f'Prometheus returned no interface data; host and interface alerting is paused ({_now_utc()})',
|
||
)
|
||
if not sup and consec >= self.fail_thresh and not ticket_id:
|
||
self._open_ticket(
|
||
event_id, 'monitor_source_unavailable', 'prometheus', '',
|
||
title='[prometheus][auto][production][issue][network][cluster-wide] Gandalf has no Prometheus data',
|
||
heading='Monitoring Source Alert',
|
||
facts=[('Prometheus URL', self.prom.url), ('Detected', _now_utc()),
|
||
('Consecutive check failures', str(consec))],
|
||
summary=('Gandalf could not get node_network_up data from Prometheus. While this lasts, '
|
||
'host and interface failures will **not** be detected. UniFi, ping and modem '
|
||
'checks keep running.'),
|
||
checks=['Is the Prometheus service running on its host?',
|
||
'Open Prometheus → Status → Targets and check the node exporters are being scraped.',
|
||
'Check disk space on the Prometheus host (a full disk stops ingestion).'],
|
||
priority='2',
|
||
)
|
||
|
||
def _ticket_interface(
|
||
self, event_id: int, host: str, iface: str, consec: int
|
||
) -> None:
|
||
self._open_ticket(
|
||
event_id, 'interface_down', host, iface,
|
||
title=f'[{host}][auto][production][issue][network][single-node] Interface {iface} link-down',
|
||
heading='Network Interface Alert',
|
||
facts=[('Host', host), ('Interface', f'`{iface}`'), ('Detected', _now_utc()),
|
||
('Consecutive check failures', str(consec))],
|
||
summary=(f'Interface {iface} on {host} was up and is now reporting link-down via '
|
||
f'Prometheus node_exporter. {host} itself is still reporting, so it may be '
|
||
f'reachable over its other interfaces.'),
|
||
checks=['Check the switch port below: is it down on the switch side too?',
|
||
'Reseat or swap the cable or DAC; for an optic, check the module for a fault.',
|
||
f'On {host}, check `ip link show {iface}` and `dmesg` for NIC or driver errors.'],
|
||
priority='2', host=host,
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# UniFi device monitoring
|
||
# ------------------------------------------------------------------
|
||
def _process_unifi(self, devices: Optional[List[dict]], suppressions: list) -> None:
|
||
if devices is None:
|
||
logger.warning('UniFi API unreachable this cycle')
|
||
return
|
||
|
||
for d in devices:
|
||
name = d['name']
|
||
if not d['connected']:
|
||
sup = db.check_suppressed(suppressions, 'unifi_device', name)
|
||
event_id, is_new, consec, ticket_id = db.upsert_event(
|
||
'unifi_device_offline', 'critical', 'unifi',
|
||
name, d.get('type', ''),
|
||
f'UniFi {name} ({d.get("ip","")}) offline ({_now_utc()})',
|
||
)
|
||
if not sup and consec >= self.unifi_fail_thresh and not ticket_id:
|
||
self._ticket_unifi(event_id, d)
|
||
else:
|
||
db.resolve_event('unifi_device_offline', name, d.get('type', ''))
|
||
|
||
def _ticket_unifi(self, event_id: int, device: dict) -> None:
|
||
name = device['name']
|
||
self._open_ticket(
|
||
event_id, 'unifi_device_offline', name, device.get('type', ''),
|
||
title=f'[{name}][auto][production][issue][network][single-node] UniFi device offline',
|
||
heading='UniFi Device Alert',
|
||
facts=[('Device', name), ('Type', device.get('type', 'unknown')), ('Model', device.get('model', '')),
|
||
('Last known IP', device.get('ip', 'unknown')), ('Detected', _now_utc())],
|
||
summary=(f'The UniFi controller reports {name} as offline. Gandalf waits '
|
||
f'{self.unifi_fail_thresh} checks first, so a firmware-update reboot should not cause this.'),
|
||
checks=['Check the device for power (status LED) and, if it is PoE powered, the PoE '
|
||
'budget on the port below.',
|
||
'Check the uplink cable and the switch port it plugs into.',
|
||
'In the UniFi controller, check whether the device is stuck adopting or upgrading.'],
|
||
priority='2', host=name,
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Cutsheet: documented cabling vs. what discovery sees
|
||
# ------------------------------------------------------------------
|
||
def _process_cutsheet(self, link_data: dict, suppressions: list) -> None:
|
||
"""Keep one event per mismatched cable run, and ticket the ones that last.
|
||
|
||
Events appear on the dashboard and the switch's page straight away; a
|
||
ticket waits for cutsheet.mismatch_ticket_hours, so recabling never pages.
|
||
"""
|
||
data = db.cutsheet_load()
|
||
items = {i['name']: i for i in data['items']}
|
||
switches = link_data.get('unifi_switches') or {}
|
||
found = cutsheet.problems(data['cables'], items, switches)
|
||
hours = self.cutsheet_ticket_hours
|
||
threshold = max(1, -(-int(hours * 3600) // max(1, self.poll_interval))) if hours else 0
|
||
|
||
current = set()
|
||
for p in found:
|
||
current.add((p['switch'], p['key']))
|
||
description = f'{p["label"]}: {p["detail"]} (path: {p["path"]})'
|
||
event_id, _, consec, ticket_id = db.upsert_event(
|
||
'cable_mismatch', 'warning', 'cutsheet', p['switch'], p['key'], description,
|
||
)
|
||
sup = (db.check_suppressed(suppressions, 'unifi_device', p['switch']) or
|
||
db.check_suppressed(suppressions, 'interface', p['switch'], p['port']))
|
||
if threshold and not sup and consec >= threshold and not ticket_id:
|
||
self._ticket_cable_mismatch(event_id, p, hours)
|
||
|
||
for target_name, target_detail in db.get_open_event_targets('cable_mismatch'):
|
||
if (target_name, target_detail) not in current:
|
||
db.resolve_event('cable_mismatch', target_name, target_detail)
|
||
|
||
def _ticket_cable_mismatch(self, event_id: int, p: dict, hours: float) -> None:
|
||
url = ticket_text.cutsheet_url(self._base_url(), p['cable_id'])
|
||
self._open_ticket(
|
||
event_id, 'cable_mismatch', p['switch'], p['key'],
|
||
title=(f'[{p["switch"]}][auto][production][issue][network][single-node] '
|
||
f'Cable {p["label"]} does not match the cutsheet'),
|
||
heading='Cutsheet Mismatch',
|
||
facts=[('Cable', ticket_text.link(p['label'], url)), ('Switch port', f'{p["switch"]} port {p["port"]}'),
|
||
('Problem', p['detail']), ('Documented path', p['path']), ('Detected', _now_utc())],
|
||
summary=f'This has not matched the cutsheet for at least {hours:g} hours.',
|
||
checks=['If the cable was moved on purpose, update its record in the cutsheet.',
|
||
'Otherwise, follow the label and check it is plugged into the documented port '
|
||
'(Trace mode on the cutsheet page can find which port a cord is in).'],
|
||
priority='4', port=(p['switch'], p['port']),
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Ping-only hosts (no node_exporter)
|
||
# ------------------------------------------------------------------
|
||
def _process_ping_hosts(self, suppressions: list, ping_states: Dict[str, Optional[bool]]) -> None:
|
||
for h in self.cfg.get('monitor', {}).get('ping_hosts', []):
|
||
name, ip = h['name'], h['ip']
|
||
reachable = ping_states.get(name)
|
||
|
||
if reachable is None:
|
||
# Pulse couldn't run the check. Neither alert nor resolve on no evidence.
|
||
logger.warning(f'Ping check for {name} ({ip}) unavailable this cycle')
|
||
continue
|
||
|
||
if not reachable:
|
||
sup = db.check_suppressed(suppressions, 'host', name)
|
||
event_id, is_new, consec, ticket_id = db.upsert_event(
|
||
'host_unreachable', 'critical', 'ping',
|
||
name, ip,
|
||
f'Host {name} ({ip}) unreachable via ping ({_now_utc()})',
|
||
)
|
||
if not sup and consec >= self.host_fail_thresh and not ticket_id:
|
||
self._ticket_unreachable(event_id, name, ip, consec)
|
||
else:
|
||
db.resolve_event('host_unreachable', name, ip)
|
||
|
||
def _ticket_unreachable(
|
||
self, event_id: int, name: str, ip: str, consec: int
|
||
) -> None:
|
||
self._open_ticket(
|
||
event_id, 'host_unreachable', name, ip,
|
||
title=f'[{name}][auto][production][issue][network][single-node] Host unreachable',
|
||
heading='Host Reachability Alert',
|
||
facts=[('Host', name), ('IP', ip), ('Detected', _now_utc()),
|
||
('Consecutive check failures', str(consec)),
|
||
('Down for at least', ticket_text.duration(consec * self._poll_seconds()))],
|
||
summary=(f'{name} ({ip}) is not answering ping from the Pulse worker Gandalf uses. '
|
||
f'It has no node_exporter, so ping is the only check Gandalf has for it.'),
|
||
checks=[f'Is {name} powered on? Check its console or management interface.',
|
||
'Check its switch port below for link and PoE.',
|
||
'If the port is up, check the IP and firewall on the device itself.'],
|
||
priority='2', host=name,
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# WAN modem monitoring
|
||
# ------------------------------------------------------------------
|
||
_MODEM_METRIC_LABELS = {
|
||
'rx_power': 'Rx power', 'tx_power': 'Tx power', 'temperature': 'Temperature',
|
||
'vcc': 'Supply voltage', 'tx_bias': 'Tx bias',
|
||
}
|
||
_MODEM_METRIC_UNITS = {'rx_power': ' dBm', 'tx_power': ' dBm', 'temperature': ' C'}
|
||
|
||
def _modem_event(
|
||
self, suppressions: list, event_type: str, severity: str, detail: str,
|
||
description: str, heading: str, priority: str, advice: str,
|
||
) -> None:
|
||
name = self.modem_name
|
||
sup = db.check_suppressed(suppressions, 'host', name)
|
||
event_id, is_new, consec, ticket_id = db.upsert_event(
|
||
event_type, severity, 'modem', name, detail, f'{description} ({_now_utc()})',
|
||
)
|
||
if not sup and consec >= self.modem_fail_thresh and not ticket_id:
|
||
self._open_ticket(
|
||
event_id, event_type, name, detail,
|
||
title=f'[{name}][auto][production][issue][network][cluster-wide] {heading}',
|
||
heading='WAN Modem Alert',
|
||
facts=[('Modem', f'{name} ({self.modem.base_url})'), ('Detected', _now_utc()),
|
||
('Consecutive check failures', str(consec))],
|
||
summary=description, checks=[advice], priority=priority,
|
||
)
|
||
|
||
def _process_modem_failure(self, suppressions: list, error: Exception) -> None:
|
||
"""Handle a failed poll. Link and optical events stay as they are: we can't see them."""
|
||
name = self.modem_name
|
||
if isinstance(error, modem_api.ModemUnreachable):
|
||
db.resolve_event('modem_data_invalid', name, '')
|
||
self._modem_event(
|
||
suppressions, 'modem_unreachable', 'critical', '',
|
||
f'WAN modem is not responding: {error}',
|
||
'WAN modem unreachable', '2',
|
||
'Internet access is probably affected too. Check modem power and the '
|
||
'cable between the modem and the UDM-Pro.',
|
||
)
|
||
else:
|
||
db.resolve_event('modem_unreachable', name, '')
|
||
self._modem_event(
|
||
suppressions, 'modem_data_invalid', 'warning', '',
|
||
f'WAN modem status could not be read: {error}',
|
||
'WAN modem status unreadable', '3',
|
||
'The modem answered but its pages could not be parsed, so fiber alerting is '
|
||
'paused. A firmware update may have changed them; see modem.py.',
|
||
)
|
||
|
||
def _process_modem(self, data: dict, suppressions: list) -> None:
|
||
"""Raise and resolve alerts from a successful modem poll."""
|
||
name = self.modem_name
|
||
db.resolve_event('modem_unreachable', name, '')
|
||
db.resolve_event('modem_data_invalid', name, '')
|
||
|
||
problems = modem_api.wan_problems(data)
|
||
if problems:
|
||
self._modem_event(
|
||
suppressions, 'wan_down', 'critical', '',
|
||
f'WAN link is down: {"; ".join(problems)}.',
|
||
'WAN link down', '1',
|
||
'Check the fiber at the modem and the AT&T service status. Rx power at or below '
|
||
'the alarm limit points to a fiber fault rather than a provider outage.',
|
||
)
|
||
else:
|
||
db.resolve_event('wan_down', name, '')
|
||
|
||
for key, metric in data.get('optics', {}).items():
|
||
status = modem_api.metric_status(metric)
|
||
if status == 'alarm' and problems and key == 'rx_power':
|
||
# wan_down already covers this, and a cut fiber would otherwise open two tickets.
|
||
status = 'ok'
|
||
label = self._MODEM_METRIC_LABELS.get(key, key)
|
||
value = metric.get('value')
|
||
reading = f' at {value}{self._MODEM_METRIC_UNITS.get(key, "")}' if value is not None else ''
|
||
limits = (
|
||
f' (warning {metric.get("warn_low")} to {metric.get("warn_high")}, '
|
||
f'alarm {metric.get("alarm_low")} to {metric.get("alarm_high")})'
|
||
) if value is not None else ''
|
||
advice = (
|
||
'Inspect and clean the fiber connector at the modem and look for tight bends. '
|
||
'If that does not help, contact AT&T: the reading is what the modem measures '
|
||
'on the provider side of the connection.'
|
||
)
|
||
|
||
if status == 'alarm':
|
||
db.resolve_event('fiber_signal_warning', name, key)
|
||
self._modem_event(
|
||
suppressions, 'fiber_signal_alarm', 'critical', key,
|
||
f'{label} is in the alarm range{reading}{limits}.',
|
||
f'Fiber {label} alarm', '2', advice,
|
||
)
|
||
elif status == 'warning':
|
||
db.resolve_event('fiber_signal_alarm', name, key)
|
||
self._modem_event(
|
||
suppressions, 'fiber_signal_warning', 'warning', key,
|
||
f'{label} is in the warning range{reading}{limits}.',
|
||
f'Fiber {label} warning', '3', advice,
|
||
)
|
||
else:
|
||
db.resolve_event('fiber_signal_alarm', name, key)
|
||
db.resolve_event('fiber_signal_warning', name, key)
|
||
|
||
self._check_rx_drift(data, suppressions)
|
||
|
||
def _check_rx_drift(self, data: dict, suppressions: list) -> None:
|
||
"""Alert on a sustained Rx drop relative to this link's own recent normal.
|
||
|
||
The modem's limits are very wide (warning at -27.9 dBm on a link that sits
|
||
around -17), so a damaged connector could lose many dB before they trip.
|
||
"""
|
||
name = self.modem_name
|
||
rx = (data.get('optics', {}).get('rx_power') or {}).get('value')
|
||
if rx is None:
|
||
return
|
||
history = db.get_modem_rx_values(days=7)
|
||
if len(history) < self.rx_drift_min_samples:
|
||
return # not enough history yet to know what normal looks like
|
||
baseline = statistics.median(history)
|
||
drop = round(baseline - rx, 1)
|
||
if drop >= self.rx_drift_db:
|
||
self._modem_event(
|
||
suppressions, 'fiber_rx_drift', 'warning', 'rx_power',
|
||
f'Rx power is {rx} dBm, {drop} dB below its 7-day median of {baseline:.1f} dBm.',
|
||
'Fiber Rx power dropped', '3',
|
||
'The signal is still in range but has changed. Look for a recently disturbed '
|
||
'or bent fiber and check the connector at the modem.',
|
||
)
|
||
else:
|
||
db.resolve_event('fiber_rx_drift', name, 'rx_power')
|
||
|
||
def _load_wan_history(self) -> dict:
|
||
raw = db.get_state('wan_history')
|
||
if raw:
|
||
try:
|
||
return json.loads(raw)
|
||
except ValueError:
|
||
logger.error('Failed to parse wan_history JSON; starting over')
|
||
return {'outages': [], 'restarts': []}
|
||
|
||
def _note_modem_restart(self, data: dict) -> None:
|
||
uptime = data.get('uptime_s')
|
||
if uptime is None:
|
||
return
|
||
previous = None
|
||
raw = db.get_state('modem_stats')
|
||
if raw:
|
||
try:
|
||
previous = json.loads(raw).get('uptime_s')
|
||
except ValueError:
|
||
logger.error('Failed to parse modem_stats JSON')
|
||
if previous is not None and uptime < previous:
|
||
logger.warning(f'Modem restarted: uptime went from {previous}s to {uptime}s')
|
||
history = self._load_wan_history()
|
||
modem_api.record_uptime_reset(history, time.time() - uptime)
|
||
db.set_state('wan_history', history)
|
||
|
||
def _one_shot_ticket(
|
||
self, suppressions: list, event_type: str, description: str,
|
||
heading: str, priority: str, advice: str,
|
||
) -> None:
|
||
"""Report something that has already happened: open a ticket and resolve the event."""
|
||
name = self.modem_name
|
||
event_id, _, _, ticket_id = db.upsert_event(event_type, 'warning', 'unifi', name, '', description)
|
||
if not db.check_suppressed(suppressions, 'host', name) and not ticket_id:
|
||
tid = self.tickets.create(
|
||
f'[{name}][auto][production][issue][network][cluster-wide] {heading}',
|
||
ticket_text.description('WAN Event', [('Modem', name), ('Reported', _now_utc())],
|
||
description, [advice], recovers=False),
|
||
priority=priority,
|
||
)
|
||
if tid:
|
||
db.set_ticket_id(event_id, tid)
|
||
db.resolve_event(event_type, name, '')
|
||
|
||
def _process_wan_log(self, suppressions: list) -> None:
|
||
"""Read WAN outages and modem restarts from the UniFi log.
|
||
|
||
These can be over in two minutes, so a 5 minute poll of the modem may never see them.
|
||
"""
|
||
now = time.time()
|
||
raw_cursor = db.get_state('unifi_log_cursor')
|
||
if raw_cursor is None:
|
||
# First run: start from now rather than ticketing old history.
|
||
db.set_state('unifi_log_cursor', str(int(now * 1000)))
|
||
return
|
||
# After long monitor downtime, don't open tickets for stale events.
|
||
since_ms = max(int(raw_cursor) + 1, int((now - 86400) * 1000))
|
||
events = self.unifi.get_system_events(since_ms, modem_api.WAN_LOG_KEYS)
|
||
if not events:
|
||
return # None = log unavailable, so the cursor stays put and it's retried
|
||
|
||
history = self._load_wan_history()
|
||
actionable = modem_api.apply_wan_events(history, events)
|
||
|
||
# UniFi repeats "failed multiple times" per failure, so ticket it once a day at most.
|
||
unstable = [e for e in actionable if e['key'] == modem_api.LOG_WAN_UNSTABLE]
|
||
ticket_unstable = bool(unstable) and now - history.get('unstable_ticketed_at', 0) >= 86400
|
||
if ticket_unstable:
|
||
history['unstable_ticketed_at'] = now
|
||
|
||
# Save first: a failed ticket call shouldn't replay these events next cycle.
|
||
db.set_state('wan_history', history)
|
||
db.set_state('unifi_log_cursor', str(int(max(e['ts'] for e in events) * 1000)))
|
||
|
||
for e in actionable:
|
||
if e['key'] == modem_api.LOG_MODEM_RESTARTED:
|
||
when = datetime.fromtimestamp(e['ts'], timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')
|
||
self._one_shot_ticket(
|
||
suppressions, 'modem_power_cycled',
|
||
f'{e.get("device") or "Smart Power"} power-cycled the modem at {when} '
|
||
f'to try to restore the internet connection.',
|
||
'Modem auto power-cycled after internet loss', '3',
|
||
'The internet connection failed badly enough to trigger auto power-cycle. '
|
||
'Check the WAN / Fiber panel for outages around this time and Rx power for a drop.',
|
||
)
|
||
if ticket_unstable:
|
||
self._one_shot_ticket(
|
||
suppressions, 'wan_unstable',
|
||
'UniFi reports the WAN connection went down multiple times in the last 24 hours.',
|
||
'WAN connection unstable', '2',
|
||
'See the WAN history on the modem page. Repeated drops with healthy Rx power '
|
||
'point toward the provider or the modem rather than the fiber.',
|
||
)
|
||
|
||
def _collect_modem(self, suppressions: list) -> None:
|
||
"""Poll the modem, then store its status and raise any alerts."""
|
||
try:
|
||
data = self.modem.poll()
|
||
except (modem_api.ModemUnreachable, modem_api.ModemDataInvalid) as e:
|
||
logger.warning(f'Modem poll failed: {e}')
|
||
self._process_modem_failure(suppressions, e)
|
||
db.set_state('modem_stats', {
|
||
'reachable': False, 'error': str(e), 'name': self.modem_name,
|
||
'lan_ip': self.modem_lan_ip, 'updated': _now_utc(),
|
||
})
|
||
return
|
||
|
||
# Uptime going backwards between polls means the gateway restarted, even if
|
||
# it came back too quickly for any poll to fail. Read before it's overwritten.
|
||
self._note_modem_restart(data)
|
||
|
||
now = time.monotonic()
|
||
data['rates'] = modem_api.counter_rates(
|
||
data['counters'], self._prev_modem_counters, now - self._prev_modem_time,
|
||
)
|
||
self._prev_modem_counters = data['counters']
|
||
self._prev_modem_time = now
|
||
data['reachable'] = True
|
||
data['name'] = self.modem_name
|
||
data['lan_ip'] = self.modem_lan_ip
|
||
data['wan_problems'] = modem_api.wan_problems(data)
|
||
data['optics_status'] = {k: modem_api.metric_status(m) for k, m in data['optics'].items()}
|
||
db.set_state('modem_stats', data)
|
||
|
||
# Alert before recording, so a sudden drop isn't compared against itself.
|
||
self._process_modem(data, suppressions)
|
||
|
||
optics = data['optics']
|
||
db.record_modem_sample(
|
||
(optics.get('rx_power') or {}).get('value'),
|
||
(optics.get('tx_power') or {}).get('value'),
|
||
(optics.get('temperature') or {}).get('value'),
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Snapshot collection (for dashboard)
|
||
# ------------------------------------------------------------------
|
||
def _collect_snapshot(
|
||
self, iface_states: Optional[Dict[str, Dict[str, bool]]],
|
||
unifi_devices: Optional[List[dict]] = None,
|
||
ping_states: Optional[Dict[str, Optional[bool]]] = None,
|
||
) -> dict:
|
||
# Accept pre-fetched devices; fall back to empty list if unavailable
|
||
display_unifi = unifi_devices if unifi_devices is not None else []
|
||
|
||
# During a Prometheus outage keep showing the hosts, with their state unknown,
|
||
# rather than emptying the dashboard or implying that they're down.
|
||
prom_ok = iface_states is not None
|
||
shown_states = iface_states if prom_ok else self._last_iface_states
|
||
|
||
hosts = {}
|
||
for instance, ifaces in shown_states.items():
|
||
if instance not in self._instance_map:
|
||
continue # skip Prometheus instances not in config (e.g. LXC app servers)
|
||
host = self._hostname(instance)
|
||
phys = {k: v for k, v in ifaces.items()}
|
||
up_count = sum(1 for v in phys.values() if v)
|
||
total = len(phys)
|
||
if not prom_ok:
|
||
status = 'unknown'
|
||
elif total == 0 or up_count == total:
|
||
status = 'up'
|
||
elif up_count == 0:
|
||
status = 'down'
|
||
else:
|
||
status = 'degraded'
|
||
|
||
hosts[host] = {
|
||
'ip': instance.split(':')[0],
|
||
'interfaces': {k: ('up' if v else 'down') for k, v in phys.items()},
|
||
'status': status,
|
||
'source': 'prometheus',
|
||
}
|
||
|
||
# Configured hosts with no data: down if Prometheus is fine, otherwise unknown
|
||
for instance, host in self._instance_map.items():
|
||
if instance not in shown_states:
|
||
hosts[host] = {
|
||
'ip': instance.split(':')[0],
|
||
'interfaces': {},
|
||
'status': 'down' if prom_ok else 'unknown',
|
||
'source': 'prometheus',
|
||
}
|
||
|
||
for h in self.cfg.get('monitor', {}).get('ping_hosts', []):
|
||
name, ip = h['name'], h['ip']
|
||
reachable = (ping_states or {}).get(name)
|
||
hosts[name] = {
|
||
'ip': ip,
|
||
'interfaces': {},
|
||
'status': 'unknown' if reachable is None else ('up' if reachable else 'down'),
|
||
'source': 'ping',
|
||
}
|
||
|
||
return {
|
||
'hosts': hosts,
|
||
'unifi': display_unifi,
|
||
'updated': datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z'),
|
||
}
|
||
|
||
# ------------------------------------------------------------------
|
||
# Main loop
|
||
# ------------------------------------------------------------------
|
||
def run(self) -> None:
|
||
logger.info(
|
||
f'Gandalf monitor started – poll_interval={self.poll_interval}s '
|
||
f'fail_thresh={self.fail_thresh} host_fail_thresh={self.host_fail_thresh} '
|
||
f'unifi_fail_thresh={self.unifi_fail_thresh}'
|
||
)
|
||
while True:
|
||
try:
|
||
logger.info('Starting network check cycle')
|
||
|
||
# 1. Fetch interface states once — shared by snapshot and alert processing
|
||
iface_states = self.prom.get_interface_states()
|
||
|
||
# 2. Fetch UniFi devices once — used by both snapshot and alert processing
|
||
unifi_devices = self.unifi.get_devices()
|
||
|
||
# 3a. Ping-only hosts once — shared by snapshot and alert processing
|
||
ping_states: Dict[str, Optional[bool]] = {
|
||
h['name']: self.pulse.ping(h['ip'])
|
||
for h in self.cfg.get('monitor', {}).get('ping_hosts', [])
|
||
}
|
||
|
||
# 3b. Collect and store snapshot for dashboard
|
||
snapshot = self._collect_snapshot(iface_states, unifi_devices, ping_states)
|
||
if iface_states is not None:
|
||
self._last_iface_states = iface_states
|
||
db.set_state('network_snapshot', snapshot)
|
||
db.set_state('last_check', _now_utc())
|
||
|
||
# 4. Collect link stats (ethtool + traffic metrics)
|
||
link_data = None
|
||
try:
|
||
link_data = self.link_stats.collect(self._instance_map)
|
||
db.set_state('link_stats', link_data)
|
||
except Exception as e:
|
||
logger.error(f'Link stats collection failed: {e}', exc_info=True)
|
||
if link_data is not None:
|
||
self._link_data = link_data
|
||
try:
|
||
self._remember_uplinks(link_data)
|
||
except Exception as e:
|
||
logger.error(f'Recording known uplinks failed: {e}', exc_info=True)
|
||
|
||
# 5. Process alerts using already-fetched data
|
||
suppressions = db.get_active_suppressions()
|
||
if iface_states is None:
|
||
self._process_prometheus_outage(suppressions)
|
||
else:
|
||
db.resolve_event('monitor_source_unavailable', 'prometheus', '')
|
||
self._process_interfaces(iface_states, suppressions)
|
||
self._process_unifi(unifi_devices, suppressions)
|
||
|
||
self._process_ping_hosts(suppressions, ping_states)
|
||
|
||
# Cutsheet: compare documented cabling with what was just discovered
|
||
if link_data is not None:
|
||
try:
|
||
self._process_cutsheet(link_data, suppressions)
|
||
except Exception as e:
|
||
logger.error(f'Cutsheet check failed: {e}', exc_info=True)
|
||
|
||
# WAN modem: isolated so a failure here can't stop the rest of the cycle
|
||
if self.modem:
|
||
try:
|
||
# Log first, so a restart it reports can explain an uptime reset below
|
||
self._process_wan_log(suppressions)
|
||
self._collect_modem(suppressions)
|
||
except Exception as e:
|
||
logger.error(f'Modem collection failed: {e}', exc_info=True)
|
||
|
||
# Note on the tickets of anything that cleared this cycle (or failed to post before)
|
||
try:
|
||
self._post_recovery_notes()
|
||
except Exception as e:
|
||
logger.error(f'Posting recovery notes failed: {e}', exc_info=True)
|
||
|
||
# Housekeeping: deactivate expired suppressions and purge old resolved events
|
||
db.cleanup_expired_suppressions()
|
||
db.purge_old_resolved_events(days=90)
|
||
if self.modem:
|
||
db.purge_old_modem_samples(days=90)
|
||
|
||
logger.info('Network check cycle complete')
|
||
|
||
except Exception as e:
|
||
logger.error(f'Monitor loop error: {e}', exc_info=True)
|
||
time.sleep(30)
|
||
continue
|
||
|
||
time.sleep(self.poll_interval)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
monitor = NetworkMonitor()
|
||
monitor.run()
|