407 lines
17 KiB
Python
407 lines
17 KiB
Python
"""Komodo как python_tool с перечислимыми действиями.
|
||
|
||
Ключ живёт в gateway и в контекст модели не попадает. В перечислении нет
|
||
`Prune*`, `Destroy*`, `DeleteServer` и терминала хоста; `exec` - это
|
||
`docker exec` в контейнер, не шелл на сервере, и контейнеры из
|
||
``exec_deny`` (periphery с docker.sock - это и есть хост) ему не даются.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import fnmatch
|
||
import json
|
||
import re
|
||
import time
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime
|
||
from typing import Any, Literal
|
||
|
||
import aiohttp
|
||
|
||
__all__ = ["Action", "Komodo"]
|
||
|
||
Action = Literal[
|
||
"status",
|
||
"stacks",
|
||
"containers",
|
||
"logs",
|
||
"search_logs",
|
||
"updates",
|
||
"update",
|
||
"deploy",
|
||
"restart",
|
||
"exec",
|
||
]
|
||
|
||
ANSI = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]")
|
||
EXIT_MARK = "__KOMODO_EXIT_CODE:"
|
||
NOT_RUNNING_LIMIT = 15
|
||
UPDATES_LIMIT = 15
|
||
UPDATES_MAX = 200
|
||
|
||
LIST_LIMIT = 500
|
||
LIST_PAGES = 20
|
||
"""Komodo режет любой ``List*`` до своего ``default_pagination_limit`` (у нас
|
||
50) и молчит об этом: ответ - просто короткий массив, неотличимый от полного.
|
||
Поэтому листаем сами, а не верим первой странице."""
|
||
|
||
RESOURCE_LISTS = (
|
||
"ListStacks",
|
||
"ListServers",
|
||
"ListRepos",
|
||
"ListResourceSyncs",
|
||
"ListProcedures",
|
||
"ListActions",
|
||
"ListAlerters",
|
||
"ListDeployments",
|
||
"ListBuilds",
|
||
)
|
||
"""Журнал операций называет ресурс id, а не именем: отсюда берём словарь."""
|
||
|
||
SCOPE_NOTE = "список ограничен правами ключа: вне их Komodo ресурсы не показывает"
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class Komodo:
|
||
url: str
|
||
key: str
|
||
secret: str
|
||
exec_deny: tuple[str, ...] = ("*periphery*", "*komodo*")
|
||
terminal: str = "beaver"
|
||
shell: str = "env PAGER=cat GIT_PAGER=cat TERM=dumb sh"
|
||
"""Terminal is a tty: without ``PAGER=cat`` psql/git open a pager and
|
||
wait for a key forever, wedging the terminal."""
|
||
timeout: float = 120.0
|
||
wait: float = 180.0
|
||
max_chars: int = 12_000
|
||
|
||
async def komodo(
|
||
self,
|
||
action: Action,
|
||
*,
|
||
stack: str | None = None,
|
||
service: str | None = None,
|
||
container: str | None = None,
|
||
command: str | None = None,
|
||
terms: str | None = None,
|
||
tail: int = 100,
|
||
update: str | None = None,
|
||
server: str | None = None,
|
||
limit: int = UPDATES_LIMIT,
|
||
operation: str | None = None,
|
||
resource: str | None = None,
|
||
) -> str:
|
||
"""Прод через Komodo: посмотреть, а по просьбе Бобра - подействовать.
|
||
|
||
action:
|
||
- status: серверы и стеки не в running.
|
||
- stacks: все стеки - сервер, состояние, есть ли незадеплоенные коммиты.
|
||
- containers: контейнеры сервера (server; без него - все серверы).
|
||
- logs: логи стека (stack, service опционально, tail строк).
|
||
- search_logs: поиск по логам стека (stack, terms - слова через пробел).
|
||
- updates: последние операции - кто, что, статус. limit записей
|
||
(по умолчанию 15, максимум 200), operation - подстрока типа операции
|
||
(deploy, sync, restart), resource - имя стека, сервера, репозитория.
|
||
Шапка говорит, какой отрезок журнала показан: журнал длинный, и
|
||
«под фильтр ничего не попало» - это не «операций больше не было».
|
||
- update: подробности одной операции по id (update).
|
||
- deploy / restart: стек целиком или один service; ждёт исхода до 3 мин.
|
||
- exec: команда внутри контейнера (container, command; server, если имя
|
||
контейнера есть на нескольких серверах). Это `docker exec` от
|
||
пользователя контейнера, свежий шелл на каждый вызов, без tty-пейджера.
|
||
Лимит ~2 мин: долгое запускай в фон с выводом в файл и читай файл.
|
||
|
||
Деплой и рестарт - только когда об этом попросили, не «заодно».
|
||
"""
|
||
match action:
|
||
case "status":
|
||
return await self._status()
|
||
case "stacks":
|
||
return await self._stacks()
|
||
case "containers":
|
||
return await self._containers(server)
|
||
case "logs":
|
||
return await self._logs(_need(stack, "stack"), service, tail)
|
||
case "search_logs":
|
||
return await self._search(
|
||
_need(stack, "stack"), _need(terms, "terms"), service
|
||
)
|
||
case "updates":
|
||
return await self._updates(limit, operation, resource)
|
||
case "update":
|
||
return await self._update(_need(update, "update"))
|
||
case "deploy":
|
||
return await self._run("DeployStack", _need(stack, "stack"), service)
|
||
case "restart":
|
||
return await self._run("RestartStack", _need(stack, "stack"), service)
|
||
case "exec":
|
||
return await self._exec(
|
||
_need(container, "container"), _need(command, "command"), server
|
||
)
|
||
|
||
async def execute(self, operation: str, params: dict[str, Any]) -> dict[str, Any]:
|
||
return await self._json("execute", {"type": operation, "params": params})
|
||
|
||
async def read(self, request: str, params: dict[str, Any]) -> Any:
|
||
return await self._json("read", {"type": request, "params": params})
|
||
|
||
async def _status(self) -> str:
|
||
servers, stacks = await asyncio.gather(
|
||
self.listing("ListServers"), self.listing("ListStacks")
|
||
)
|
||
lines = [f"{s['name']:24} {s['info']['state']}" for s in servers]
|
||
bad = [s for s in stacks if s["info"]["state"] != "running"]
|
||
lines.append(f"стеков: {len(stacks)}, не running: {len(bad)}")
|
||
lines.extend(
|
||
f" {s['info']['state']:12} {s['name']} @{s['info'].get('server_name')}"
|
||
for s in bad[:NOT_RUNNING_LIMIT]
|
||
)
|
||
if len(bad) > NOT_RUNNING_LIMIT:
|
||
lines.append(
|
||
f"показаны {NOT_RUNNING_LIMIT} из {len(bad)}, остальные скрыты лимитом"
|
||
)
|
||
lines.append(SCOPE_NOTE)
|
||
return "\n".join(lines)
|
||
|
||
async def _stacks(self) -> str:
|
||
stacks = await self.listing("ListStacks")
|
||
lines = []
|
||
for s in sorted(stacks, key=lambda s: s["name"]):
|
||
info = s["info"]
|
||
behind = info.get("latest_hash") and info.get("deployed_hash") != info.get(
|
||
"latest_hash"
|
||
)
|
||
lines.append(
|
||
f"{info['state']:12} {s['name']:28} @{info.get('server_name')}"
|
||
+ (" (есть незадеплоенные коммиты)" if behind else "")
|
||
)
|
||
lines.append(f"стеков: {len(stacks)}; {SCOPE_NOTE}")
|
||
return "\n".join(lines)
|
||
|
||
async def _containers(self, server: str | None) -> str:
|
||
names = (
|
||
[server]
|
||
if server
|
||
else [s["name"] for s in await self.listing("ListServers")]
|
||
)
|
||
lines = []
|
||
for name in names:
|
||
containers = await self.listing("ListDockerContainers", {"server": name})
|
||
lines.append(f"{name}:")
|
||
lines.extend(
|
||
f" {c['state']:10} {c['name']:36} {c.get('status', '')}"
|
||
for c in sorted(containers, key=lambda c: c["name"])
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
async def _logs(self, stack: str, service: str | None, tail: int) -> str:
|
||
log = await self.read(
|
||
"GetStackLog",
|
||
{"stack": stack, "services": [service] if service else [], "tail": tail},
|
||
)
|
||
return self._clip(_log_text(log))
|
||
|
||
async def _search(self, stack: str, terms: str, service: str | None) -> str:
|
||
log = await self.read(
|
||
"SearchStackLog",
|
||
{
|
||
"stack": stack,
|
||
"services": [service] if service else [],
|
||
"terms": terms.split(),
|
||
},
|
||
)
|
||
return self._clip(_log_text(log))
|
||
|
||
async def _updates(
|
||
self, limit: int, operation: str | None, resource: str | None
|
||
) -> str:
|
||
want = min(max(limit, 1), UPDATES_MAX)
|
||
names = await self._resource_names()
|
||
query: dict[str, Any] = {}
|
||
applied = []
|
||
if operation:
|
||
query["operation"] = {"$regex": re.escape(operation), "$options": "i"}
|
||
applied.append(f"операция ~{operation}")
|
||
if resource:
|
||
ids = [i for i, name in names.items() if resource.lower() in name.lower()]
|
||
if not ids:
|
||
return (
|
||
f"ресурса ~{resource} ключу руки не видно: "
|
||
"либо имя другое, либо он вне прав ключа"
|
||
)
|
||
query["target.id"] = {"$in": ids}
|
||
applied.append(f"ресурс ~{resource} ({len(ids)} шт)")
|
||
updates, more = await self._update_pages(query, want)
|
||
|
||
head = [f"операций: {len(updates)}"]
|
||
if updates:
|
||
head.append(
|
||
f"{_ts(updates[-1]['start_ts'])} … {_ts(updates[0]['start_ts'])}"
|
||
)
|
||
head.append("раньше в журнале есть ещё" if more else "это весь журнал")
|
||
lines = [", ".join(head)]
|
||
lines.append(
|
||
"фильтрует Komodo: " + "; ".join(applied)
|
||
if applied
|
||
else "фильтра нет: это просто хвост журнала"
|
||
)
|
||
if want != limit:
|
||
lines.append(f"запрошено {limit} записей, предел руки {want}")
|
||
unseen = sum(1 for u in updates if u["target"]["id"] not in names)
|
||
if unseen:
|
||
lines.append(
|
||
f"записей о невидимых ключу ресурсах: {unseen} - id вместо имени"
|
||
)
|
||
lines.extend(_update_line(u, names) for u in updates)
|
||
return "\n".join(lines)
|
||
|
||
async def _update_pages(
|
||
self, query: dict[str, Any], want: int
|
||
) -> tuple[list[dict[str, Any]], bool]:
|
||
"""Страница ListUpdates - ровно 100 записей, limit там не работает."""
|
||
out: list[dict[str, Any]] = []
|
||
page: int | None = 0
|
||
while page is not None and len(out) < want:
|
||
body: dict[str, Any] = {"page": page}
|
||
if query:
|
||
body["query"] = query
|
||
answer = await self.read("ListUpdates", body)
|
||
out.extend(answer["updates"])
|
||
page = answer.get("next_page")
|
||
return out[:want], len(out) > want or page is not None
|
||
|
||
async def _resource_names(self) -> dict[str, str]:
|
||
lists = await asyncio.gather(*(self.listing(r) for r in RESOURCE_LISTS))
|
||
return {r["id"]: r["name"] for rows in lists for r in rows}
|
||
|
||
async def _update(self, update_id: str) -> str:
|
||
return self._clip(
|
||
_update_detail(await self.read("GetUpdate", {"id": update_id}))
|
||
)
|
||
|
||
async def _run(self, operation: str, stack: str, service: str | None) -> str:
|
||
receipt = await self.execute(
|
||
operation, {"stack": stack, "services": [service] if service else []}
|
||
)
|
||
update_id = receipt.get("id") or receipt.get("_id", {}).get("$oid")
|
||
if not update_id:
|
||
return f"{operation} {stack}: Komodo вернул {receipt!r}"
|
||
deadline = time.monotonic() + self.wait
|
||
while time.monotonic() < deadline:
|
||
await asyncio.sleep(5)
|
||
current = await self.read("GetUpdate", {"id": update_id})
|
||
if current.get("status") == "Complete":
|
||
return self._clip(_update_detail(current))
|
||
return f"{operation} {stack}: ещё идёт, update={update_id} - проверь позже"
|
||
|
||
async def _exec(self, container: str, command: str, server: str | None) -> str:
|
||
if any(fnmatch.fnmatchcase(container, p) for p in self.exec_deny):
|
||
return f"exec в {container} запрещён: это контейнер инфраструктуры"
|
||
server = server or await self._server_of(container)
|
||
if server is None:
|
||
return f"контейнер {container} не найден ни на одном сервере"
|
||
body = {
|
||
"target": {
|
||
"type": "Container",
|
||
"params": {"server": server, "container": container},
|
||
},
|
||
"terminal": self.terminal,
|
||
"command": command,
|
||
"init": {"command": self.shell, "recreate": "Always"},
|
||
}
|
||
status, text = await self._post("terminal/execute", body)
|
||
if status != 200:
|
||
return f"exec {container}: Komodo {status}: {text[:500]}"
|
||
output, _, code = ANSI.sub("", text).rpartition(EXIT_MARK)
|
||
header = f"[{container}@{server}] exit {code.strip() or '?'}"
|
||
return f"{header}\n{self._clip(output.strip())}"
|
||
|
||
async def _server_of(self, container: str) -> str | None:
|
||
for s in await self.listing("ListServers"):
|
||
containers = await self.listing(
|
||
"ListDockerContainers", {"server": s["name"]}
|
||
)
|
||
if any(c["name"] == container for c in containers):
|
||
return s["name"]
|
||
return None
|
||
|
||
async def listing(
|
||
self, request: str, params: dict[str, Any] | None = None
|
||
) -> list[Any]:
|
||
"""Любой ``List*`` целиком: страницами, пока Komodo не отдаст неполную.
|
||
|
||
Публичная: этим же листанием ходят джобы, второго такого не пишем.
|
||
"""
|
||
out: list[Any] = []
|
||
for page in range(LIST_PAGES):
|
||
chunk = await self.read(
|
||
request, (params or {}) | {"page": page, "limit": LIST_LIMIT}
|
||
)
|
||
out.extend(chunk)
|
||
if len(chunk) < LIST_LIMIT:
|
||
break
|
||
return out
|
||
|
||
async def _json(self, route: str, body: dict[str, Any]) -> Any:
|
||
status, text = await self._post(route, body)
|
||
if status != 200:
|
||
msg = f"Komodo {route} {body.get('type')}: {status} {text[:500]}"
|
||
raise RuntimeError(msg)
|
||
return json.loads(text)
|
||
|
||
async def _post(self, route: str, body: dict[str, Any]) -> tuple[int, str]:
|
||
headers = {"X-Api-Key": self.key, "X-Api-Secret": self.secret}
|
||
async with (
|
||
aiohttp.ClientSession() as http,
|
||
http.post(
|
||
f"{self.url.rstrip('/')}/{route}",
|
||
json=body,
|
||
headers=headers,
|
||
timeout=aiohttp.ClientTimeout(total=self.timeout),
|
||
) as response,
|
||
):
|
||
return response.status, await response.text()
|
||
|
||
def _clip(self, text: str) -> str:
|
||
if len(text) <= self.max_chars:
|
||
return text
|
||
head, tail = self.max_chars * 2 // 3, self.max_chars // 3
|
||
cut = len(text) - head - tail
|
||
return f"{text[:head]}\n…[обрезано {cut} символов]…\n{text[-tail:]}"
|
||
|
||
|
||
def _need(value: str | None, name: str) -> str:
|
||
if not value:
|
||
msg = f"для этого действия нужен параметр {name}"
|
||
raise ValueError(msg)
|
||
return value
|
||
|
||
|
||
def _log_text(log: dict[str, Any]) -> str:
|
||
if "error" in log:
|
||
return f"Komodo: {log['error']}"
|
||
return ANSI.sub("", (log.get("stdout") or "") + (log.get("stderr") or "")).strip()
|
||
|
||
|
||
def _ts(ms: int) -> str:
|
||
return datetime.fromtimestamp(ms / 1000, tz=UTC).strftime("%m-%d %H:%M")
|
||
|
||
|
||
def _update_line(u: dict[str, Any], names: dict[str, str]) -> str:
|
||
ok = "ok" if u.get("success") else "FAIL"
|
||
target = names.get(u["target"]["id"], u["target"]["id"])
|
||
head = f"{_ts(u['start_ts'])} {u['operation']:20} {target:26} {u['status']:10} {ok}"
|
||
return f"{head} user={u.get('username')} id={u.get('id')}"
|
||
|
||
|
||
def _update_detail(u: dict[str, Any]) -> str:
|
||
lines = [f"{u['operation']} {u['status']} {'ok' if u.get('success') else 'FAIL'}"]
|
||
for stage in u.get("logs", []):
|
||
out = ANSI.sub(
|
||
"", (stage.get("stdout") or "") + " " + (stage.get("stderr") or "")
|
||
)
|
||
lines.append(f"-- {stage['stage']} {'ok' if stage['success'] else 'FAIL'}")
|
||
lines.append(re.sub("<[^>]+>", "", out).strip()[:1500])
|
||
return "\n".join(lines)
|