Analyse Mirrorbrain update logs

get_total_by_mirror.py

import pathlib
import re
from dataclasses import dataclass


@dataclass
class Folder:
    name: str
    time_spent: float


@dataclass
class Mirror:
    name: str
    folders: list[Folder]

    def time_spent(self) -> float:
        return sum(folder.time_spent for folder in self.folders)


scan_re = re.compile(r""".*Scanning mirror '(?P<mirror>.*)' at (?P<folder>.*)""")
completed_re = re.compile(r""".*Completed in (?P<value>.*) (?P<unit>.*)""")


mirrors: dict[str, Mirror] = {}

line_no = 0
folder: Folder | None = None
for line in (
    pathlib.Path("/Users/benoit/Downloads/Explore-logs-2023-12-11 10 23 27.txt")
    .read_text()
    .splitlines()
):
    scan_match = scan_re.match(line)
    if scan_match:
        mirror_name = scan_match.group("mirror")
        folder_name = scan_match.group("folder")

        if mirror_name not in mirrors:
            mirrors[mirror_name] = Mirror(name=mirror_name, folders=[])
        mirror = mirrors[mirror_name]
        folder = Folder(name=folder_name, time_spent=0)
        mirror.folders.append(folder)
        continue

    completed_match = completed_re.match(line)
    if completed_match and folder:
        value = float(completed_match.group("value"))
        unit = completed_match.group("unit")
        if unit == "hours":
            folder.time_spent = value * 3600
        elif unit == "minutes":
            folder.time_spent = value * 60
        elif unit == "seconds":
            folder.time_spent = value
        else:
            raise RuntimeError(f"Unexpected unit: {unit}")

for mirror_name in sorted(mirrors.keys()):
    mirror = mirrors[mirror_name]
    print(f"{mirror.name}: {mirror.time_spent()}")

mb_update_db_logs.md

• In grafana loki, identify a job name an extract all its logs in a txt file
• Launch the script to display stats by server