Expand source code
import click
from rugged.lib.host_header import print_host_header
from rugged.lib.logger import get_logger, get_log_entries, truncate_log
from rugged.lib.task_queue import run_task
from rugged.workers import get_workers
log = get_logger()
@click.command("logs")
@click.option(
"--local",
is_flag=True,
default=False,
help="Print the local logs. "
"Defaults to true, unless a worker is specified.",
)
@click.option(
"--worker",
default=[],
multiple=True,
help="The specific worker from which to retrieve logs. "
"Can be passed multiple times.",
)
@click.option(
"--limit",
default=10,
help="The number of lines to print for each log.",
)
@click.option(
"--truncate",
is_flag=True,
default=False,
help="Instead of printing logs, empty log file(s).",
)
def logs_cmd(local, worker, limit, truncate):
""" Print the logs for a TUF repository. """
if local or not worker:
if truncate:
_truncate_local_logs()
else:
_print_local_logs(limit)
if local and not worker:
return
workers = get_workers(worker)
if truncate:
_truncate_worker_logs(workers)
else:
_print_worker_logs(workers, limit)
def _print_local_logs(limit):
""" Print the local logs for a TUF repository. """
logs = get_log_entries(log)
_print_logs('local operations', logs, limit)
def _print_worker_logs(workers, limit):
""" Print the workers' logs for a TUF repository. """
for worker in workers:
logs = run_task(worker, 'logs')
_print_logs(worker, logs, limit)
def _print_logs(host, logs, limit):
""" Print formatted logs for a given host. """
for filename in list(logs):
print_host_header("Log", f"{ host }: { filename }")
for line in logs[filename][-limit:]:
print(line, end='')
def _truncate_local_logs():
""" Truncate local Rugged TUF logs. """
truncated_logs = truncate_log(log)
for logfile in truncated_logs:
click.echo(f"Truncated local operations log at: { logfile }")
def _truncate_worker_logs(workers):
""" Truncate workers' Rugged TUF logs. """
for index, worker in enumerate(workers):
truncated_logs = run_task(worker, 'truncate_logs')
for logfile in truncated_logs:
click.echo(f"Truncated { worker} log at: { logfile }")