#!/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' 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()) 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 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 = int(cfg.get('history_days') or 7) 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') 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) VALUES " "(%d,'%s','%s',FROM_BASE64('%s'));\n" "DELETE FROM l2_topology_history WHERE ts < %d;\n" % (now, cap, st, b64, 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() # ---------- port status maps ---------- pmap = {} # (ip, ifName) -> row for r in ps: pmap[(r['switch_ip'], r['ifName'])] = r # ---------- 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}) # ---------- terminals (with history for offline detection) ---------- prev_term = {} try: prev = json.load(open(OUT, encoding='utf-8')) for t in prev.get('terminals', []): prev_term[t['mac']] = t except Exception: pass seen_mac = set(); terminals = [] for t in eps: mac = (t['mac'] or '').lower() if not mac: continue seen_mac.add(mac) 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 online = (oper == 'up' and mc > 0) 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': online, 'last_seen': now if online else (prev_term.get(mac, {}).get('last_seen') or None)}) # carry previously-known but now-absent terminals as offline for mac, pt in prev_term.items(): if mac not in seen_mac: nt = dict(pt); nt['online'] = False terminals.append(nt) # terminal counts per switch tcnt = collections.defaultdict(lambda: {'online': 0, 'offline': 0}) for t in terminals: c = tcnt[t['sw_ip']] c['online' if t['online'] else 'offline'] += 1 # ---------- nodes ---------- nodes = [] for ip in sorted(nodes_ip, key=lambda x: (depth.get(x, 99), x)): al = ip_alarms.get(ip, []) maxsev = max([a['sev'] for a in al], default=-1) nodes.append({ 'id': ip, 'name': sname.get(ip, ip), 'ip': ip, 'tier': tier(ip), 'depth': depth.get(ip, 99), 'hostid': hid_by_ip.get(ip), 'oper': 'down' if avail.get(ip) == '2' else 'up', 'alarms': sorted(al, key=lambda a: -a['sev']), 'maxsev': maxsev, 'term_online': tcnt[ip]['online'], 'term_offline': tcnt[ip]['offline'], 'latency_ms': lat.get(ip), 'children': len(children.get(ip, [])) }) # ---------- edges ---------- edges = [] 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' edges.append({'source': a, 'target': b, 'speed': speed, 'state': state, 'a_port': a_port, 'b_port': b_port}) stats = { 'switches': len(nodes), 'edges': len(edges), 'term_total': len(terminals), 'term_online': sum(1 for t in terminals if t['online']), 'term_offline': sum(1 for t in terminals if not t['online']), '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(nodes_ip) 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()