fix(apache): Sekunden-Kollision behoben (log_time + converter) — Node lieferte gar nichts mehr
Echte x22-Logs: 2000 Zeilen -> 286 Kollisionen (duplicate sample). Telegraf uebernimmt den Log-Zeitstempel; bei mehreren Treffern pro Sekunde kollidiert der +1ms-Aufschlag mit echten Zeilen. Prometheus weist den Batch ab, Telegraf retryt endlos -> FIFO steht -> Node liefert GAR NICHTS mehr. Fix: eigenes Grok-Muster mit log_time als Feld (Telegraf nutzt eigene ns-Uhr) + processors.converter, damit resp_code/verb Tags bleiben. Gemessen: 2000/2000 Samples, 0 Kollisionen. Ausserdem: Einheiten-Waechter im Archiver (ms statt s -> Abbruch statt nutzloser Block).
This commit is contained in:
mm committed
1 parent
5a213eb580
commit
5483a4fea8
7 files changed
+626
-7
No files matched your search
@@ -0,0 +1,329 @@
|
||||
#!/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=<node>", 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 (siehe README).
|
||||
"""
|
||||
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<ip>\S+) \S+ \S+ \[(?P<ts>[^\]]+)\] "(?P<req>[^"]*)" (?P<status>\d{3}|-) (?P<bytes>\d+|-)'
|
||||
)
|
||||
RE_ERR = re.compile(r'^\[(?P<ts>[^\]]+)\] \[(?P<mod>[^\]:]+):(?P<lvl>\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} <blockdir>")
|
||||
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(f" Dort einspielen:")
|
||||
print(f" tar -xzf /tmp/{os.path.basename(tar_path)} -C /data/stacks/netbird-observability/stacks/central-imac/data/prometheus/")
|
||||
print(f" # Bloecke nach <prometheus>/ — Prometheus liest sie beim naechsten Start")
|
||||
print(f" docker restart observability-prometheus")
|
||||
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())
|
||||
Reference in new issue
Block a user