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'
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()