#!/usr/bin/env python3 """ apache-log-archive.py — Alte Apache-Logs parsen und als Prometheus-Archiv ablegen. ABLAUF 1. Logs einlesen (auch rotierte .gz), Access- UND Error-Logs. 2. Pro Stunde/Domain/Status zu Metriken aggregieren. 3. OpenMetrics-Dateien schreiben (Prometheus-Textformat mit Zeitstempeln). 4. `promtool tsdb create-blocks-from openmetrics` erzeugt daraus TSDB-Blöcke. 5. Ergebnis als tar.gz bereitstellen — zur Uebergabe an den zentralen Server. WARUM AGGREGIERT UND NICHT PRO REQUEST Prometheus ist eine TSDB, kein Logstore. Eine Serie pro Request waere bei Millionen Zeilen sofort toedlich (Kardinalitaet). Aggregiert auf Stunde/Domain/Status bleiben es wenige tausend Serien — und beantwortet genau die Fragen, fuer die man alte Logs anschaut: Wie viele Requests, welche Fehlerquote, welcher Traffic, zu welcher Zeit. Rohzeilen gehoeren in ein Logarchiv (Loki/tar.gz), nicht in die TSDB. Dieses Skript produziert beides: Metriken fuer Prometheus UND ein tar.gz der Rohlogs. ERZEUGTE METRIKEN (Name : Labels) apache_archive_requests_total : domain, method, status, status_class apache_archive_response_bytes_total : domain apache_archive_unique_clients : domain (geschaetzt, pro Stunde) apache_archive_errors_total : domain, level apache_archive_log_lines_total : domain, kind (access|error|unparsed) Alle tragen zusaetzlich label "archived=1" und "instance=", damit sie im Dashboard von Live-Daten unterscheidbar sind. AUFRUF (auf dem Quellhost, als root) python3 apache-log-archive.py --node x22.idx.wf-keyhelp python3 apache-log-archive.py --node x22 --out /root/apache-archive Danach die .tar.gz auf den imac bringen und dort einspielen — AUSSCHLIESSLICH ueber prom-archive-import.sh in die getrennte ARCHIV-Instanz (:9091). NIEMALS die Bloecke von Hand in die Produktions-TSDB entpacken: ein Block mit Zeitstempeln ausserhalb des Live-Fensters verschiebt dort die Zeitreferenz und laesst jeden echten Sample als "too old" verwerfen (real passiert am 14.09.2026, 13 Nodes verloren ihre Metriken). """ import argparse import glob import gzip import os import re import subprocess import sys import tarfile import tempfile from collections import defaultdict from datetime import datetime, timezone PATTERNS = [ "/var/log/apache2/*access*.log", "/var/log/apache2/*access*.log.*", "/var/log/apache2/*error*.log", "/var/log/apache2/*error*.log.*", "/home/users/*/logs/*/access.log", "/home/users/*/logs/*/access.log.*", "/home/users/*/logs/*/error.log", "/home/users/*/logs/*/error.log.*", "/var/www/vhosts/*/logs/access_log*", "/var/www/vhosts/*/logs/error_log*", ] RE_LINE = re.compile( r'^(?P\S+) \S+ \S+ \[(?P[^\]]+)\] "(?P[^"]*)" (?P\d{3}|-) (?P\d+|-)' ) RE_ERR = re.compile(r'^\[(?P[^\]]+)\] \[(?P[^\]:]+):(?P\w+)\]') TS_FORMATS = ["%d/%b/%Y:%H:%M:%S %z", "%a %b %d %H:%M:%S.%f %Y", "%a %b %d %H:%M:%S %Y"] # Metriknamen, die das Skript erzeugt — auch fuer den Import-Loader relevant. METRICS = [ "apache_archive_requests_total", "apache_archive_response_bytes_total", "apache_archive_unique_clients", "apache_archive_errors_total", "apache_archive_log_lines_total", ] def parse_ts(raw): for fmt in TS_FORMATS: try: dt = datetime.strptime(raw.strip(), fmt) except ValueError: continue if dt.tzinfo is None: dt = dt.astimezone() return dt.astimezone(timezone.utc) return None def status_class(code): try: c = int(code) except (TypeError, ValueError): return "unknown" return f"{c // 100}xx" def esc(v): """Label-Wert fuer OpenMetrics escapen (Backslash, Anfuehrungszeichen, LF).""" return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") def collect(patterns): """Alle Logdateien finden und parsen. Liefert Zaehler und Rohdateiliste.""" files = sorted({f for pat in patterns for f in glob.glob(pat)}) # requests[(hour, domain, method, status)] = count requests = defaultdict(int) bytes_by_domain = defaultdict(int) bytes_by_hour = defaultdict(int) # (hour, domain) -> Bytes clients = defaultdict(set) # (hour, domain) -> {ip} errors = defaultdict(int) # (hour, domain, level) -> count lines = defaultdict(int) # (domain, kind) -> count for path in files: domain = os.path.basename(os.path.dirname(path)) or os.path.basename(path) try: fh = (gzip.open(path, "rt", errors="replace") if path.endswith(".gz") else open(path, "r", errors="replace")) except OSError: continue with fh: for line in fh: m = RE_LINE.match(line) if m: lines[(domain, "access")] += 1 ts = parse_ts(m.group("ts")) req = m.group("req") or "" method = req.split(" ", 1)[0] if " " in req else "UNKNOWN" status = m.group("status") or "unknown" hour = ts.strftime("%Y-%m-%dT%H:00:00Z") if ts else "unknown" requests[(hour, domain, method, status)] += 1 try: nb = int(m.group("bytes") or 0) bytes_by_domain[domain] += nb if ts: bytes_by_hour[(hour, domain)] += nb except ValueError: pass if ts: clients[(hour, domain)].add(m.group("ip")) continue m = RE_ERR.match(line) if m: lines[(domain, "error")] += 1 ts = parse_ts(m.group("ts")) hour = ts.strftime("%Y-%m-%dT%H:00:00Z") if ts else "unknown" errors[(hour, domain, m.group("lvl").lower())] += 1 continue lines[(domain, "unparsed")] += 1 return files, requests, bytes_by_domain, bytes_by_hour, clients, errors, lines def write_openmetrics(path, node, requests, bytes_by_domain, bytes_by_hour, clients, errors, lines): """OpenMetrics-Textdatei schreiben. Zeitstempel in SEKUNDEN. ACHTUNG: OpenMetrics will Sekunden (float). Nicht Millisekunden — Prometheus multipliziert beim Import mit 1000 und legt die Bloecke sonst ins Jahr 58.662, wo sie verworfen werden. """ base_labels = f'archived="1",instance="{esc(node)}"' n = 0 _max_ts = 0 # groesster geschriebener Zeitstempel (Einheiten-Waechter) with open(path, "w") as fh: # EINHEITEN — hier lag ein realer Fehler: # OpenMetrics-Zeitstempel sind SEKUNDEN (float). Prometheus wandelt beim # Import ms = s * 1000. Wer hier Millisekunden schreibt, bekommt Bloecke # im Jahr 58.662 — sie werden als "too far into the future" verworfen # und die Abfrage liefert stillschweigend nichts. Kein * 1000. fh.write("# TYPE apache_archive_requests_total counter\n") fh.write("# TYPE apache_archive_response_bytes_total counter\n") fh.write("# TYPE apache_archive_unique_clients gauge\n") fh.write("# TYPE apache_archive_errors_total counter\n") fh.write("# TYPE apache_archive_log_lines_total counter\n") for (hour, domain, method, status), cnt in sorted(requests.items()): if hour == "unknown": continue ms = int(datetime.strptime(hour, "%Y-%m-%dT%H:00:00Z") .replace(tzinfo=timezone.utc).timestamp()) labels = (f'{base_labels},domain="{esc(domain)}",method="{esc(method)}",' f'status="{esc(status)}",status_class="{esc(status_class(status))}"') fh.write(f"apache_archive_requests_total{{{labels}}} {cnt} {ms}\n") _max_ts = max(_max_ts, ms) n += 1 # Bytes pro Stunde: das ergibt die Traffic-Zeitreihe im Archiv. for (hour, domain), total in sorted(bytes_by_hour.items()): if hour == "unknown": continue ms = int(datetime.strptime(hour, "%Y-%m-%dT%H:00:00Z") .replace(tzinfo=timezone.utc).timestamp()) labels = f'{base_labels},domain="{esc(domain)}"' fh.write(f"apache_archive_response_bytes_total{{{labels}}} {total} {ms}\n") _max_ts = max(_max_ts, ms) n += 1 for (hour, domain), ips in sorted(clients.items()): try: ms = int(datetime.strptime(hour, "%Y-%m-%dT%H:00:00Z") .replace(tzinfo=timezone.utc).timestamp()) except ValueError: continue labels = f'{base_labels},domain="{esc(domain)}"' fh.write(f"apache_archive_unique_clients{{{labels}}} {len(ips)} {ms}\n") _max_ts = max(_max_ts, ms) n += 1 for (hour, domain, level), cnt in sorted(errors.items()): if hour == "unknown": continue ms = int(datetime.strptime(hour, "%Y-%m-%dT%H:00:00Z") .replace(tzinfo=timezone.utc).timestamp()) labels = f'{base_labels},domain="{esc(domain)}",level="{esc(level)}"' fh.write(f"apache_archive_errors_total{{{labels}}} {cnt} {ms}\n") _max_ts = max(_max_ts, ms) n += 1 # Zeilenzahlen sind Zustandswerte ohne eigenen Zeitbezug. Sie werden auf # die LETZTE Log-Stunde gestempelt, nicht auf "jetzt" — sonst haengt ein # Archiv-Eintrag am Abruftag und die Historie wirkt verschoben. _hours = [h for (h, _, _, _) in requests if h != "unknown"] if _hours: stamp_s = int(datetime.strptime(max(_hours), "%Y-%m-%dT%H:00:00Z") .replace(tzinfo=timezone.utc).timestamp()) else: stamp_s = int(datetime.now(timezone.utc).timestamp()) for (domain, kind), cnt in sorted(lines.items()): labels = f'{base_labels},domain="{esc(domain)}",kind="{esc(kind)}"' fh.write(f"apache_archive_log_lines_total{{{labels}}} {cnt} {stamp_s}\n") _max_ts = max(_max_ts, stamp_s) n += 1 # EINHEITEN-WAECHTER: Der groesste Zeitstempel muss ein plausibles # Sekunden-Datum ergeben. Faengt 1000x-Fehler (ms statt s) sofort ab, # statt sie still in einen nutzlosen Block zu schreiben. if _max_ts > 4102444800: # nach dem Jahr 2100 -> kein Sekundenwert raise SystemExit( f"FEHLER: Zeitstempel {_max_ts} liegt nach dem Jahr 2100.\n" f" OpenMetrics erwartet SEKUNDEN, nicht Millisekunden.\n" f" Pruefe die *1000-Multiplikationen in diesem Skript.\n" f" Ohne Korrektur entstehen Bloecke, die Prometheus verwirft.") # PFLICHT: OpenMetrics verlangt "# EOF" als letzte Zeile. Ohne das # lehnt promtool ab: "data does not end with # EOF". fh.write("# EOF\n") return n def main(): ap = argparse.ArgumentParser(description="Apache-Logs als Prometheus-Archiv aufbereiten") ap.add_argument("--node", required=True, help="Node-Name fuer das instance-Label") ap.add_argument("--out", default="/root/apache-archive", help="Ausgabeverzeichnis") ap.add_argument("--include-raw", action="store_true", help="Rohlogs mit ins tar.gz packen (empfohlen)") args = ap.parse_args() os.makedirs(args.out, exist_ok=True) print(f"[1/5] Logs einlesen …") files, requests, bytes_by_domain, bytes_by_hour, clients, errors, lines = collect(PATTERNS) if not files: print("Keine Logdateien gefunden. Muster:", ", ".join(PATTERNS[:4]), file=sys.stderr) return 1 total_lines = sum(lines.values()) print(f" {len(files)} Dateien, {total_lines:,} Zeilen") print(f" {sum(requests.values()):,} Zugriffe, {sum(errors.values()):,} Fehlereintraege") print(f"[2/5] OpenMetrics schreiben …") om = os.path.join(args.out, "apache-archive.openmetrics") n = write_openmetrics(om, args.node, requests, bytes_by_domain, bytes_by_hour, clients, errors, lines) size = os.path.getsize(om) print(f" {n:,} Datenpunkte, {size/1024:.1f} KiB") # promtool suchen — liegt beim prometheus-Container bzw. als Binary vor. promtool = None for cand in ("promtool", "/usr/local/bin/promtool", "/usr/bin/promtool"): try: subprocess.run([cand, "--version"], capture_output=True, check=True) promtool = cand break except (OSError, subprocess.CalledProcessError): continue if promtool is None: print(f"[3/5] promtool nicht gefunden — openmetrics-Datei bleibt liegen.") print(f" Der zentrale Server kann sie direkt verarbeiten:") print(f" promtool tsdb create-blocks-from openmetrics {om} ") blocks_dir = None else: print(f"[3/5] TSDB-Bloecke erzeugen (promtool) …") blocks_dir = os.path.join(args.out, "blocks") os.makedirs(blocks_dir, exist_ok=True) r = subprocess.run([promtool, "tsdb", "create-blocks-from", "openmetrics", om, blocks_dir], capture_output=True, text=True) if r.returncode != 0: print(" promtool-Fehler:", r.stderr.strip()[:400], file=sys.stderr) blocks_dir = None else: print(" " + (r.stdout.strip() or "Bloecke erzeugt")) print(f"[4/5] Archiv packen …") tar_path = os.path.join(args.out, f"apache-archive-{args.node}.tar.gz") with tarfile.open(tar_path, "w:gz") as tf: tf.add(om, arcname="openmetrics/apache-archive.openmetrics") if blocks_dir: tf.add(blocks_dir, arcname="blocks") if args.include_raw: # Rohlogs mitsichern: Prometheus hat nur die Aggregate — fuer die # Fehlersuche im Einzelfall braucht es die Zeilen. for f in files: try: tf.add(f, arcname="raw/" + f.lstrip("/")) except (OSError, tarfile.TarError): pass print(f" {tar_path} ({os.path.getsize(tar_path)/1024/1024:.1f} MiB)") print(f"[5/5] Fertig.") print() print(f" Uebergabe an den zentralen Server:") print(f" scp {tar_path} mm@192.168.188.177:/tmp/") print() print(f" ============================================================") print(f" EINSPIELEN — NUR in die ARCHIV-Instanz (:9091)") print(f" ============================================================") print(f" ssh mm@192.168.188.177 \\") print(f" 'sudo /usr/local/bin/prom-archive-import.sh --file /tmp/{os.path.basename(tar_path)}'") print() print(f" Das Import-Skript prueft die Zeitstempel VOR dem Einspielen, legt ein") print(f" Manifest an (Schutz gegen Doppel-Import) und schreibt ausschliesslich") print(f" ins Archiv-Verzeichnis.") print() print(f" NIEMALS von Hand entpacken — insbesondere NICHT nach") print(f" central-imac/data/prometheus/. Genau dieser Weg hat am 14.09.2026 die") print(f" Zeitreferenz der Produktions-TSDB verschoben: 13 Nodes verloren ihre") print(f" Metriken, jeder Live-Sample galt danach als 'too old'.") print() print(f" Hinweis: instance=\"{args.node}\", archived=\"1\" — im Dashboard von") print(f" Live-Daten unterscheidbar (Filter archived!=\"1\" blendet das Archiv aus).") return 0 if __name__ == "__main__": sys.exit(main())