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.
This commit is contained in:
Kalina committed
1 parent
33534811cb
commit
f8c5f2eb3c
2 files changed
+441
No files matched your search
@@ -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:
|
||||
|
||||
<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):
|
||||
"""Rohlogs aus einem Archiv-Tarball (Pfade unter raw/).
|
||||
|
||||
Die Tarballs des Backfills enthalten raw/<absoluter-pfad-ohne-fuehrenden-slash>.
|
||||
"""
|
||||
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())
|
||||
Reference in new issue
Block a user