crawler_alert, monitoring: AI crawler metrics and dashboards

crawler-alert.py --metrics (5-minute timer) reads what the Traefik
access log gained since the last run, judges every request that names
an AI bot exactly like the weekly report (vendor IP ranges, forward-
confirmed rDNS, Meta's ASN) and writes cumulative counters to the
node_exporter textfile directory. Verdicts are cached per bot and IP
for a week, so DNS only sees new addresses; range files are refetched
once a day. It covers every router, while the mails keep watching
Forgejo only.

New dashboard "AI crawlers", and a row of it on the overview.

The overview's load gauge divided node_load1 by a count with fewer
labels, which matches nothing; aggregate both sides by host.

Assisted-by: Claude:opus-5-5

ansible/roles/crawler_alert/files/crawler-alert.py

@@ -286,6 +286,177 @@ def fcrdns(ip, suffixes):
     _rdns_cache[key] = result
     return result
 
+# --- --metrics: Prometheus counters for Grafana --------------------------
+# Run every few minutes from a timer. Reads only what the live log gained
+# since the last run (byte offset + inode in a state file, so rotation and
+# truncation restart at 0), classifies each request the same way as the
+# weekly report, and writes cumulative counters to node_exporter's
+# textfile directory. Verdicts per (bot, IP) are cached for a week, so DNS
+# is only asked about addresses it hasn't seen.
+METRICS = "--metrics" in sys.argv
+TEXTFILE = os.environ.get("CRAWLER_TEXTFILE", "/var/lib/node_exporter/textfile/crawler.prom")
+STATE = os.environ.get("CRAWLER_STATE", "/var/cache/crawler-alert/metrics.json")
+LIVE_LOG = os.environ.get("CRAWLER_LIVE_LOG", sorted(glob.glob(LOG_GLOB) or [""])[0])
+ALL_BOTS = sorted(set(BOT_VENDOR) | set(BOT_RDNS) | set(BOT_ASN), key=len, reverse=True)
+GENERIC_BOT = re.compile(r"bot|crawler|spider|scrapy|externalagent", re.I)
+UA_RE = re.compile(r"\"[^\"]*\" \"([^\"]*)\" \d+ \"")
+# The mail reports watch one router (Forgejo); the dashboard wants them all.
+METRICS_ROUTER = os.environ.get("CRAWLER_METRICS_ROUTER", ROUTER)
+
+
+def vendor_of(bot):
+    if bot in BOT_VENDOR:
+        return BOT_VENDOR[bot]
+    if bot in BOT_ASN:
+        return "meta"
+    return bot.lower()
+
+
+def index_ranges(vendors, cf):
+    idx = {}
+    for vendor, nets in vendors.items():
+        b = {}
+        for s in nets:
+            n = ipaddress.ip_network(s)
+            b.setdefault(int(n.network_address) >> 24 if n.version == 4 else -1, []).append(n)
+        idx[vendor] = b
+    return idx, [ipaddress.ip_network(s) for s in cf]
+
+
+def cached_ranges(max_age=86400):
+    """Range files change rarely: refetch once a day, not every run."""
+    import time
+    try:
+        if time.time() - os.path.getmtime(CACHE) < max_age:
+            c = json.load(open(CACHE))
+            return index_ranges(c["vendors"], c.get("cloudflare", []))
+    except Exception:
+        pass
+    idx, cf, _stale = load_ranges()
+    return idx, cf
+
+
+def judge(bot, ip, ranges, cfnets, cache, budget):
+    """verified | spoofed | proxied | private | unknown, like the report:
+    IP ranges first, then forward-confirmed rDNS, then single-tenant ASN."""
+    try:
+        addr = ipaddress.ip_address(ip)
+    except ValueError:
+        return "unknown"
+    if addr.is_private:
+        return "private"
+    if any(addr in n for n in cfnets):
+        return "proxied"
+    if bot in BOT_VENDOR and in_index(ranges.get(BOT_VENDOR[bot]), addr):
+        return "verified"
+    if bot not in BOT_RDNS and bot not in BOT_ASN:
+        return "spoofed"          # range-only vendor, address outside ranges
+    key = f"{bot}|{ip}"
+    if key in cache:
+        return cache[key][0]
+    if budget[0] <= 0:
+        return "unknown"          # over this run's DNS budget; retried next run
+    budget[0] -= 1
+    v = "unknown"
+    ok, why = False, "no-ptr"
+    if bot in BOT_RDNS:
+        ok, _host, why = fcrdns(ip, BOT_RDNS[bot])
+    if ok:
+        v = "verified"
+    elif why == "no-ptr":
+        if ASN_ENABLE and bot in BOT_ASN and asn_of(ip) in BOT_ASN[bot]:
+            v = "verified"
+    else:
+        v = "spoofed"             # a PTR exists and points elsewhere
+    cache[key] = [v, int(date.today().toordinal())]
+    return v
+
+
+def run_metrics():
+    try:
+        st = json.load(open(STATE))
+    except Exception:
+        st = {}
+    counts = Counter(st.get("counts", {}))
+    cache = {k: v for k, v in st.get("verdicts", {}).items()
+             if v[1] > date.today().toordinal() - 7}
+    try:
+        fs = os.stat(LIVE_LOG)
+    except OSError:
+        sys.exit(f"cannot read {LIVE_LOG}")
+    if "inode" not in st:
+        offset = fs.st_size       # first run: start at the end, not the backlog
+    elif st["inode"] != fs.st_ino or fs.st_size < st.get("offset", 0):
+        offset = 0                # rotated or truncated
+    else:
+        offset = st["offset"]
+    ranges, cfnets = cached_ranges()
+    socket.setdefaulttimeout(3)
+    budget = [RDNS_MAX]
+    with open(LIVE_LOG, "rb") as f:
+        f.seek(offset)
+        for raw in f:
+            if not raw.endswith(b"\n"):
+                break             # half-written line: take it next run
+            offset += len(raw)
+            line = raw.decode("utf-8", "replace")
+            if METRICS_ROUTER and METRICS_ROUTER not in line:
+                continue
+            parts = line.split()
+            counts["requests"] += 1
+            m = UA_RE.search(line)
+            u = m.group(1) if m else "-"
+            bot = next((b for b in ALL_BOTS if b in u), None)
+            if bot:
+                v = judge(bot, parts[0], ranges, cfnets, cache, budget)
+                counts[f"ai|{vendor_of(bot)}|{bot}|{v}"] += 1
+                if v == "verified" and len(parts) > 9 and parts[9].isdigit():
+                    counts[f"bytes|{vendor_of(bot)}"] += int(parts[9])
+            elif u == "-":
+                counts["class|none"] += 1
+            elif GENERIC_BOT.search(u):
+                counts["class|other-bot"] += 1
+            else:
+                counts["class|browser-like"] += 1
+    if len(cache) > 50000:        # keep the newest verdicts only
+        cache = dict(sorted(cache.items(), key=lambda kv: kv[1][1])[-50000:])
+    os.makedirs(os.path.dirname(STATE), exist_ok=True)
+    json.dump({"inode": fs.st_ino, "offset": offset, "counts": counts,
+               "verdicts": cache}, open(STATE + ".tmp", "w"))
+    os.replace(STATE + ".tmp", STATE)
+
+    out = ["# HELP crawler_requests_total Requests seen in the Traefik access log.",
+           "# TYPE crawler_requests_total counter",
+           f"crawler_requests_total {counts['requests']}",
+           "# HELP crawler_class_requests_total Requests without an AI-bot user agent, by user-agent class.",
+           "# TYPE crawler_class_requests_total counter"]
+    for k, c in sorted(counts.items()):
+        if k.startswith("class|"):
+            out.append(f'crawler_class_requests_total{{class="{k[6:]}"}} {c}')
+    out += ["# HELP crawler_ai_requests_total Requests claiming an AI-bot user agent, by verification verdict.",
+            "# TYPE crawler_ai_requests_total counter"]
+    for k, c in sorted(counts.items()):
+        if k.startswith("ai|"):
+            _, vendor, bot, v = k.split("|")
+            out.append(f'crawler_ai_requests_total{{vendor="{vendor}",bot="{bot}",verdict="{v}"}} {c}')
+    out += ["# HELP crawler_ai_verified_bytes_total Bytes served to verified AI crawlers.",
+            "# TYPE crawler_ai_verified_bytes_total counter"]
+    for k, c in sorted(counts.items()):
+        if k.startswith("bytes|"):
+            out.append(f'crawler_ai_verified_bytes_total{{vendor="{k[6:]}"}} {c}')
+    import time
+    out += ["# HELP crawler_metrics_last_run_timestamp_seconds Last successful --metrics run.",
+            "# TYPE crawler_metrics_last_run_timestamp_seconds gauge",
+            f"crawler_metrics_last_run_timestamp_seconds {int(time.time())}"]
+    with open(TEXTFILE + ".tmp", "w") as f:
+        f.write("\n".join(out) + "\n")
+    os.replace(TEXTFILE + ".tmp", TEXTFILE)
+
+
+if METRICS:
+    run_metrics()
+    sys.exit(0)
+
 days = [date.today() - timedelta(days=i) for i in range(7 if WEEKLY else 1, 0, -1)]
 stamps = [d.strftime("%d/%b/%Y") for d in days]
 

ansible/roles/crawler_alert/files/crawler-metrics.service

@@ -0,0 +1,12 @@
+[Unit]
+Description=Crawler counters for Prometheus (node_exporter textfile)
+After=docker.service
+
+[Service]
+Type=oneshot
+EnvironmentFile=/etc/crawler-alert.env
+# Every router, not only Forgejo's; fewer DNS lookups per 5-minute run
+Environment=CRAWLER_METRICS_ROUTER= CRAWLER_RDNS_MAX=100
+ExecStart=/usr/local/bin/crawler-alert.py --metrics
+Nice=10
+TimeoutStartSec=4min

ansible/roles/crawler_alert/files/crawler-metrics.timer

@@ -0,0 +1,9 @@
+[Unit]
+Description=Crawler counters every 5 minutes
+
+[Timer]
+OnCalendar=*:0/5
+RandomizedDelaySec=30
+
+[Install]
+WantedBy=timers.target

ansible/roles/crawler_alert/tasks/main.yaml

@@ -22,6 +22,8 @@
     - crawler-alert.timer
     - crawler-weekly.service
     - crawler-weekly.timer
+    - crawler-metrics.service
+    - crawler-metrics.timer
   notify: Reload systemd
 
 - name: Flush handlers
@@ -35,3 +37,4 @@
   loop:
     - crawler-alert.timer
     - crawler-weekly.timer
+    - crawler-metrics.timer

ansible/roles/crawler_alert/templates/crawler-alert.env.j2

@@ -6,3 +6,5 @@ CRAWLER_THRESHOLD={{ crawler_threshold }}
 CRAWLER_MAIL_CONTAINER=forgejo
 CRAWLER_MAIL_INI=/data/gitea/conf/app.ini
 CRAWLER_ROBOTS={{ crawler_robots_url }}
+CRAWLER_LIVE_LOG=/opt/traefik/logs/access.log
+CRAWLER_TEXTFILE={{ monitoring_textfile_dir }}/crawler.prom