#!/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(f"Alert history for {device}", ticket_text.history_url(base, device))}') inventory = db.get_device_link(device) if inventory: lines.append(f'- {ticket_text.link(f"{device} in the inventory", inventory)}') 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()