mirror of
https://github.com/ArchiveBox/ArchiveBox.git
synced 2026-09-12 19:50:57 +05:00
564 lines
20 KiB
Python
564 lines
20 KiB
Python
__package__ = "archivebox.config"
|
|
|
|
import inspect
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import quote, urlencode
|
|
|
|
from admin_data_views.typing import ItemContext, SectionData, TableContext
|
|
from admin_data_views.utils import ItemLink, render_with_item_view, render_with_table_view
|
|
from django.http import HttpRequest
|
|
from django.utils import timezone
|
|
from django.utils.html import format_html
|
|
from django.utils.safestring import mark_safe
|
|
|
|
from archivebox.config import CONSTANTS
|
|
from archivebox.machine.models import Binary
|
|
from archivebox.misc.util import parse_date
|
|
|
|
ENVIRONMENT_BINARIES_BASE_URL = "/admin/environment/binaries/"
|
|
INSTALLED_BINARIES_BASE_URL = "/admin/machine/binary/"
|
|
LOG_DETAIL_TAIL_BYTES = 100_000
|
|
|
|
|
|
def _read_text_tail(path: Path, max_bytes: int = LOG_DETAIL_TAIL_BYTES) -> str:
|
|
with path.open("rb") as log_file:
|
|
size = path.stat().st_size
|
|
log_file.seek(max(size - max_bytes, 0))
|
|
return log_file.read(max_bytes).decode("utf-8", errors="replace")
|
|
|
|
|
|
def is_superuser(request: HttpRequest) -> bool:
|
|
return bool(request.user.is_superuser)
|
|
|
|
|
|
def format_parsed_datetime(value: object) -> str:
|
|
parsed = parse_date(value)
|
|
return parsed.strftime("%Y-%m-%d %H:%M:%S") if parsed else ""
|
|
|
|
|
|
def get_environment_binary_url(name: str) -> str:
|
|
return f"{ENVIRONMENT_BINARIES_BASE_URL}{quote(name)}/"
|
|
|
|
|
|
def get_installed_binary_change_url(name: str, binary: Binary | None) -> str | None:
|
|
if binary is None or not binary.id:
|
|
return None
|
|
|
|
base_url = binary.admin_change_url
|
|
changelist_filters = urlencode({"q": name})
|
|
return f"{base_url}?{urlencode({'_changelist_filters': changelist_filters})}"
|
|
|
|
|
|
def render_binary_detail_description(name: str, merged: dict[str, Any], db_binary: Any) -> str:
|
|
installed_binary_url = get_installed_binary_change_url(name, db_binary)
|
|
|
|
if installed_binary_url:
|
|
return str(
|
|
format_html(
|
|
'<code>{}</code><br/><a href="{}">View Installed Binary Record</a>',
|
|
merged["abspath"],
|
|
installed_binary_url,
|
|
),
|
|
)
|
|
|
|
return str(format_html("<code>{}</code>", merged["abspath"]))
|
|
|
|
|
|
def obj_to_yaml(obj: Any, indent: int = 0) -> str:
|
|
indent_str = " " * indent
|
|
if indent == 0:
|
|
indent_str = "\n" # put extra newline between top-level entries
|
|
|
|
if isinstance(obj, dict):
|
|
if not obj:
|
|
return "{}"
|
|
result = "\n"
|
|
for key, value in obj.items():
|
|
result += f"{indent_str}{key}:{obj_to_yaml(value, indent + 1)}\n"
|
|
return result
|
|
|
|
elif isinstance(obj, list):
|
|
if not obj:
|
|
return "[]"
|
|
result = "\n"
|
|
for item in obj:
|
|
result += f"{indent_str}- {obj_to_yaml(item, indent + 1).lstrip()}\n"
|
|
return result.rstrip()
|
|
|
|
elif isinstance(obj, str):
|
|
if "\n" in obj:
|
|
return f" |\n{indent_str} " + obj.replace("\n", f"\n{indent_str} ")
|
|
else:
|
|
return f" {obj}"
|
|
|
|
elif isinstance(obj, (int, float, bool)):
|
|
return f" {obj!s}"
|
|
|
|
elif callable(obj):
|
|
source = (
|
|
"\n".join("" if "def " in line else line for line in inspect.getsource(obj).split("\n") if line.strip())
|
|
.split("lambda: ")[-1]
|
|
.rstrip(",")
|
|
)
|
|
return f" {indent_str} " + source.replace("\n", f"\n{indent_str} ")
|
|
|
|
else:
|
|
return f" {obj!s}"
|
|
|
|
|
|
def _binary_sort_key(binary: Binary) -> tuple[int, int, int, Any]:
|
|
return (
|
|
int(binary.status == Binary.StatusChoices.INSTALLED),
|
|
int(bool(binary.version)),
|
|
int(bool(binary.abspath)),
|
|
binary.modified_at,
|
|
)
|
|
|
|
|
|
def get_db_binaries_by_name() -> dict[str, Binary]:
|
|
"""Group Binary rows by a URL-safe canonical name.
|
|
|
|
Hooks occasionally emit ``BinaryEvent.name`` carrying an abspath rather
|
|
than a short binary name (see ``services/binary_service.py``). That used
|
|
to leak ``name='/Users/.../bin/foo'`` rows into the DB, which then broke
|
|
``/admin/environment/binaries`` because the admin URL regex is
|
|
``(?P<key>[^/]+)``. Canonicalize at the keying step so duplicates fold
|
|
into the real binary and the admin link key stays slash-free regardless
|
|
of legacy DB state.
|
|
"""
|
|
from archivebox.machine.models import _canonical_binary_name
|
|
|
|
grouped: dict[str, list[Binary]] = {}
|
|
binary_name_aliases = {
|
|
"youtube-dl": "yt-dlp",
|
|
}
|
|
for binary in Binary.objects.all():
|
|
canonical_name = _canonical_binary_name(binary.name)
|
|
canonical_name = binary_name_aliases.get(canonical_name, canonical_name)
|
|
if not canonical_name:
|
|
continue
|
|
grouped.setdefault(canonical_name, []).append(binary)
|
|
|
|
return {name: max(records, key=_binary_sort_key) for name, records in grouped.items()}
|
|
|
|
|
|
@render_with_table_view
|
|
def binaries_list_view(request: HttpRequest, **kwargs) -> TableContext:
|
|
assert is_superuser(request), "Must be a superuser to view configuration settings."
|
|
|
|
rows = {
|
|
"Binary Name": [],
|
|
"Found Version": [],
|
|
"Provided By": [],
|
|
"Found Abspath": [],
|
|
}
|
|
|
|
db_binaries = get_db_binaries_by_name()
|
|
all_binary_names = sorted(db_binaries.keys())
|
|
|
|
for name in all_binary_names:
|
|
binary = db_binaries.get(name)
|
|
binary_is_valid = bool(binary and binary.is_valid)
|
|
|
|
rows["Binary Name"].append(ItemLink(name, key=name))
|
|
|
|
if binary_is_valid:
|
|
rows["Found Version"].append(f"✅ {binary.version}" if binary.version else "✅ found")
|
|
rows["Provided By"].append(binary.binprovider or "-")
|
|
rows["Found Abspath"].append(binary.abspath or "-")
|
|
else:
|
|
rows["Found Version"].append("❌ missing")
|
|
rows["Provided By"].append("-")
|
|
rows["Found Abspath"].append("-")
|
|
|
|
return TableContext(
|
|
title="Binaries",
|
|
table=rows,
|
|
)
|
|
|
|
|
|
@render_with_item_view
|
|
def binary_detail_view(request: HttpRequest, key: str, **kwargs) -> ItemContext:
|
|
assert is_superuser(request), "Must be a superuser to view configuration settings."
|
|
|
|
key = {
|
|
"youtube-dl": "yt-dlp",
|
|
}.get(key, key)
|
|
db_binary = get_db_binaries_by_name().get(key)
|
|
binary_is_valid = bool(db_binary and db_binary.is_valid)
|
|
if binary_is_valid:
|
|
binary_data = db_binary.to_json()
|
|
section: SectionData = {
|
|
"name": key,
|
|
"description": mark_safe(render_binary_detail_description(key, binary_data, db_binary)),
|
|
"fields": {
|
|
"name": key,
|
|
"binprovider": db_binary.binprovider or "-",
|
|
"abspath": db_binary.abspath or "not found",
|
|
"version": db_binary.version or "unknown",
|
|
"sha256": db_binary.sha256,
|
|
"status": db_binary.status,
|
|
},
|
|
"help_texts": {},
|
|
}
|
|
return ItemContext(
|
|
slug=key,
|
|
title=key,
|
|
data=[section],
|
|
)
|
|
|
|
section: SectionData = {
|
|
"name": key,
|
|
"description": "No persisted Binary record found",
|
|
"fields": {
|
|
"name": key,
|
|
"binprovider": db_binary.binprovider if db_binary else "not recorded",
|
|
"abspath": db_binary.abspath if db_binary else "not recorded",
|
|
"version": db_binary.version if db_binary else "N/A",
|
|
"status": db_binary.status if db_binary else "unrecorded",
|
|
},
|
|
"help_texts": {},
|
|
}
|
|
return ItemContext(
|
|
slug=key,
|
|
title=key,
|
|
data=[section],
|
|
)
|
|
|
|
|
|
@render_with_table_view
|
|
def worker_list_view(request: HttpRequest, **kwargs) -> TableContext:
|
|
assert is_superuser(request), "Must be a superuser to view configuration settings."
|
|
|
|
rows = {
|
|
"Name": [],
|
|
"Type": [],
|
|
"State": [],
|
|
"PID": [],
|
|
"Started": [],
|
|
"Command": [],
|
|
"Logfile": [],
|
|
"Exit Status": [],
|
|
}
|
|
|
|
from archivebox.workers.supervisord_util import get_existing_supervisord_process
|
|
|
|
supervisor = get_existing_supervisord_process()
|
|
if supervisor is None:
|
|
return TableContext(
|
|
title="No running worker processes",
|
|
table=rows,
|
|
)
|
|
|
|
all_config: dict[str, dict[str, object]] = {}
|
|
config_items = supervisor.getAllConfigInfo()
|
|
if not isinstance(config_items, list):
|
|
config_items = []
|
|
for config_data in config_items:
|
|
if not isinstance(config_data, dict):
|
|
continue
|
|
config_name = config_data.get("name")
|
|
if not isinstance(config_name, str):
|
|
continue
|
|
all_config[config_name] = config_data
|
|
|
|
# Collect every PID we plan to show so we can resolve them to Process rows
|
|
# in a single query. supervisord's per-worker description carries the pid
|
|
# in the form ``pid 12345, uptime 0:01:23`` (or just the bare ``pid``
|
|
# placeholder when stopped); we ignore non-numeric values.
|
|
process_items = supervisor.getAllProcessInfo()
|
|
if not isinstance(process_items, list):
|
|
process_items = []
|
|
|
|
def _parse_worker_pid_and_uptime(description: str) -> tuple[int | None, str]:
|
|
body = description.replace("pid ", "", 1)
|
|
pid_part, _, uptime_part = body.partition(", uptime ")
|
|
try:
|
|
return int(pid_part.strip()), uptime_part.strip()
|
|
except ValueError:
|
|
return None, ""
|
|
|
|
pids: set[int] = set()
|
|
supervisor_pid = supervisor.getPID()
|
|
if isinstance(supervisor_pid, int):
|
|
pids.add(supervisor_pid)
|
|
for proc_data in process_items:
|
|
if not isinstance(proc_data, dict):
|
|
continue
|
|
pid_int, _ = _parse_worker_pid_and_uptime(str(proc_data.get("description") or ""))
|
|
if pid_int is not None:
|
|
pids.add(pid_int)
|
|
|
|
pid_to_process_id: dict[int, str] = {}
|
|
pid_to_process_type: dict[int, str] = {}
|
|
if pids:
|
|
try:
|
|
from archivebox.machine.models import Machine, Process
|
|
|
|
for row in (
|
|
Process.objects.filter(machine=Machine.current(), pid__in=pids)
|
|
.order_by("pid", "-started_at", "-created_at")
|
|
.only("id", "pid", "process_type")
|
|
):
|
|
if row.pid in pid_to_process_id:
|
|
continue # keep the most recent row per PID
|
|
pid_to_process_id[row.pid] = str(row.id)
|
|
pid_to_process_type[row.pid] = row.process_type
|
|
|
|
except (ImportError, RuntimeError, TypeError, ValueError):
|
|
pass
|
|
|
|
def _pid_cell(pid_value: int | None, uptime_str: str = ""):
|
|
if pid_value is None:
|
|
return ""
|
|
pid_text = str(pid_value)
|
|
process_id = pid_to_process_id.get(pid_value)
|
|
if process_id:
|
|
link = format_html('<a href="/admin/machine/process/{}/change/">{}</a>', process_id, pid_text)
|
|
else:
|
|
link = format_html("{}", pid_text)
|
|
if uptime_str:
|
|
return format_html("{}, uptime {}", link, uptime_str)
|
|
return link
|
|
|
|
# Add top row for supervisord process manager. supervisord exposes its
|
|
# state + pid over XML-RPC but not its own start time / exit status / uptime,
|
|
# so we read those from the OS process (or fall back to the Process row
|
|
# recorded in _record_supervisord_process). Exit status stays blank while
|
|
# it's RUNNING — supervisord wouldn't be answering RPC if it had exited.
|
|
rows["Name"].append(ItemLink("supervisord", key="supervisord"))
|
|
rows["Type"].append("supervisord")
|
|
supervisor_state = supervisor.getState()
|
|
state_name = str(supervisor_state.get("statename") if isinstance(supervisor_state, dict) else "")
|
|
rows["State"].append(state_name)
|
|
|
|
supervisor_started = ""
|
|
supervisor_uptime = ""
|
|
try:
|
|
import time as _time
|
|
|
|
import psutil
|
|
|
|
ps_proc = psutil.Process(supervisor_pid)
|
|
create_time = ps_proc.create_time()
|
|
supervisor_started = format_parsed_datetime(create_time)
|
|
seconds = max(int(_time.time() - create_time), 0)
|
|
hours, remainder = divmod(seconds, 3600)
|
|
minutes, secs = divmod(remainder, 60)
|
|
supervisor_uptime = f"{hours}:{minutes:02d}:{secs:02d}"
|
|
except (ImportError, RuntimeError, TypeError, ValueError):
|
|
try:
|
|
from archivebox.machine.models import Machine, Process
|
|
|
|
row = (
|
|
Process.objects.filter(
|
|
machine=Machine.current(),
|
|
process_type=Process.TypeChoices.SUPERVISORD,
|
|
pid=supervisor_pid,
|
|
)
|
|
.order_by("-started_at")
|
|
.first()
|
|
)
|
|
if row and row.started_at:
|
|
supervisor_started = row.started_at.strftime("%Y-%m-%d %H:%M:%S")
|
|
seconds = max(int((timezone.now() - row.started_at).total_seconds()), 0)
|
|
hours, remainder = divmod(seconds, 3600)
|
|
minutes, secs = divmod(remainder, 60)
|
|
supervisor_uptime = f"{hours}:{minutes:02d}:{secs:02d}"
|
|
except (RuntimeError, TypeError, ValueError):
|
|
pass
|
|
|
|
rows["PID"].append(_pid_cell(supervisor_pid if isinstance(supervisor_pid, int) else None, supervisor_uptime))
|
|
rows["Started"].append(supervisor_started or "-")
|
|
|
|
rows["Command"].append("supervisord --configuration=tmp/supervisord.conf")
|
|
rows["Logfile"].append(
|
|
format_html(
|
|
'<a href="/admin/environment/logs/{}/">{}</a>',
|
|
"supervisord",
|
|
"logs/supervisord.log",
|
|
),
|
|
)
|
|
rows["Exit Status"].append("" if state_name == "RUNNING" else "-")
|
|
|
|
# Add a row for each worker process managed by supervisord
|
|
for proc_data in process_items:
|
|
if not isinstance(proc_data, dict):
|
|
continue
|
|
proc_name = str(proc_data.get("name") or "")
|
|
proc_description = str(proc_data.get("description") or "")
|
|
proc_start = proc_data.get("start")
|
|
proc_logfile = str(proc_data.get("stdout_logfile") or "")
|
|
proc_config = all_config.get(proc_name, {})
|
|
pid_int, uptime_str = _parse_worker_pid_and_uptime(proc_description)
|
|
|
|
rows["Name"].append(ItemLink(proc_name, key=proc_name))
|
|
# Prefer the Process row's process_type when we have one (e.g. "worker",
|
|
# "hook"); otherwise fall back to the generic "worker" label since
|
|
# everything in this loop is supervisord-managed.
|
|
rows["Type"].append(pid_to_process_type.get(pid_int, "worker") if pid_int else "worker")
|
|
rows["State"].append(str(proc_data.get("statename") or ""))
|
|
rows["PID"].append(_pid_cell(pid_int, uptime_str))
|
|
rows["Started"].append(format_parsed_datetime(proc_start))
|
|
rows["Command"].append(str(proc_config.get("command") or ""))
|
|
rows["Logfile"].append(
|
|
format_html(
|
|
'<a href="/admin/environment/logs/{}/">{}</a>',
|
|
proc_logfile.split("/")[-1].split(".")[0],
|
|
proc_logfile,
|
|
),
|
|
)
|
|
rows["Exit Status"].append(str(proc_data.get("exitstatus") or ""))
|
|
|
|
return TableContext(
|
|
title="Running worker processes",
|
|
table=rows,
|
|
)
|
|
|
|
|
|
@render_with_item_view
|
|
def worker_detail_view(request: HttpRequest, key: str, **kwargs) -> ItemContext:
|
|
assert is_superuser(request), "Must be a superuser to view configuration settings."
|
|
|
|
from archivebox.workers.supervisord_util import CONFIG_FILE_NAME, get_existing_supervisord_process, get_sock_file, get_worker
|
|
|
|
SOCK_FILE = get_sock_file()
|
|
CONFIG_FILE = SOCK_FILE.parent / CONFIG_FILE_NAME
|
|
|
|
supervisor = get_existing_supervisord_process()
|
|
if supervisor is None:
|
|
return ItemContext(
|
|
slug="none",
|
|
title="error: No running supervisord process.",
|
|
data=[],
|
|
)
|
|
|
|
all_config: list[dict[str, object]] = []
|
|
config_items = supervisor.getAllConfigInfo()
|
|
if not isinstance(config_items, list):
|
|
config_items = []
|
|
for config_data in config_items:
|
|
if isinstance(config_data, dict):
|
|
all_config.append(config_data)
|
|
|
|
if key == "supervisord":
|
|
relevant_config = CONFIG_FILE.read_text()
|
|
supervisor_log = CONSTANTS.LOGS_DIR / "supervisord.log"
|
|
with supervisor_log.open("rb") as log_file:
|
|
startup_logs = log_file.read(LOG_DETAIL_TAIL_BYTES).decode("utf-8", errors="replace")
|
|
relevant_logs = _read_text_tail(supervisor_log)
|
|
startup_lines = [line for line in startup_logs.split("\n") if "RPC interface 'supervisor' initialized" in line]
|
|
start_ts = startup_lines[-1].split(",", 1)[0] if startup_lines else ""
|
|
start_dt = parse_date(start_ts)
|
|
uptime = str(timezone.now() - start_dt).split(".")[0] if start_dt else ""
|
|
supervisor_state = supervisor.getState()
|
|
|
|
proc: dict[str, object] = {
|
|
"name": "supervisord",
|
|
"pid": supervisor.getPID(),
|
|
"statename": str(supervisor_state.get("statename") if isinstance(supervisor_state, dict) else ""),
|
|
"start": start_ts,
|
|
"stop": None,
|
|
"exitstatus": "",
|
|
"stdout_logfile": "logs/supervisord.log",
|
|
"description": f"pid 000, uptime {uptime}",
|
|
}
|
|
else:
|
|
worker_data = get_worker(supervisor, key)
|
|
proc = worker_data if isinstance(worker_data, dict) else {}
|
|
relevant_config = next((config for config in all_config if config.get("name") == key), {})
|
|
log_result = supervisor.tailProcessStdoutLog(key, -LOG_DETAIL_TAIL_BYTES, LOG_DETAIL_TAIL_BYTES)
|
|
relevant_logs = str(log_result[0] if isinstance(log_result, tuple) else log_result)
|
|
|
|
section: SectionData = {
|
|
"name": key,
|
|
"description": key,
|
|
"fields": {
|
|
"Command": str(proc.get("name") or ""),
|
|
"PID": str(proc.get("pid") or ""),
|
|
"State": str(proc.get("statename") or ""),
|
|
"Started": format_parsed_datetime(proc.get("start")),
|
|
"Stopped": format_parsed_datetime(proc.get("stop")),
|
|
"Exit Status": str(proc.get("exitstatus") or ""),
|
|
"Logfile": str(proc.get("stdout_logfile") or ""),
|
|
"Uptime": str(str(proc.get("description") or "").split("uptime ", 1)[-1]),
|
|
"Config": obj_to_yaml(relevant_config) if isinstance(relevant_config, dict) else str(relevant_config),
|
|
"Recent Logs (last 100 KB)": relevant_logs,
|
|
},
|
|
"help_texts": {"Uptime": "How long the process has been running ([days:]hours:minutes:seconds)"},
|
|
}
|
|
|
|
return ItemContext(
|
|
slug=key,
|
|
title=key,
|
|
data=[section],
|
|
)
|
|
|
|
|
|
@render_with_table_view
|
|
def log_list_view(request: HttpRequest, **kwargs) -> TableContext:
|
|
assert is_superuser(request), "Must be a superuser to view configuration settings."
|
|
|
|
log_files: list[Path] = []
|
|
for logfile in sorted(CONSTANTS.LOGS_DIR.glob("*.log"), key=os.path.getmtime)[::-1]:
|
|
if isinstance(logfile, Path):
|
|
log_files.append(logfile)
|
|
|
|
rows = {
|
|
"Name": [],
|
|
"Last Updated": [],
|
|
"Size": [],
|
|
"Most Recent Lines": [],
|
|
}
|
|
|
|
# Add a row for each worker process managed by supervisord
|
|
for logfile in log_files:
|
|
st = logfile.stat()
|
|
rows["Name"].append(ItemLink("logs" + str(logfile).rsplit("/logs", 1)[-1], key=logfile.name))
|
|
rows["Last Updated"].append(format_parsed_datetime(st.st_mtime))
|
|
rows["Size"].append(f"{st.st_size // 1000} kb")
|
|
|
|
with open(logfile, "rb") as f:
|
|
try:
|
|
f.seek(-1024, os.SEEK_END)
|
|
except OSError:
|
|
f.seek(0)
|
|
last_lines = f.read().decode("utf-8", errors="replace").split("\n")
|
|
non_empty_lines = [line for line in last_lines if line.strip()]
|
|
rows["Most Recent Lines"].append(non_empty_lines[-1] if non_empty_lines else "")
|
|
|
|
return TableContext(
|
|
title="Debug Log files",
|
|
table=rows,
|
|
)
|
|
|
|
|
|
@render_with_item_view
|
|
def log_detail_view(request: HttpRequest, key: str, **kwargs) -> ItemContext:
|
|
assert is_superuser(request), "Must be a superuser to view configuration settings."
|
|
|
|
log_file = next(logfile for logfile in CONSTANTS.LOGS_DIR.glob("*.log") if key in logfile.name)
|
|
|
|
log_text = _read_text_tail(log_file)
|
|
log_stat = log_file.stat()
|
|
|
|
section: SectionData = {
|
|
"name": key,
|
|
"description": key,
|
|
"fields": {
|
|
"Path": str(log_file),
|
|
"Size": f"{log_stat.st_size // 1000} kb",
|
|
"Last Updated": format_parsed_datetime(log_stat.st_mtime),
|
|
"Tail": "\n".join(log_text.split("\n")[-20:]),
|
|
"Recent Log (last 100 KB)": log_text,
|
|
},
|
|
}
|
|
|
|
return ItemContext(
|
|
slug=key,
|
|
title=key,
|
|
data=[section],
|
|
)
|