#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Merge collectors' TSVs + Zabbix alarms -> snapshot.json for the L2 topology page.""" import csv, json, os, re, collections, time, datetime, urllib.request, gzip, base64, subprocess import l2cfg TOK = 'd3abb125eb301a021ef80a2847a803a7cd2d0b1d3c8337a14d09db9d571369b5' API = 'http://192.168.87.20:8080/api_jsonrpc.php' OUT = '/opt/zabbix/zabbix-server/web-modules/WecomCenter/assets/l2topology/snapshot.json' STP = '/tmp/stp_topology.tsv' EP = '/tmp/fdb_endpoints.tsv' PS = '/tmp/port_status.tsv' LAT = '/tmp/latency.tsv' POE = '/tmp/poe_switches.tsv' MYSQL_CONTAINER = os.environ.get('L2_MYSQL_CONTAINER', 'zabbix-mysql') MYSQL_USER = os.environ.get('L2_MYSQL_USER', 'root') MYSQL_PASS = os.environ.get('L2_MYSQL_PASS', 'CobcoIT132') MYSQL_DB = os.environ.get('L2_MYSQL_DB', 'zabbix') SEV = ['NC', 'INFO', 'WARN', 'AVG', 'HIGH', 'DISASTER'] # names localized on the frontend PHYS = {'6', '62', '117'} def api(m, p): d = json.dumps({"jsonrpc": "2.0", "method": m, "params": p, "id": 1}).encode() q = urllib.request.Request(API, data=d, headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + TOK}) return json.loads(urllib.request.urlopen(q, timeout=60).read()) def load(path): with open(path, encoding='utf-8') as f: return list(csv.DictReader(f, delimiter='\t')) def norm(s): return re.sub(r'\s+', ' ', (s or '').strip()) PHYS_TYPES = {'6', '62', '117'} IFACE_SPEED_RE = re.compile(r'^(?:(100GE|50GE|40GE|25GE|10GE|XGE|2\.5GE|GE|FE))', re.I) def port_max_mbps(name, current=0): """Best effort front-panel capability from the interface name. ``ifSpeed``/``ifHighSpeed`` is the negotiated link rate, not the port's maximum capability (a 10GE port with a 1G module reports 1000). Huawei interface names retain the hardware family, so use that prefix; fall back to the observed rate when the name is not descriptive. """ s = (name or '').strip().upper().replace(' ', '') m = IFACE_SPEED_RE.match(s) if m: fam = m.group(1).upper() return { '100GE': 100000, '50GE': 50000, '40GE': 40000, '25GE': 25000, '10GE': 10000, 'XGE': 10000, '2.5GE': 2500, 'GE': 1000, 'FE': 100 }.get(fam, current or 0) if s.startswith('XGIGABITETHERNET'): return 10000 if s.startswith('GIGABITETHERNET'): return 1000 if s.startswith('FASTETHERNET'): return 100 try: return int(current or 0) except Exception: return 0 def is_physical_port(row): itype = str(row.get('ifType') or '').strip() if itype not in PHYS_TYPES: return False name = (row.get('ifName') or row.get('ifDescr') or '').strip() if not name: return False if re.search(r'(?i)^(ETH-TRUNK|VLANIF|NULL|INLOOPBACK|STACK-PORT|LOOPBACK|TUNNEL|METH|VBDIF)', name): return False return True def load_latency(): lat = {} try: with open(LAT, encoding='utf-8') as f: for line in f: p = line.rstrip('\n').split('\t') if len(p) >= 2 and p[0] and p[1]: try: lat[p[0]] = float(p[1]) except ValueError: pass except Exception: pass return lat def load_poe(): out = {} try: for row in load(POE): ip = (row.get('switch_ip') or '').strip() if ip: out[ip] = (row.get('poe') or '0').strip() == '1' except Exception: pass return out def mysql(sql): return subprocess.run( ['docker', 'exec', '-i', MYSQL_CONTAINER, 'mysql', '-u' + MYSQL_USER, '-p' + MYSQL_PASS, MYSQL_DB, '--default-character-set=utf8mb4'], input=sql.encode('utf-8'), capture_output=True, timeout=120) def write_history(snap, now): cfg = l2cfg.config() days = max(1, min(30, int(cfg.get('history_days') or 7))) interval = max(5, int(cfg.get('history_interval_min') or 5)) interval_sec = interval * 60 raw = json.dumps(snap, ensure_ascii=False, separators=(',', ':')).encode('utf-8') b64 = base64.b64encode(gzip.compress(raw, 9)).decode('ascii') st = json.dumps(snap.get('stats', {}), ensure_ascii=False) cap = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') # One row per configured interval. The guard makes both run_full.sh and # refresh_alarms.sh safe to call this function: whichever runs first at a # boundary inserts, the other calls become no-ops. Old rows are always # pruned, so retention is a hard cap rather than a best effort. sql = ( "CREATE TABLE IF NOT EXISTS l2_topology_history (" "id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, ts BIGINT NOT NULL, " "captured_at DATETIME NOT NULL, stats TEXT, snapshot MEDIUMBLOB, " "PRIMARY KEY (id), KEY idx_l2_ts (ts)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;\n" "INSERT INTO l2_topology_history (ts,captured_at,stats,snapshot) " "SELECT %d,'%s','%s',FROM_BASE64('%s') FROM DUAL " "WHERE NOT EXISTS (SELECT 1 FROM l2_topology_history WHERE ts > %d);\n" "DELETE FROM l2_topology_history WHERE ts < %d;\n" % (now, cap, st, b64, now - interval_sec, now - days * 86400) ) r = mysql(sql) if r.returncode != 0: print('history write failed:', r.stderr.decode('utf-8', 'ignore')[:400]) def main(): now = int(time.time()) edges_raw = load(STP); eps = load(EP); ps = load(PS) lat = load_latency() poe = load_poe() # Previous snapshot is the switch inventory / last-known-position source. # Terminals are deliberately NOT compared with it: a terminal is shown # only if it is present in the current collection. prev_snap = {} try: with open(OUT, encoding='utf-8') as f: prev_snap = json.load(f) except Exception: pass prev_nodes = {str(n.get('id')): n for n in prev_snap.get('nodes', []) if n.get('id')} prev_edges = prev_snap.get('edges', []) or [] days = int(l2cfg.config().get('history_days') or 7) cutoff = now - days * 86400 # ---------- port status maps ---------- pmap = {} # (ip, ifName) -> row for r in ps: pmap[(r['switch_ip'], r['ifName'])] = r # Full physical front-panel inventory, including unused ports. This is # what the topology page uses to draw the switch port matrix when the # "show terminals" option is enabled. ports_by_ip = collections.defaultdict(list) for r in ps: if not is_physical_port(r): continue ip = (r.get('switch_ip') or '').strip() name = (r.get('ifName') or r.get('ifDescr') or '').strip() if not ip or not name: continue try: speed = int(float(r.get('speed_mbps') or 0)) except Exception: speed = 0 try: order = int(r.get('ifIndex') or 0) except Exception: order = 0 ports_by_ip[ip].append({ 'name': name, 'order': order, 'speed': speed, 'max': port_max_mbps(name, speed), 'up': (r.get('oper_status') == 'up'), }) for ip in ports_by_ip: # Keep one row per interface name; sort by ifIndex so the front-panel # order is deterministic across collection runs. uniq = {} for p in ports_by_ip[ip]: uniq[p['name']] = p ports_by_ip[ip] = sorted(uniq.values(), key=lambda p: (p['order'], p['name'])) # ---------- switch identity ---------- sname = {} for e in edges_raw: sname[e['down_ip']] = norm(e['down_switch']) sname[e['up_ip']] = norm(e['up_switch']) # ---------- undirected graph & depth ---------- adj = collections.defaultdict(set); pair = {} for e in edges_raw: a, b = e['down_ip'], e['up_ip'] if a == b: continue adj[a].add(b); adj[b].add(a) k = tuple(sorted([a, b])) pair.setdefault(k, {}).setdefault('ports', {}) pair[k]['ports'].setdefault(a, {})['down'] = e['down_port'] pair[k]['ports'].setdefault(b, {})['up'] = e['up_port'] downcnt = collections.Counter(e['down_ip'] for e in edges_raw) nodes_ip = set(adj) # components + true root (fewest times as downstream) -> depth, parent, children depth = {}; parent = {}; children = collections.defaultdict(list); roots = [] seen = set() for start in nodes_ip: if start in seen: continue comp = []; st = [start] while st: u = st.pop() if u in seen: continue seen.add(u); comp.append(u); st.extend(adj[u] - seen) root = min(comp, key=lambda n: (downcnt.get(n, 0), -len(adj[n]), n)) roots.append(root); depth[root] = 0; parent[root] = None order = [root]; qi = 0 while qi < len(order): u = order[qi]; qi += 1 for v in sorted(adj[u]): if v not in depth: depth[v] = depth[u] + 1; parent[v] = u; children[u].append(v); order.append(v) def tier(ip): d = depth.get(ip, 99) if d == 0: return 'core' if children.get(ip): return 'distribution' return 'access' # ---------- alarms ---------- problems = api('problem.get', {'output': ['eventid', 'objectid', 'name', 'severity', 'clock', 'acknowledged'], 'recent': False})['result'] trigids = sorted(set(p['objectid'] for p in problems)) trig = {} if trigids: tg = api('trigger.get', {'triggerids': trigids, 'output': ['triggerid'], 'selectHosts': ['hostid', 'name', 'interfaces']})['result'] for t in tg: trig[t['triggerid']] = t.get('hosts', []) ip_alarms = collections.defaultdict(list) # hosts & availability (need hostid -> ip for alarms) hs = api('host.get', {'output': ['hostid', 'name'], 'selectInterfaces': ['ip', 'type', 'available', 'error']})['result'] hid_ip = {}; ip_name = {}; hid_by_ip = {}; avail = {} for h in hs: cand = [i for i in h['interfaces'] if i.get('ip')] if not cand: continue snmp = [i for i in cand if i['type'] == '2'] pick = (snmp or cand)[0] hid_ip.setdefault(h['hostid'], pick['ip']) ip_name[pick['ip']] = h['name']; hid_by_ip[pick['ip']] = h['hostid'] for i in snmp: avail[i['ip']] = i.get('available', '0') for p in problems: for h in trig.get(p['objectid'], []): ip = hid_ip.get(h['hostid']) if not ip: continue ip_alarms[ip].append({'sev': int(p['severity']), 'name': p['name'], 'clock': int(p['clock']), 'ack': p['acknowledged'] == '1'}) for ip, av in avail.items(): if av == '2': ip_alarms[ip].append({'sev': 4, 'name': 'SNMP unreachable / host unavailable', 'clock': now, 'ack': False}) # ---------- switch inventory / hosts ---------- def is_switch_name(name): n = (name or '').upper() return bool(re.search(r'-(DIST|ACCESS|CORE)(-|$)', n)) or bool(re.match(r'^COBCO-', n)) switch_hosts = {} for h in hs: hn = norm(h.get('name') or '') if not is_switch_name(hn): continue for i in h.get('interfaces', []): if i.get('type') == '2' and i.get('ip'): switch_hosts[i['ip']] = { 'name': hn, 'hostid': h.get('hostid'), 'available': str(i.get('available', '0')), 'error': i.get('error', '') or '' } # ---------- terminals (current collection only; online only) ---------- # An FDB entry on a port that is not up is stale and is not an "offline # terminal": the product only displays terminals that are present now. terminals = [] for t in eps: mac = (t['mac'] or '').lower() if not mac: continue st = pmap.get((t['switch_ip'], t['port']), {}) oper = st.get('oper_status', '') try: mc = int(st.get('mac_count', '0') or 0) except: mc = 0 if oper != 'up' or mc <= 0: continue terminals.append({'mac': mac, 'ip': t['ip'], 'name': t['name'], 'sw_ip': t['switch_ip'], 'sw_name': norm(t['switch_name']), 'port': t['port'], 'alias': t.get('port_alias', ''), 'speed': st.get('speed_mbps', ''), 'online': True, 'last_seen': now}) # terminal counts per switch (online only) tcnt = collections.defaultdict(lambda: {'online': 0, 'offline': 0}) for t in terminals: tcnt[t['sw_ip']]['online'] += 1 # ---------- switch inventory: online topology + recorded offline ---------- offline_ips = set() # Hosts whose SNMP interface is unavailable are offline even if they have no # current STP edge. This is how a failed switch stays on the map. for ip, hn in switch_hosts.items(): if hn['available'] == '2': offline_ips.add(ip) # Keep a previously recorded switch for history_days when it disappears # from the protocol-derived graph. The last_online stamp prevents old # decommissioned devices from living forever. for ip, pn in prev_nodes.items(): if ip in nodes_ip: continue last_online = int(pn.get('last_online') or pn.get('last_seen') or prev_snap.get('generated') or 0) if last_online >= cutoff: offline_ips.add(ip) known_ips = set(nodes_ip) | offline_ips def name_role(name): n = (name or '').upper() if re.search(r'-(DIST)(-|$)', n): return 'distribution' if re.search(r'-(ACCESS)(-|$)', n): return 'access' if re.search(r'-(CORE)(-|$)', n): return 'core' return '' def node_tier(ip, name, pn=None): pn = pn or {} ch = int(pn.get('children') or 0) pt = str(pn.get('tier') or '').lower() # Graph degree is the protocol truth: a switch with children is an # aggregation/distribution point, regardless of a wrong name suffix. if ch > 0: return 'distribution' if pt in ('distribution', 'access'): return pt nr = name_role(name) if nr: return nr if pt == 'core': return 'core' return 'access' nodes = [] for ip in sorted(known_ips, key=lambda x: (depth.get(x, 99), x)): online = ip in nodes_ip pn = prev_nodes.get(ip, {}) hn = switch_hosts.get(ip, {}) nm = sname.get(ip) or pn.get('name') or hn.get('name') or ip_name.get(ip) or ip al = ip_alarms.get(ip, []) maxsev = max([a['sev'] for a in al], default=-1) if online: ntier = tier(ip) ndepth = depth.get(ip, 99) nchildren = len(children.get(ip, [])) oper = 'down' if avail.get(ip) == '2' else 'up' last_online = now else: ntier = node_tier(ip, nm, pn) ndepth = int(pn.get('depth', 99)) nchildren = int(pn.get('children') or 0) oper = 'down' last_online = int(pn.get('last_online') or pn.get('last_seen') or prev_snap.get('generated') or now) nodes.append({ 'id': ip, 'name': nm, 'ip': ip, 'tier': ntier, 'depth': ndepth, 'hostid': hid_by_ip.get(ip) or hn.get('hostid') or pn.get('hostid'), 'oper': oper, 'offline': not online, 'alarms': sorted(al, key=lambda a: -a['sev']), 'maxsev': maxsev, 'term_online': tcnt[ip]['online'], 'term_offline': 0, 'latency_ms': lat.get(ip) if online else pn.get('latency_ms'), 'children': nchildren, 'poe': bool(poe.get(ip, pn.get('poe', False))), 'ports': ports_by_ip.get(ip, []), 'last_online': last_online }) # ---------- edges ---------- edges = [] edge_pairs = set() for k, d in pair.items(): a, b = k p = d['ports'] pa = p.get(a, {}); pb = p.get(b, {}) a_port = pa.get('down') or pa.get('up') or '' b_port = pb.get('up') or pb.get('down') or '' speeds = [] for ip, port in ((a, a_port), (b, b_port)): row = pmap.get((ip, port), {}) try: sp = int(row.get('speed_mbps', '0') or 0) except: sp = 0 speeds.append(sp) speed = max(speeds) if speeds else 0 ops = [] for ip, port in ((a, a_port), (b, b_port)): row = pmap.get((ip, port)) ops.append(row.get('oper_status') if row else None) state = 'down' if ('down' in ops) else 'up' if a not in known_ips or b not in known_ips: continue edges.append({'source': a, 'target': b, 'speed': speed, 'state': state, 'a_port': a_port, 'b_port': b_port}) edge_pairs.add(tuple(sorted((a, b)))) # Keep the last-known uplink for an offline switch so it is shown in its # workshop frame and its historical position is not lost. for e in prev_edges: a, b = str(e.get('source')), str(e.get('target')) k = tuple(sorted((a, b))) if k in edge_pairs or a not in known_ips or b not in known_ips: continue if a in nodes_ip and b in nodes_ip: continue edges.append({'source': a, 'target': b, 'speed': e.get('speed', 0), 'state': 'down', 'a_port': e.get('a_port', ''), 'b_port': e.get('b_port', '')}) edge_pairs.add(k) stats = { 'switches': len(nodes), 'switches_online': len(nodes_ip), 'switches_offline': len(offline_ips), 'poe_switches': sum(1 for n in nodes if n.get('poe')), 'ports': sum(len(n.get('ports') or []) for n in nodes), 'edges': len(edges), 'term_total': len(terminals), 'term_online': len(terminals), 'term_offline': 0, 'alarms': sum(len(n['alarms']) for n in nodes), 'devices_alarm': sum(1 for n in nodes if n['alarms']), 'maxsev': max([n['maxsev'] for n in nodes], default=-1) } in_nodes = set(known_ips) other = [] for ip, al in ip_alarms.items(): if ip in in_nodes: continue for a in al: other.append({'ip': ip, 'host': ip_name.get(ip, ip), 'sev': a['sev'], 'name': a['name'], 'clock': a['clock']}) other.sort(key=lambda x: -x['sev']) stats['other_alarms'] = len(other) snap = {'generated': now, 'generated_iso': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'refresh_default': 300, 'roots': roots, 'stats': stats, 'other_alarms': other, 'nodes': nodes, 'edges': edges, 'terminals': terminals} os.makedirs(os.path.dirname(OUT), exist_ok=True) tmp = OUT + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(snap, f, ensure_ascii=False, separators=(',', ':')) os.replace(tmp, OUT) print('snapshot', OUT, os.path.getsize(OUT), 'bytes;', stats) if os.environ.get('L2_WRITE_HISTORY', '0') == '1': try: write_history(snap, now) except Exception as e: print('history write error:', e) if __name__ == '__main__': main()