#!/usr/bin/env python3
"""Schreibt waehrend des Lasttests die Serverseite mit.

    beobachten.py <sekunden> [ausgabedatei]

Die Antwortzeiten allein sagen nicht, WARUM es langsam wird. Diese Datei
beantwortet das: waren die FPM-Arbeiter voll, hat Varnish nichts mehr
abgefangen, hat Galera gebremst.

Ausgabe ist JSON Lines - eine Zeile je Messpunkt.

Warum Python und nicht Shell: Die Abfragen enthalten einfache und doppelte
Anfuehrungszeichen und laufen durch ssh, also durch zwei Shell-Ebenen. In
bash war das nicht mehr lesbar und beim dritten Anlauf immer noch kaputt.
"""
import json
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from pathlib import Path

SCHLUESSEL = "/root/.ssh/riccardo-prod"
FRONTENDS = ["10.0.1.1", "10.0.1.2"]
DATENBANK = ["10.0.3.1", "10.0.3.2", "10.0.3.3"]
TAKT = 5

SSH = ["ssh", "-n", "-i", SCHLUESSEL, "-o", "StrictHostKeyChecking=no",
       "-o", "UserKnownHostsFile=/dev/null", "-o", "LogLevel=ERROR",
       "-o", "ConnectTimeout=4", "-o", "BatchMode=yes"]

FRONTEND_BEFEHL = r"""
arbeiter=$(pgrep -c -f 'php-fpm: pool riccardo' || echo 0)
max=$(grep -oP '^pm.max_children\s*=\s*\K\d+' /etc/php/8.5/fpm/pool.d/riccardo.conf 2>/dev/null || echo 0)
last=$(cut -d' ' -f1 /proc/loadavg)
kerne=$(nproc)
treffer=$(varnishstat -1 -f MAIN.cache_hit 2>/dev/null | awk '{print $2}')
fehl=$(varnishstat -1 -f MAIN.cache_miss 2>/dev/null | awk '{print $2}')
wartend=$(varnishstat -1 -f MAIN.busy_sleep 2>/dev/null | awk '{print $2}')
echo "${arbeiter:-0} ${max:-0} ${last:-0} ${treffer:-0} ${fehl:-0} ${wartend:-0} ${kerne:-1}"
"""

DB_BEFEHL = (
    "mariadb -N -B -e \"SHOW GLOBAL STATUS WHERE Variable_name IN "
    "('Threads_connected','wsrep_flow_control_paused',"
    "'wsrep_local_recv_queue_avg','Queries')\""
)


def frag(ip: str, befehl: str) -> str:
    try:
        return subprocess.run(SSH + [f"root@{ip}", befehl], capture_output=True,
                              text=True, timeout=TAKT - 1).stdout
    except subprocess.TimeoutExpired:
        return ""


def zahl(wert, art=float, vorgabe=0):
    try:
        return art(wert)
    except (TypeError, ValueError):
        return vorgabe


def frontend(ip: str, jetzt: str) -> dict:
    teile = frag(ip, FRONTEND_BEFEHL).split()
    if len(teile) < 7:
        return {"zeit": jetzt, "art": "frontend", "knoten": ip, "erreichbar": False}
    arbeiter, hoechstens = zahl(teile[0], int), zahl(teile[1], int)
    return {
        "zeit": jetzt, "art": "frontend", "knoten": ip, "erreichbar": True,
        "arbeiter": arbeiter,
        "arbeiterMax": hoechstens,
        "arbeiterProzent": round(100 * arbeiter / hoechstens) if hoechstens else None,
        "last": zahl(teile[2]),
        # Die nackte Lastzahl sagt ohne die Kernzahl nichts: 6,9 ist auf
        # zwei Kernen dreifach ueberlastet und auf acht Kernen Alltag.
        "kerne": zahl(teile[6], int, 1),
        "lastJeKern": round(zahl(teile[2]) / max(zahl(teile[6], int, 1), 1), 2),
        "varnishTreffer": zahl(teile[3], int),
        "varnishFehl": zahl(teile[4], int),
        "varnishWartend": zahl(teile[5], int),
    }


def datenbank(ip: str, jetzt: str) -> dict:
    werte = {}
    for zeile in frag(ip, DB_BEFEHL).splitlines():
        t = zeile.split()
        if len(t) == 2:
            werte[t[0]] = t[1]
    if not werte:
        return {"zeit": jetzt, "art": "datenbank", "knoten": ip, "erreichbar": False}
    return {
        "zeit": jetzt, "art": "datenbank", "knoten": ip, "erreichbar": True,
        "verbindungen": zahl(werte.get("Threads_connected"), int),
        "bremst": zahl(werte.get("wsrep_flow_control_paused")),
        "warteschlange": zahl(werte.get("wsrep_local_recv_queue_avg")),
        "abfragen": zahl(werte.get("Queries"), int),
    }


def main() -> int:
    dauer = int(sys.argv[1]) if len(sys.argv) > 1 else 600
    datei = Path(sys.argv[2]) if len(sys.argv) > 2 else Path(__file__).parent / "verlauf.jsonl"
    datei.write_text("")
    print(f"Messe {dauer} s im {TAKT}-Sekunden-Takt nach {datei}", flush=True)

    ende = time.time() + dauer
    punkte = 0
    # Alle Knoten gleichzeitig fragen: nacheinander waeren fuenf
    # ssh-Anmeldungen je Messpunkt, und die dauern laenger als der Takt.
    with ThreadPoolExecutor(max_workers=8) as pool, datei.open("a") as f:
        while time.time() < ende:
            jetzt = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
            auftraege = [pool.submit(frontend, ip, jetzt) for ip in FRONTENDS]
            auftraege += [pool.submit(datenbank, ip, jetzt) for ip in DATENBANK]
            for auftrag in auftraege:
                f.write(json.dumps(auftrag.result(), ensure_ascii=False) + "\n")
                punkte += 1
            f.flush()
            time.sleep(TAKT)

    print(f"Fertig: {punkte} Messpunkte", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())