Cronjob script to capture prepared HAProxy logs

haproxy_log_parser.py

from systemd import journal # apt-get install python3-systemd
from datetime import datetime, timedelta
from dataclasses import dataclass
from collections import defaultdict
import json
import re

HAPROXY_SERVICE="haproxy.service"
COUNTER_FILE="/root/haproxy_log_parser/stats.json"
# cron
# 10 * * * * /root/.local/bin/uv run --project /root/hlp /root/hlp/haproxy_log_parser.py 1>/root/hlp/hlp_stdout.log 2>/root/hlp/hlp_stderr.log

@dataclass(frozen=True)
class Entry:
    ja4: str
    user_agent: str
    ip: str

def floor_to_hour(t):
    return t.replace(second=0, microsecond=0, minute=0, hour=t.hour)

# At 6:10 I should collect logs between 5:00 and 6:00
def times():
    t = datetime.now().astimezone()
    end = floor_to_hour(t)
    start = floor_to_hour(t-timedelta(hours=1))
    return (start, end)

def parse_log(line):
    pattern = r'(?P<ip>[0-9.]*):.*\{(?P<ja4>[^|]*)\|(?P<ua>[^|]*)\|(?P<real_ip>[^|]*)\}.*'
    match = re.search(pattern, line)
    if match is not None:
        # prefer the real ip
        ip = match.group("real_ip") if match.group("real_ip")!="" and match.group("real_ip")!=match.group("ip") else match.group("ip")
        ja4 = match.group("ja4")
        ua = match.group("ua")
        return Entry(ja4=ja4, user_agent=ua, ip=ip)
    else:
        return None

def get_logs(start, end, service):
    j = journal.Reader()
    j.add_match(f"_SYSTEMD_UNIT={service}")
    j.seek_realtime(start)
    counter = defaultdict(int)
    for entry in j:
        if entry['__REALTIME_TIMESTAMP'] > end:
            break
        log = parse_log(entry['MESSAGE'])
        counter[log]+=1
    return counter

def main():
    start, end = times()
    counter = get_logs(start, end, HAPROXY_SERVICE)
    with open(COUNTER_FILE, 'r') as f:
        old_data = json.load(f)
    for key in counter:
        if str(key) in old_data:
            old_data[str(key)]+=counter[key]
        else:
            old_data[str(key)]=counter[key]
    with open(COUNTER_FILE, 'w') as f:
        json.dump(old_data, f, indent=2)

if __name__ == "__main__":
    main()