1from __future__ import annotations
2
3import argparse
4import json
5import subprocess
6from typing import Any
7
8import daggerml.api as codec_mod
9from daggerml.api import DmlRepoError
10from daggerml.contrib import adapters as areg
11from daggerml.contrib.executors import _base as ereg
12
13
14def _diag(*, scope: str, code: str, message: str) -> dict[str, str]:
15 return {
16 "severity": "error",
17 "scope": scope,
18 "code": code,
19 "message": message,
20 }
21
22
23def _fqn(obj: Any) -> str:
24 module = getattr(obj, "__module__", None) or type(obj).__module__
25 qualname = getattr(obj, "__qualname__", None) or type(obj).__qualname__
26 return f"{module}:{qualname}"
27
28
29def _implements(obj: Any, names: tuple[str, ...]) -> dict[str, bool]:
30 return {name: callable(getattr(obj, name, None)) for name in names}
31
32
33def _registration(
34 kind: str, key: str, obj: Any, diagnostics: list[dict[str, Any]] | None = None
35) -> dict[str, Any]:
36 required = {
37 "adapter": ("resolve_runnable", "send", "cli"),
38 "executor": ("resolve_runnable", "start", "poll", "cleanup", "cancel"),
39 "codec": ("can_encode", "encode"),
40 }[kind]
41 implements = _implements(obj, required)
42 missing = [name for name, implemented in implements.items() if not implemented]
43 if missing and diagnostics is not None:
44 diagnostics.append(
45 _diag(
46 scope=kind,
47 code="required_operation_missing",
48 message=f"{key} is missing required operations: {', '.join(missing)}",
49 )
50 )
51 return {
52 "key": key,
53 "fqn": _fqn(obj),
54 "effective": not missing,
55 "implements": implements,
56 }
57
58
59def _load_plugins(scope: str, load: Any, diagnostics: list[dict[str, Any]]) -> None:
60 try:
61 load()
62 except Exception as e:
63 diagnostics.append(_diag(scope=scope, code="plugin_load_failed", message=str(e)))
64
65
66def _adapter_status(diagnostics: list[dict[str, Any]]) -> list[dict[str, Any]]:
67 _load_plugins("adapter", areg.load_adapter_plugins, diagnostics)
68 with areg._LOCK:
69 items = sorted(areg._ADAPTER_SPECS.items())
70 return [_registration("adapter", name, spec, diagnostics) for name, spec in items]
71
72
73def _executor_status(diagnostics: list[dict[str, Any]]) -> list[dict[str, Any]]:
74 _load_plugins("executor", ereg.load_executor_plugins, diagnostics)
75 with ereg._LOCK:
76 items = sorted(ereg._EXECUTOR_SPECS.items())
77 return [_registration("executor", f"{adapter}:{name}", spec, diagnostics) for (adapter, name), spec in items]
78
79
80def _codec_status(diagnostics: list[dict[str, Any]]) -> list[dict[str, Any]]:
81 try:
82 codecs = list(codec_mod.iter_codecs())
83 except Exception as e:
84 diagnostics.append(_diag(scope="codec", code="plugin_load_failed", message=str(e)))
85 codecs = []
86 return [_registration("codec", str(order), codec, diagnostics) for order, codec in enumerate(codecs)]
87
88
89def status() -> dict[str, object]:
90 diagnostics: list[dict[str, Any]] = []
91 adapters = _adapter_status(diagnostics)
92 executors = _executor_status(diagnostics)
93 codecs = _codec_status(diagnostics)
94 diagnostics.sort(key=lambda item: (item["scope"], item["code"], item["message"]))
95
96 return {
97 "schema_version": 0,
98 "summary": {
99 "has_errors": bool(diagnostics),
100 "diagnostic_count": len(diagnostics),
101 "adapter_registration_count": len(adapters),
102 "adapter_effective_count": sum(item["effective"] for item in adapters),
103 "executor_registration_count": len(executors),
104 "executor_effective_count": sum(item["effective"] for item in executors),
105 "codec_registration_count": len(codecs),
106 "codec_effective_count": sum(item["effective"] for item in codecs),
107 },
108 "adapters": adapters,
109 "executors": executors,
110 "codecs": codecs,
111 "diagnostics": diagnostics,
112 }
113
114
115def cli(argv: list[str] | None = None) -> int:
116 parser = argparse.ArgumentParser(prog="dml contrib", allow_abbrev=False)
117 commands = parser.add_subparsers(dest="command", required=True)
118 commands.add_parser("status", help="Report adapter, executor, and codec registrations and diagnostics.",
119 description="Report adapter, executor, and codec registrations and diagnostics.")
120 for name, help_text in (
121 ("docker-gc", "Remove old stopped containers and unused images (DML-owned only by default)."),
122 ("install-docker-gc", "Install an hourly systemd user timer (Linux only)."),
123 ):
124 command = commands.add_parser(name, help=help_text, allow_abbrev=False)
125 command.add_argument("--all", action="store_true", help="Include non-DML resources on the Docker daemon.")
126 command.add_argument("--older-than-hours", type=float, default=24,
127 help="Minimum age in hours (default: 24; images use creation time).")
128 if name == "docker-gc":
129 command.add_argument("--dry-run", action="store_true", help="Report candidates without deleting anything.")
130 args = parser.parse_args(argv)
131 if args.command == "status":
132 result = status()
133 else:
134 from daggerml.contrib import docker_gc
135
136 try:
137 if args.command == "install-docker-gc":
138 result = {"directory": str(docker_gc.install(
139 all_resources=args.all, older_than_hours=args.older_than_hours))}
140 else:
141 result = docker_gc.collect(all_resources=args.all, older_than_hours=args.older_than_hours,
142 dry_run=args.dry_run)
143 except (DmlRepoError, ValueError, OSError, subprocess.CalledProcessError) as exc:
144 parser.exit(1, f"{parser.prog}: {exc}\n")
145 print(json.dumps(result, separators=(",", ":"), sort_keys=True))
146 return 0