#!/usr/bin/env python3
# VICIfast Research, Report 01: The Two Bottlenecks Inside a Busy VICIdial Server
# Load rig (§7.2). FOR A THROWAWAY TEST SERVER, NOT A LIVE FLOOR. Read it before you run it.
# One line of key=value every 5 s to /root/vf-lt/sample.log: CPU by process group, VICIdial daemon CPU, Asterisk threads,
# the AMI router queue, AMI listen overflows, channels, and VICIdial's live-call counts. Usage: rig-sampler.py <aio|dialer|sink|agents>.
# Changes the server? No (reads only; writes its own log).
# AMI-wall rig 2026-10-23 sampler (from the 10-15 rig; role aio = dialer + web on one box). One line of key=value every 5 s to /root/vf-lt/sample.log
# until /root/vf-lt/stop-sample exists.  Usage: sample3.py <aio|web|dialer|sink|agents>
# Every 60 s dialers and the sink also dump `pjsip show channelstats` and
# `core show taskprocessors` raw to /root/vf-lt/raw/ for offline parsing.
import os, re, subprocess, sys, time

ROLE = sys.argv[1]
D = '/root/vf-lt'
os.makedirs(D + '/raw', exist_ok=True)
out = open(D + '/sample.log', 'a', buffering=1)
HZ = os.sysconf('SC_CLK_TCK')
NCPU = os.cpu_count()
GROUPS = [
    ('ast', re.compile(r'^asterisk$')),
    ('db', re.compile(r'^(mariadbd|mysqld)$')),
    ('web', re.compile(r'^(apache2|httpd|php-fpm.*|php)$')),
    ('vd', re.compile(r'^(AST_|ADMIN_|FastAGI|agi-|VD_|perl)')),
    ('chrome', re.compile(r'chrom|headless')),
    ('node', re.compile(r'^node$')),
    ('rec', re.compile(r'^(lame|sox|soxmix|ftp|curl)$')),
]


def sh(cmd, t=8):
    try:
        return subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=t).stdout
    except Exception:
        return ''


def cpu():
    f = open('/proc/stat').readline().split()[1:9]
    return [int(x) for x in f]  # user nice system idle iowait irq softirq steal


def procs():
    g = {}
    for pid in os.listdir('/proc'):
        if not pid.isdigit():
            continue
        try:
            s = open(f'/proc/{pid}/stat').read()
        except Exception:
            continue
        r = s.rfind(')')
        comm = s[s.find('(') + 1:r]
        f = s[r + 2:].split()
        key = 'other'
        for name, rx in GROUPS:
            if rx.search(comm):
                key = name
                break
        g[pid] = (key, int(f[11]) + int(f[12]))
    return g


def net():
    tot = [0, 0, 0, 0]
    priv = [0, 0]
    for l in open('/proc/net/dev').readlines()[2:]:
        name, rest = l.split(':', 1)
        name = name.strip()
        if name == 'lo':
            continue
        v = rest.split()
        rb, rp, tb, tp = int(v[0]), int(v[1]), int(v[8]), int(v[9])
        if '4000' in name or name.startswith('vlan'):
            priv[0] += rp
            priv[1] += tp
        else:
            tot[0] += rb; tot[1] += rp; tot[2] += tb; tot[3] += tp
    return tot, priv


def listen_ovf():
    """TcpExt ListenOverflows (cumulative) and the AMI accept queue (Recv-Q on :5038)."""
    ov = 0
    try:
        L = [l.split() for l in open('/proc/net/netstat') if l.startswith('TcpExt:')]
        ov = int(dict(zip(L[0][1:], L[1][1:]))['ListenOverflows'])
    except Exception:
        pass
    q = sh("ss -ltnH 'sport = :5038'").split()
    return ov, (q[1] if len(q) > 1 else '-')


def udp():
    lines = [l.split() for l in open('/proc/net/snmp') if l.startswith('Udp:')]
    return dict(zip(lines[0][1:], map(int, lines[1][1:])))


def ast_threads():
    """tid -> utime+stime ticks for every Asterisk thread."""
    try:
        pid = open('/var/run/asterisk/asterisk.pid').read().strip()
    except Exception:
        return {}
    out = {}
    try:
        tids = os.listdir(f'/proc/{pid}/task')
    except Exception:
        return {}
    for tid in tids:
        try:
            s = open(f'/proc/{pid}/task/{tid}/stat').read()
        except Exception:
            continue
        f = s[s.rfind(')') + 2:].split()
        out[tid] = int(f[11]) + int(f[12])
    return out


def thread_map():
    """tid -> the function that started it (core show threads)."""
    m = {}
    for l in sh("asterisk -rx 'core show threads'", 15).splitlines():
        v = l.split()
        if len(v) >= 3 and v[1].isdigit():
            m[v[1]] = v[2]
    return m


DAEMONS = {'ami2': 'AST_update_AMI2', 'listen': 'AST_manager_listen', 'send': 'AST_manager_send', 'autodial': 'AST_VDauto_dial'}


def daemons():
    """name -> total ticks of the VICIdial daemon processes."""
    out = {k: 0 for k in DAEMONS}
    for pid in os.listdir('/proc'):
        if not pid.isdigit():
            continue
        try:
            cmd = open(f'/proc/{pid}/cmdline').read()
            if 'astguiclient' not in cmd:
                continue
            s = open(f'/proc/{pid}/stat').read()
        except Exception:
            continue
        f = s[s.rfind(')') + 2:].split()
        for k, name in DAEMONS.items():
            if name in cmd and 'SCREEN' not in cmd:
                out[k] += int(f[11]) + int(f[12])
    return out


def web_sql():
    db = sh("awk -F'=> *' '/^VARDB_database/{print $2}' /etc/astguiclient.conf").strip()
    q = (
        "select concat('ac_',replace(server_ip,'.','_'),'_',status),count(*) from vicidial_auto_calls group by server_ip,status;"
        "select concat('la_',status),count(*) from vicidial_live_agents group by status;"
        "select 'hopper',count(*) from vicidial_hopper;"
        "select concat('su_',replace(server_ip,'.','_')),unix_timestamp(now())-unix_timestamp(last_update) from server_updater;"
        "select 'maxq',ifnull(max(time),0) from information_schema.processlist where command='Query';"
        "show global status where Variable_name in ('Questions','Com_select','Com_insert','Com_update','Com_delete',"
        "'Threads_running','Threads_connected','Table_locks_waited','Table_locks_immediate','Slow_queries',"
        "'Created_tmp_disk_tables','Innodb_row_lock_waits','Max_used_connections');"
    )
    o = sh(f"mysql {db} -N -B -e \"{q}\"")
    return ' '.join(f'{a}={b}' for a, b in (l.split('\t', 1) for l in o.strip().split('\n') if '\t' in l))


p_cpu, p_pr, (p_net, p_priv), p_udp = cpu(), procs(), net(), udp()
p_t = time.time()
tick = 0
p_thr, tmap, p_dmn = ast_threads(), thread_map(), daemons()
while not os.path.exists(D + '/stop-sample'):
    time.sleep(max(0.0, 5 - (time.time() - p_t)))
    t = time.time()
    dt = t - p_t
    c, pr, (n, pv), u = cpu(), procs(), net(), udp()
    d = [a - b for a, b in zip(c, p_cpu)]
    tot = sum(d) or 1
    busy = 100 * (tot - d[3] - d[4]) / tot
    grp = {}
    for pid, (key, tm) in pr.items():
        prev = p_pr.get(pid)
        delta = tm - (prev[1] if prev and prev[0] == key else 0)
        grp[key] = grp.get(key, 0) + delta
    cores = {k: v / HZ / dt for k, v in grp.items()}
    load1 = open('/proc/loadavg').read().split()[0]
    mi = dict(l.split(':', 1) for l in open('/proc/meminfo'))
    memused = (int(mi['MemTotal'].split()[0]) - int(mi['MemAvailable'].split()[0])) // 1024
    f = [
        f't={t:.3f}', f'role={ROLE}', f'cpu={busy:.1f}', f'iowait={100*d[4]/tot:.1f}',
        f'softirq={100*d[6]/tot:.1f}', f'steal={100*d[7]/tot:.1f}', f'load={load1}', f'memMB={memused}',
        f'rxpps={(n[1]-p_net[1])/dt:.0f}', f'txpps={(n[3]-p_net[3])/dt:.0f}',
        f'rxmbps={(n[0]-p_net[0])*8/dt/1e6:.2f}', f'txmbps={(n[2]-p_net[2])*8/dt/1e6:.2f}',
        f'privrxpps={(pv[0]-p_priv[0])/dt:.0f}', f'privtxpps={(pv[1]-p_priv[1])/dt:.0f}',
        f'udpinerr={u.get("InErrors",0)-p_udp.get("InErrors",0)}',
        f'udprcvbuf={u.get("RcvbufErrors",0)-p_udp.get("RcvbufErrors",0)}',
        f'udpnoport={u.get("NoPorts",0)-p_udp.get("NoPorts",0)}',
    ] + [f'c_{k}={v:.2f}' for k, v in sorted(cores.items())]
    if ROLE in ('aio', 'dialer'):
        ov, aq = listen_ovf()
        f += [f'listenovf={ov}', f'amiacceptq={aq}']
        mg = sh("asterisk -rx 'core show taskprocessors like stasis/m:manager'")
        for l in mg.splitlines():
            v = l.split()
            if len(v) >= 6 and v[0].startswith('stasis/m:manager') and v[1].isdigit():
                f += [f'mgrproc={v[1]}', f'mgrq={v[2]}', f'mgrmax={v[3]}', f'mgrhigh={v[5]}']
                break
        try:
            apid = open('/var/run/asterisk/asterisk.pid').read().strip()
            rss = [l for l in open(f'/proc/{apid}/status') if l.startswith('VmRSS')][0].split()[1]
            f.append(f'astrssMB={int(rss)//1024}')
        except Exception:
            pass
    if ROLE in ('aio', 'dialer', 'sink'):
        o = sh("asterisk -rx 'core show channels count'")
        m1 = re.search(r'(\d+) active channels?', o)
        m2 = re.search(r'(\d+) active calls?', o)
        f += [f'chans={m1.group(1) if m1 else -1}', f'calls={m2.group(1) if m2 else -1}']
        thr = ast_threads()
        if tick % 6 == 0:
            tmap = thread_map()
        deltas = {tid: (v - p_thr.get(tid, v)) / HZ / dt * 100 for tid, v in thr.items()}
        if deltas:
            top = max(deltas, key=deltas.get)
            f += [f'thrmax={deltas[top]:.0f}', f'thrmaxfn={tmap.get(top, "?")[:30]}', f'thrmaxtid={top}']
            byfn = {}
            for tid, v in deltas.items():
                fn = tmap.get(tid, 'other')
                byfn[fn] = byfn.get(fn, 0) + v
            f += [f'fn_{k[:24]}={v:.0f}' for k, v in sorted(byfn.items(), key=lambda x: -x[1])[:5]]
            tps = [v for tid, v in deltas.items() if tmap.get(tid) == 'default_tps_processing_function']
            if tps:
                f.append(f'tpsmax={max(tps):.0f}')
        p_thr = thr
        dm = daemons()
        f += [f'd_{k}={(dm[k] - p_dmn.get(k, dm[k])) / HZ / dt * 100:.0f}' for k in DAEMONS]
        p_dmn = dm
        if tick % 6 == 0:
            tp = sh("asterisk -rx 'core show taskprocessors'", 15)
            mq, mname = 0, ''
            for l in tp.splitlines():
                v = l.split()
                if len(v) >= 6 and v[2].isdigit() and int(v[2]) > mq:
                    mq, mname = int(v[2]), v[0]
            f += [f'tpq={mq}', f'tpqname={mname[:40]}']
        if tick % 12 == 0:
            ts = int(t)
            open(f'{D}/raw/chanstats-{ts}.txt', 'w').write(sh("asterisk -rx 'pjsip show channelstats'", 20))
            open(f'{D}/raw/taskproc-{ts}.txt', 'w').write(sh("asterisk -rx 'core show taskprocessors'", 20))
    if ROLE in ('aio', 'web'):
        f.append(web_sql())
    out.write(' '.join(f) + '\n')
    p_cpu, p_pr, p_net, p_priv, p_udp, p_t = c, pr, n, pv, u, t
    tick += 1