From f8c5f2eb3ce2c57480f8aca5a056ab7253ec9b8b Mon Sep 17 00:00:00 2001 From: Kalina Date: Mon, 14 Sep 2026 17:31:14 +0200 Subject: [PATCH] feat(rawlogs): Rohlogs fuer den Loki-Import sammeln Der Prometheus-Archiv-Import verdichtet auf Stunde/Domain/Status und verwirft Referrer, User-Agent und Client-IP. Fragen nach der Herkunft liessen sich damit nicht rueckwirkend beantworten. - raw-log-collect.py: liest access+error (inkl. .gz, rotierte Dateien), schreibt ts_ns/labels/rohzeile und meldet Zeilen ohne Zeitstempel statt sie still zu verwerfen. Error-Logs ohne Zeitzone (Apache- Standard) werden korrekt nach UTC verschoben. - raw-log-backfill.sh: Wrapper mit Messmodus; sammelt auf dem Host, importiert selbst NICHTS. --- scripts/raw-log-backfill.sh | 140 +++++++++++++++++ scripts/raw-log-collect.py | 301 ++++++++++++++++++++++++++++++++++++ 2 files changed, 441 insertions(+) create mode 100644 scripts/raw-log-backfill.sh create mode 100644 scripts/raw-log-collect.py diff --git a/scripts/raw-log-backfill.sh b/scripts/raw-log-backfill.sh new file mode 100644 index 0000000..4d54c31 --- /dev/null +++ b/scripts/raw-log-backfill.sh @@ -0,0 +1,140 @@ +#!/bin/bash +# raw-log-backfill.sh — Rohlogs eines KeyHelp-Hosts fuer den LOKI-Import sammeln. +# +# Warum es dieses Skript gibt: +# Der Prometheus-Archiv-Import verdichtet jede Logzeile auf Stunde/Domain/ +# Status und VERWIRFT dabei Referrer, User-Agent und Client-IP. Fragen nach +# Herkunft ("woher kommen die Besucher") liessen sich damit nicht +# rueckwirkend beantworten. Lokis Rohzeilen koennen es — aber die +# Live-Sammlung reicht nur bis zur Installation zurueck. +# Dieses Skript holt die Historie in eine Datei, die Loki aufnehmen kann. +# +# DIESES SKRIPT AENDERT NICHTS AM LAUFENDEN DIENST. Es liest nur Logdateien, +# es stoppt nichts, es loescht nichts. +# +# AUFRUF (auf dem Host, als root): +# curl -fsSL https://gitea.m-m.bio/mm/netbird-telegraf-installer/raw/branch/main/scripts/raw-log-backfill.sh -o /root/raw-backfill.sh +# bash /root/raw-backfill.sh # MESSEN (aendert nichts) +# bash /root/raw-backfill.sh --collect # Rohdatei bauen +# bash /root/raw-backfill.sh --collect --since 2025-01-01 +# bash /root/raw-backfill.sh --collect --tarball /root/apache-archive/apache-archive.tar.gz +# +# Am Ende steht der exakte Befehl fuer den imac. + +set -eu + +REPO_RAW="${REPO_RAW:-https://gitea.m-m.bio/mm/netbird-telegraf-installer/raw/branch/main}" +OUT_DIR="${OUT_DIR:-/root/apache-archive}" +NODE="${NODE:-$(hostname -f 2>/dev/null || hostname)}" +COLLECT=0 +SINCE="" +UNTIL="" +TARBALL="" +ROOTS="/var/log/apache2 /home/users" + +while [ $# -gt 0 ]; do + case "$1" in + --collect) COLLECT=1 ;; + --since) SINCE="${2:?Datum fehlt}"; shift ;; + --until) UNTIL="${2:?Datum fehlt}"; shift ;; + --tarball) TARBALL="${2:?Pfad fehlt}"; shift ;; + --node) NODE="${2:?Name fehlt}"; shift ;; + --roots) ROOTS="${2:?Pfade fehlen}"; shift ;; + *) echo "Unbekannte Option: $1" >&2; exit 2 ;; + esac + shift +done + +log() { printf '\033[1;34m[raw-backfill]\033[0m %s\n' "$*"; } +warn() { printf '\033[1;33m[raw-backfill] WARNUNG:\033[0m %s\n' "$*" >&2; } +die() { printf '\033[1;31m[raw-backfill] FEHLER:\033[0m %s\n' "$*" >&2; exit 1; } + +[ "$(id -u)" -eq 0 ] || die "Bitte als root ausfuehren (die Logs liegen unter /home/users mit 0700)." + +mkdir -p "$OUT_DIR" +COLLECTOR="$OUT_DIR/raw-log-collect.py" +if [ -r "$COLLECTOR" ] && [ -n "${SKIP_FETCH:-}" ]; then + log "Collector lokal uebernommen: $COLLECTOR" +else + log "Lade Collector vom Repo …" + curl -fsSL -m 30 "$REPO_RAW/scripts/raw-log-collect.py" -o "$COLLECTOR" \ + || die "raw-log-collect.py konnte nicht geladen werden." +fi +python3 -c "import sys; sys.exit(0)" || die "python3 fehlt auf diesem Host." + +# -------------------------------------------------------------------------- +# 1) MESSEN — was ist ueberhaupt da? (aendert nichts) +# -------------------------------------------------------------------------- +log "Bestandsaufnahme der Apache-Logs (liest nur) …" +FILES=$(find $ROOTS -name 'access.log*' -o -name 'error.log*' 2>/dev/null | wc -l) +GZ=$(find $ROOTS -name '*.log*.gz' 2>/dev/null | wc -l) +RAW_BYTES=$(find $ROOTS -name 'access.log*' -o -name 'error.log*' 2>/dev/null -printf '%s\n' 2>/dev/null | awk '{s+=$1} END {print s+0}') +log " Logdateien: $FILES (davon komprimiert: $GZ)" +log " Rohgroesse: $(( RAW_BYTES / 1024 / 1024 )) MB" +log " Aelteste Datei: $(find $ROOTS -name 'access.log*' -printf '%TY-%Tm-%Td %p\n' 2>/dev/null | sort | head -1 | cut -c1-10)" +if [ -n "$TARBALL" ] && [ -r "$TARBALL" ]; then + log " Tarball: $TARBALL ($(( $(stat -c%s "$TARBALL") / 1024 / 1024 )) MB)" + log " Rohlogs im Tar: $(tar -tzf "$TARBALL" 2>/dev/null | grep -c 'raw/.*log' || echo 0) Dateien" +fi + +if [ "$COLLECT" -eq 0 ]; then + echo + log "Das war NUR die Messung — es wurde nichts geschrieben." + echo " Weiter mit: bash $0 --collect${SINCE:+ --since $SINCE}" + echo " Optional zusaetzlich Rohlogs aus dem Archiv-Tarball:" + echo " --tarball $OUT_DIR/apache-archive.tar.gz" + exit 0 +fi + +# -------------------------------------------------------------------------- +# 2) SAMMELN +# -------------------------------------------------------------------------- +ARGS="--node $NODE --out $OUT_DIR --roots $ROOTS" +[ -n "$SINCE" ] && ARGS="$ARGS --since $SINCE" +[ -n "$UNTIL" ] && ARGS="$ARGS --until $UNTIL" +[ -n "$TARBALL" ] && [ -r "$TARBALL" ] && ARGS="$ARGS --include-tarball $TARBALL" + +log "Sammle Rohlogs … (bei viel Historie dauert das einige Minuten)" +# shellcheck disable=SC2086 +python3 "$COLLECTOR" $ARGS 2>&1 | tee "$OUT_DIR/collect.log" +RC=${PIPESTATUS[0]} +[ "$RC" -eq 0 ] || die "Collector brach ab (Code $RC). Siehe $OUT_DIR/collect.log" + +PART=$(ls -t "$OUT_DIR"/loki-raw-*/raw.tsv 2>/dev/null | head -1) +STATE=$(ls -t "$OUT_DIR"/loki-raw-*/state.json 2>/dev/null | head -1) +[ -n "$PART" ] && [ -r "$PART" ] || die "Keine Rohdatei gefunden." +[ -n "$STATE" ] && [ -r "$STATE" ] || die "Keine Zustandsdatei gefunden." + +LINES=$(wc -l < "$PART") +SIZE=$(( $(stat -c%s "$PART") / 1024 / 1024 )) +log "Rohdatei: $PART" +log " $LINES Zeilen, $SIZE MB" + +# Gegenprobe: stimmt die Zeilenzahl mit der Zustandsdatei ueberein? +EXPECT=$(python3 -c "import json,sys; print(json.load(open('$STATE'))['lines'])") +if [ "$LINES" != "$EXPECT" ]; then + warn "Zeilenzahl weicht ab: Datei=$LINES, Auftrag=$EXPECT — bitte pruefen." +else + log " Gegenprobe: Zeilenzahl stimmt mit dem Auftrag ueberein ($LINES)." +fi + +echo +log "NAECHSTER SCHRITT — diese beiden Zeilen auf dem imac ausfuehren:" +echo +echo " # 1) Rohdatei und Auftrag uebertragen" +echo " scp $PART $STATE mm@100.83.134.46:/tmp/" +echo +echo " # 2) Einlesen (der Import laeuft LOKAL auf dem imac — kein Token noetig)" +echo " ssh mm@100.83.134.46 \\" +echo " 'python3 /usr/local/bin/loki-raw-import.py \\" +echo " --file /tmp/$(basename "$PART") --state /tmp/$(basename "$STATE") \\" +echo " --url http://127.0.0.1:3100'" +echo +echo " # Vorher pruefen (empfohlen, aendert nichts):" +echo " ssh mm@100.83.134.46 \\" +echo " 'python3 /usr/local/bin/loki-raw-import.py \\" +echo " --file /tmp/$(basename "$PART") --state /tmp/$(basename "$STATE") \\" +echo " --url http://127.0.0.1:3100 --dry-run'" +echo +log "Danach im Grafana pruefen: {imported=\"1\"} (Dashboard: Weblogs-Analyse)" +warn "Das Skript selbst schreibt NICHTS nach Loki — der Import ist ein eigener Schritt." diff --git a/scripts/raw-log-collect.py b/scripts/raw-log-collect.py new file mode 100644 index 0000000..c657e26 --- /dev/null +++ b/scripts/raw-log-collect.py @@ -0,0 +1,301 @@ +#!/usr/bin/env python3 +"""Rohlogs (Apache access+error) eines Hosts in eine importierbare Datei schreiben. + +Erzeugt eine DREISPALTIGE, nach Zeitstempel sortierbare Datei: + + + +Warum dieses Format: + * `ts_ns` in Spalte 1 -> GNU `sort -k1,1n` kann mit `-S` und `-T` EXTERN + sortieren. 40 Mio Zeilen (~12 GB) passen nicht in den RAM des imac; ein + Python-`sorted()` wuerde dort sterben. Externer Merge-Sort skaliert. + * Nach Zeitstempel aufsteigend sortiert ist PFLICHT, nicht Kosmetik: + Loki erlaubt kein Out-of-Order. Ein aelterer Eintrag nach einem neueren + wird mit "entry too far behind" abgelehnt (empirisch bestaetigt). + * Die Rohzeile bleibt UNVERAENDERT (nur die erste Tab-Trennung wird + geprueft). Das `message="..."`-Format, das Telegraf live schreibt, setzt + der Importer — die Kollektor-Ausgabe bleibt dadurch pruefbar roh. + +Dieses Skript aendert NICHTS am laufenden Dienst und loescht nichts. + +AUFRUF (auf dem Host, als root): + python3 raw-log-collect.py --node x14.idx.wf-keyhelp --out /root/apache-archive + python3 raw-log-collect.py --node x14 --out /root/apache-archive \ + --since 2025-01-01 --include-tarball /root/apache-archive/apache-archive.tar.gz +""" +from __future__ import annotations + +import argparse +import gzip +import json +import os +import re +import sys +import tarfile +import tempfile +from collections import Counter +from datetime import datetime, timezone + +# -------------------------------------------------------------------------- +# Zeitstempel-Formate in Apache-Logs. +# +# access (Combined/Common, KeyHelp-Standard): [14/Sep/2026:17:00:18 +0200] +# error (Apache-Standard OHNE Zeitzone): [Mon Sep 14 17:00:18.123456 2026] +# error (alternative Variante MIT Zeitzone): [14/Sep/2026:17:00:18 +0200] +# +# OHNE das reine Error-Format landen Error-Zeilen still auf hour="unknown" +# und werden uebersprungen — genau dieser Fehler ist schon einmal passiert. +# -------------------------------------------------------------------------- +TS_ACCESS = re.compile(r"\[(\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2})\s*([+-]\d{4})?\]") +TS_ERROR_NTZ = re.compile( + r"\[(\w{3}\s+\w{3}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2}(?:\.\d+)?\s+\d{4})\]") +TS_BRACKET = re.compile(r"\[([^\]]+)\]") + +MONTHS = {m: i + 1 for i, m in enumerate( + ["Jan", "Feb", "Mar", "Apr", "May", "Jun", + "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"])} + + +def parse_ts(line: str) -> int | None: + """Liefert Unix-Nanosekunden oder None, wenn kein Zeitstempel erkennbar.""" + m = TS_ACCESS.search(line) + if m: + base, off = m.group(1), m.group(2) + try: + dt = datetime.strptime(base, "%d/%b/%Y:%H:%M:%S") + if off: + sign = 1 if off[0] == "+" else -1 + mins = int(off[1:3]) * 60 + int(off[3:5]) + dt = dt.replace(tzinfo=timezone.utc) if False else dt + # Offset anwenden: lokale Zeit -> UTC + from datetime import timedelta + dt = dt - timedelta(minutes=sign * mins) + return int(dt.replace(tzinfo=timezone.utc).timestamp() * 1e9) + except ValueError: + pass + m = TS_ERROR_NTZ.search(line) + if m: + raw = re.sub(r"\s+", " ", m.group(1)).strip() + for fmt in ("%a %b %d %H:%M:%S.%f %Y", "%a %b %d %H:%M:%S %Y"): + try: + dt = datetime.strptime(raw, fmt) + # Apache schreibt Error-Logs in LOKALER Zeit (hier CEST, +02:00). + # Ohne Offset-Angabe wird die Zeile sonst um 2 h verschoben + # einsortiert. Der Aufrufer kann den Offset setzen. + return int(dt.replace(tzinfo=timezone.utc).timestamp() * 1e9) + except ValueError: + continue + return None + + +def classify(path: str) -> dict: + """Labels aus dem Dateipfad ableiten — identisch zum Live-Schema von Telegraf.""" + kind = "error" if "error" in os.path.basename(path) else "access" + # Domain aus dem Pfad: /home/users//logs//access.log + m = re.search(r"/logs/([^/]+)/", path) + domain = m.group(1) if m else "" + return {"kind": kind, "domain": domain} + + +def iter_lines(path: str): + """Zeilen einer Logdatei liefern — auch aus .gz.""" + if path.endswith(".gz"): + with gzip.open(path, "rt", errors="replace") as fh: + for ln in fh: + yield ln.rstrip("\n") + else: + with open(path, "rt", errors="replace") as fh: + for ln in fh: + yield ln.rstrip("\n") + + +def collect_from_dir(root: str): + """Alle access.log*/error.log* unterhalb von root finden.""" + for dirpath, _dirnames, filenames in os.walk(root): + for fn in filenames: + if fn.startswith(("access.log", "error.log")): + yield os.path.join(dirpath, fn) + + +def collect_from_tarball(tar_path: str): + """Rohlogs aus einem Archiv-Tarball (Pfade unter raw/). + + Die Tarballs des Backfills enthalten raw/. + """ + with tarfile.open(tar_path, "r:gz") as tf: + for member in tf.getmembers(): + if not member.isfile(): + continue + name = member.name + if "/raw/" not in name and not name.startswith("raw/"): + continue + base = os.path.basename(name) + if not base.startswith(("access.log", "error.log")): + continue + fh = tf.extractfile(member) + if fh is None: + continue + logpath = "/" + name.split("raw/", 1)[1] if "raw/" in name else name + for raw in fh: + yield logpath, raw.decode("utf-8", "replace").rstrip("\n") + + +def build_labels(node: str, logpath: str, imported: str) -> dict: + """Label-Set wie das Live-Schema von Telegraf, plus imported=1. + + Unterschiede zum Live-Schema (bewusst): + * `imported="1"` — trennt die Historie vom Live-Store. Nur fuer diese + Streams gilt die lange Retention (retention_stream in loki.yml). + Ohne das Label wuerde der Compactor die Historie nach 90 Tagen + loeschen. + * `kind`, `domain` — aus dem Pfad abgeleitet. Erlaubt Queries ohne + teure Regex-Arbeit auf `logfile` und macht Dashboards schneller. + """ + meta = classify(logpath) + return { + "source": "apache", + "host": node, + "node": node, + "logfile": logpath, + "service_name": "unknown_service", + "detected_level": "unknown", + "imported": imported, + "kind": meta["kind"], + "domain": meta["domain"] or "unknown", + } + + +def main() -> int: + ap = argparse.ArgumentParser( + description="Apache-Rohlogs fuer den Loki-Import aufbereiten") + ap.add_argument("--node", required=True, help="Node-Name (host-Label)") + ap.add_argument("--out", default="/root/apache-archive", + help="Ausgabeverzeichnis") + ap.add_argument("--roots", nargs="*", + default=["/var/log/apache2", "/home/users"], + help="Wurzelverzeichnisse der Logs") + ap.add_argument("--include-tarball", action="append", default=[], + help="Zusaetzlich Rohlogs aus diesem .tar.gz lesen") + ap.add_argument("--since", default=None, + help="Nur Zeilen ab diesem Datum (JJJJ-MM-TT)") + ap.add_argument("--until", default=None, help="Nur Zeilen bis (JJJJ-MM-TT)") + ap.add_argument("--tz-offset", type=int, default=2, + help="UTC-Offset der Error-Logs in Stunden (Default 2 = CEST)") + ap.add_argument("--limit-lines", type=int, default=0, + help="Nach N Zeilen abbrechen (fuer Tests)") + args = ap.parse_args() + + since_ns = (int(datetime.strptime(args.since, "%Y-%m-%d") + .replace(tzinfo=timezone.utc).timestamp() * 1e9) + if args.since else None) + until_ns = (int(datetime.strptime(args.until, "%Y-%m-%d") + .replace(tzinfo=timezone.utc).timestamp() * 1e9) + if args.until else None) + + os.makedirs(args.out, exist_ok=True) + tmpdir = tempfile.mkdtemp(prefix="loki-raw-", dir=args.out) + part = os.path.join(tmpdir, "raw.tsv") + state = os.path.join(tmpdir, "state.json") + + files: list[str] = [] + for r in args.roots: + files.extend(collect_from_dir(r)) + files = sorted(files) + print(f"[collect] {len(files)} Logdateien unter {args.roots} gefunden") + + stats = Counter() + n_ok = n_nots = n_skip = 0 + examples: list[str] = [] + + with open(part, "w", encoding="utf-8") as out: + sources = ((p, iter_lines(p)) for p in files) + for path, lines in sources: + lab = build_labels(args.node, path, "1") + lab_s = json.dumps(lab, ensure_ascii=False, separators=(",", ":")) + for raw in lines: + if not raw.strip(): + continue + ts = parse_ts(raw) + if ts is None: + n_nots += 1 + if len(examples) < 5: + examples.append(raw[:120]) + continue + if since_ns and ts < since_ns: + n_skip += 1 + continue + if until_ns and ts >= until_ns: + n_skip += 1 + continue + # Error-Logs ohne Zeitzone sind LOKALzeit -> nach UTC schieben + if lab["kind"] == "error" and TS_ERROR_NTZ.search(raw) \ + and not TS_ACCESS.search(raw): + ts -= args.tz_offset * 3600 * 10**9 + raw = raw.replace("\t", " ") + out.write(f"{ts}\t{lab_s}\t{raw}\n") + n_ok += 1 + stats[lab["kind"]] += 1 + if args.limit_lines and n_ok >= args.limit_lines: + break + if args.limit_lines and n_ok >= args.limit_lines: + break + + # Tarballs zuletzt (sie liegen historisch VOR den Live-Dateien). + for tb in args.include_tarball: + print(f"[collect] Tarball: {tb}") + for logpath, raw in collect_from_tarball(tb): + if not raw.strip(): + continue + ts = parse_ts(raw) + if ts is None: + n_nots += 1 + continue + if since_ns and ts < since_ns: + n_skip += 1 + continue + if until_ns and ts >= until_ns: + n_skip += 1 + continue + lab = build_labels(args.node, logpath, "1") + if lab["kind"] == "error" and TS_ERROR_NTZ.search(raw) \ + and not TS_ACCESS.search(raw): + ts -= args.tz_offset * 3600 * 10**9 + lab_s = json.dumps(lab, ensure_ascii=False, separators=(",", ":")) + out.write(f"{ts}\t{lab_s}\t{raw.replace(chr(9), ' ')}\n") + n_ok += 1 + stats[lab["kind"]] += 1 + + print(f"[collect] {n_ok} Zeilen geschrieben (access={stats['access']} " + f"error={stats['error']})") + if n_nots: + print(f"[collect] WARNUNG: {n_nots} Zeilen OHNE erkennbaren Zeitstempel " + f"wurden uebersprungen (kein stilles Verwerfen):") + for e in examples: + print(f" {e}") + if n_skip: + print(f"[collect] {n_skip} Zeilen ausserhalb des Zeitfensters " + f"(since={args.since} until={args.until})") + + if n_ok == 0: + print("[collect] FEHLER: keine Zeilen gesammelt — Abbruch.", file=sys.stderr) + return 1 + + # Groesse + Sortierung festhalten, damit der Import den Bestand beweisen kann. + size = os.path.getsize(part) + with open(state, "w") as fh: + json.dump({"node": args.node, "lines": n_ok, + "access": stats["access"], "error": stats["error"], + "without_ts": n_nots, "skipped": n_skip, + "since": args.since, "until": args.until, + "bytes": size, "tz_offset": args.tz_offset}, fh, indent=2) + print(f"[collect] Rohdatei: {part} ({size/1e6:.1f} MB, unsortiert)") + print(f"[collect] Zustand: {state}") + print() + print("[collect] NAECHSTER SCHRITT — von Hand auf dem imac (nicht hier):") + print(f" scp {part} mm@100.83.134.46:/tmp/") + print(f" scp {state} mm@100.83.134.46:/tmp/") + return 0 + + +if __name__ == "__main__": + sys.exit(main())