zabbix-deploy/collector/l2_snapshot.py

#!/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()