ansible/roles/rsyslog_collector/files/rotate

#!/usr/bin/env python3
import argparse
import datetime
import logging
import pathlib
import subprocess


def compress_file(filename):
    subprocess.run(["xz", "-z", str(filename)])


def find_files(base_dir, extension, days):
    delta = datetime.timedelta(days=days)
    now = datetime.datetime.now()
    for path in base_dir.rglob(f"*{extension}"):
        stem = path.name.removesuffix(extension)
        date = datetime.datetime.fromisoformat(stem)
        if date < now - delta:
            yield path


def compress_logs(base_dir, days):
    for path in find_files(base_dir, ".log", days):
        logging.info("Compressing log file %s", str(path))
        compress_file(path)


def remove_logs(base_dir, days):
    for path in find_files(base_dir, ".log.xz", days):
        logging.info("Removing log file %s", str(path))
        path.unlink()


def main():

    parser = argparse.ArgumentParser()
    parser.add_argument("--compress-days", type=int, default=0)
    parser.add_argument("--keep-days", type=int, default=0)
    parser.add_argument(
        "--base-dir", type=pathlib.Path, default="/var/log/remote"
    )

    args = parser.parse_args()

    logging.basicConfig(format="%(levelname)s %(message)s", level=logging.INFO)

    logging.info("Rotate script started")

    if args.compress_days > 0:
        compress_logs(args.base_dir, args.compress_days)

    if args.keep_days > 0:
        remove_logs(args.base_dir, args.keep_days)

    logging.info("Rotate script done")


if __name__ == "__main__":
    main()