Kalina
d65289e65f
Die .gz-Dateien INNERHALB eines Backfill-Tarballs sind nochmals gzip-komprimiert; tarfile entpackt nur die aeussere Ebene. Wer den Member-Inhalt direkt dekodiert, liest gzip-Magic als Text und verliert jede Zeile dieser Dateien — real 68.975 von 404.825 (17 %). - .gz-Member werden ein zweites Mal entpackt, die Zahl wird gemeldet - --limit-lines greift jetzt auch im Tarball-Zweig - Beispiele unlesbarer Zeilen zeigen den Dateinamen
335 lines
14 KiB
Python
335 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Rohlogs (Apache access+error) eines Hosts in eine importierbare Datei schreiben.
|
|
|
|
Erzeugt eine DREISPALTIGE, nach Zeitstempel sortierbare Datei:
|
|
|
|
<ts_ns> <TAB> <labels-json> <TAB> <rohzeile>
|
|
|
|
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/<user>/logs/<domain>/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, stats: Counter | None = None):
|
|
"""Rohlogs aus einem Archiv-Tarball (Pfade unter raw/).
|
|
|
|
Die Tarballs des Backfills enthalten raw/<absoluter-pfad-ohne-fuehrenden-slash>.
|
|
|
|
FALLE (real, 17 % Datenverlust): die `.gz`-Dateien IM Tarball sind selbst
|
|
noch gzip-komprimiert. `tarfile` entpackt nur die aeussere Ebene — wer den
|
|
Member-Inhalt direkt dekodiert, liest das gzip-Magic (\x1f\x8b) als Text und
|
|
verliert jede Zeile dieser Dateien. Gemessen am x14-Tarball: 68.975 von
|
|
404.825 Zeilen (17 %) fielen so aus, verteilt auf genau die `.gz`-Member
|
|
(access.log.2/.3/.4.gz, error.log.2/.3/.4.gz). Deshalb: Member mit `.gz`
|
|
ein zweites Mal entpacken.
|
|
"""
|
|
gz_ok = gz_fail = 0
|
|
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
|
|
data = fh.read()
|
|
text = None
|
|
if name.endswith(".gz"):
|
|
try:
|
|
text = gzip.decompress(data).decode("utf-8", "replace")
|
|
gz_ok += 1
|
|
except Exception: # noqa: BLE001
|
|
gz_fail += 1
|
|
text = data.decode("utf-8", "replace")
|
|
else:
|
|
text = data.decode("utf-8", "replace")
|
|
for raw in text.splitlines():
|
|
yield logpath, raw
|
|
if stats is not None:
|
|
stats["gz_member_ok"] += gz_ok
|
|
stats["gz_member_fail"] += gz_fail
|
|
|
|
|
|
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, stats):
|
|
if args.limit_lines and n_ok >= args.limit_lines:
|
|
break
|
|
if not raw.strip():
|
|
continue
|
|
ts = parse_ts(raw)
|
|
if ts is None:
|
|
n_nots += 1
|
|
if len(examples) < 5:
|
|
examples.append(f"[{os.path.basename(logpath)}] "
|
|
f"{raw[:110]}")
|
|
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 stats["gz_member_ok"] or stats["gz_member_fail"]:
|
|
print(f"[collect] gzip-Member im Tarball entpackt: "
|
|
f"{stats['gz_member_ok']}"
|
|
+ (f", FEHLGESCHLAGEN: {stats['gz_member_fail']}"
|
|
if stats["gz_member_fail"] else ""))
|
|
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())
|