1221 lines
61 KiB
Python
1221 lines
61 KiB
Python
#!/usr/bin/env python3
|
||
"""eva-agent — клиент модуля wiki хаба (sdlc-hub) для виртуалки с доступом к Eva банка.
|
||
|
||
Опрашивает хаб, выполняет команды wiki_sync и wiki_fetch (только чтение Eva), ведёт журнал на диске. Только
|
||
стандартная библиотека, Python 3.9. Общие части (HTTP, папка, журнал, цикл, самообновление, запускатель) — копия
|
||
jira-agent v14. Спека: jira-hub/openspec/changes/wiki-module/.
|
||
Связь с хабом — HTTP (`hub_url = http://…`) или общая папка (`hub_url = folder:<путь>`), если сеть закрыта VPN.
|
||
|
||
python3 agent.py check # доступ к хабу (проверка Eva добавится позже), без записи
|
||
python3 agent.py run # основной цикл (по умолчанию)
|
||
python3 agent.py selftest # самопроверка новой версии перед обновлением: версия + check; код 0 или 3
|
||
python3 agent.py prune # почистить journal.jsonl от подтверждённых записей старше 30 дней
|
||
|
||
Самообновление — через запускатель run.py (спека agent, «Самообновление агента»).
|
||
"""
|
||
import argparse
|
||
import base64
|
||
import fcntl
|
||
import hashlib
|
||
import http.client
|
||
import json
|
||
import os
|
||
import re
|
||
import signal
|
||
import ssl
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
from html.parser import HTMLParser
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Callable, Dict, List, Optional, Tuple
|
||
|
||
AGENT_VERSION = 1
|
||
EXIT_LOCKED = 73 # «журнал занят другим процессом»: для run.py это не сбой версии (design 8)
|
||
EXIT_UPDATE = 75 # «обновление готово»: run.py подменит agent.py на agent.py.new
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
|
||
|
||
def now_iso() -> str:
|
||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||
|
||
|
||
def log(msg: str) -> None:
|
||
print("[%s] %s" % (datetime.now().strftime("%H:%M:%S"), msg), flush=True)
|
||
|
||
|
||
def load_config(path: str) -> dict:
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
cfg = json.load(f)
|
||
except FileNotFoundError:
|
||
sys.exit("нет файла настроек %s — скопируйте agent.example.json в agent.json и заполните" % path)
|
||
for key in ("hub_url", "hub_token", "eva"):
|
||
if not cfg.get(key):
|
||
sys.exit("в %s нет обязательного поля %s" % (path, key))
|
||
if not isinstance(cfg["eva"], dict):
|
||
sys.exit("в %s поле eva должно быть объектом {base_url, token}" % path)
|
||
for key in ("base_url", "token"):
|
||
if not cfg["eva"].get(key):
|
||
sys.exit("в %s нет eva.%s" % (path, key))
|
||
if cfg["hub_url"].startswith("folder:") and not os.path.isabs(cfg["hub_url"][len("folder:"):]):
|
||
sys.exit("hub_url: после folder: нужен абсолютный путь общей папки, например "
|
||
"folder:/Volumes/My Shared Files/RSHB/jira-bridge")
|
||
cfg.setdefault("timeout", 60)
|
||
cfg.setdefault("poll_seconds", 5)
|
||
cfg.setdefault("verify_ssl", True)
|
||
return cfg
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Самообновление: update.json рядом с agent.py (спека agent «Самообновление агента», design 7)
|
||
# ---------------------------------------------------------------------------
|
||
def read_update_state(path: str) -> Optional[dict]:
|
||
"""Состояние обновления; нет файла или мусор — None."""
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
except FileNotFoundError:
|
||
return None
|
||
except (OSError, ValueError):
|
||
log("update.json не читается — считаю, что состояния нет")
|
||
return None
|
||
return data if isinstance(data, dict) else None
|
||
|
||
|
||
def write_update_state(path: str, state: str, from_version, to_version, sha256, error=None) -> dict:
|
||
data = {"state": state, "from_version": from_version, "to_version": to_version, "sha256": sha256,
|
||
"error": error, "at": now_iso()}
|
||
tmp = path + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False)
|
||
os.replace(tmp, path)
|
||
return data
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# HTTP
|
||
# ---------------------------------------------------------------------------
|
||
class HttpError(Exception):
|
||
def __init__(self, code: int, text: str, headers=None):
|
||
super().__init__("HTTP %d: %s" % (code, text[:300]))
|
||
self.code, self.text, self.headers = code, text, headers or {}
|
||
|
||
|
||
class RedirectAway(Exception):
|
||
"""Сервер перенаправил запрос на другой адрес (схема, хост или порт): учётные данные туда не отправляются."""
|
||
def __init__(self, host: str):
|
||
super().__init__("перенаправление на %s" % host)
|
||
self.host = host
|
||
|
||
|
||
class TooBig(Exception):
|
||
"""Ответ длиннее предела: чтение оборвано."""
|
||
|
||
|
||
def _origin(url: str) -> Tuple[str, str, Optional[int]]:
|
||
u = urllib.parse.urlsplit(url)
|
||
scheme = (u.scheme or "").lower()
|
||
return scheme, (u.hostname or "").lower(), u.port or {"http": 80, "https": 443}.get(scheme)
|
||
|
||
|
||
class _SameOriginRedirect(urllib.request.HTTPRedirectHandler):
|
||
"""Перенаправления — только в пределах адреса клиента: urllib переносит Authorization и на чужой хост."""
|
||
def __init__(self, base: str):
|
||
self.origin = _origin(base)
|
||
|
||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||
if _origin(newurl) != self.origin:
|
||
fp.close()
|
||
raise RedirectAway(urllib.parse.urlsplit(newurl).hostname or newurl)
|
||
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
||
|
||
|
||
class Http:
|
||
def __init__(self, base: str, headers: Dict[str, str], timeout: float = 60, ctx=None):
|
||
self.base, self.headers, self.timeout, self.ctx = base.rstrip("/"), headers, timeout, ctx
|
||
|
||
def __repr__(self):
|
||
return "<%s %s>" % (self.__class__.__name__, self.base)
|
||
|
||
def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any:
|
||
data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
|
||
return self.send(method, path, data, "application/json; charset=utf-8" if data is not None else None,
|
||
params, timeout)
|
||
|
||
def send(self, method: str, path: str, data: Optional[bytes], content_type: Optional[str], params=None,
|
||
timeout=None) -> Any:
|
||
"""Запрос с готовым телом (JSON или multipart); ответ — JSON. Ошибки — HttpError и ConnectionError."""
|
||
url = self.base + path
|
||
if params:
|
||
url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
|
||
req = urllib.request.Request(url, data=data, method=method)
|
||
for k, v in self.headers.items():
|
||
req.add_header(k, v)
|
||
req.add_header("Accept", "application/json")
|
||
if content_type is not None:
|
||
req.add_header("Content-Type", content_type)
|
||
try:
|
||
with self._opener().open(req, timeout=timeout or self.timeout) as resp:
|
||
raw = resp.read()
|
||
status = resp.status
|
||
except RedirectAway as e:
|
||
self._touch()
|
||
raise HttpError(302, "перенаправление на %s отклонено: учётные данные туда не отправляются" % e.host) from None
|
||
except urllib.error.HTTPError as e:
|
||
self._touch()
|
||
try:
|
||
headers = {k.lower(): v for k, v in e.headers.items()}
|
||
text = e.read().decode("utf-8", "replace")
|
||
except (OSError, ConnectionError, TimeoutError, http.client.HTTPException) as err:
|
||
raise ConnectionError("%s: обрыв при чтении ответа: %s" % (self.base, err)) from None
|
||
finally:
|
||
e.close()
|
||
raise HttpError(e.code, text, headers) from None
|
||
except (urllib.error.URLError, ConnectionError, TimeoutError, OSError, http.client.HTTPException) as e:
|
||
raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None
|
||
self._touch()
|
||
try:
|
||
return json.loads(raw.decode("utf-8")) if raw else {}
|
||
except ValueError as e: # включая UnicodeDecodeError и JSONDecodeError
|
||
raise HttpError(status, "ответ не JSON: %s; начало: %s" % (e, raw[:200].decode("utf-8", "replace"))) from None
|
||
|
||
def _opener(self):
|
||
"""Открыватель запросов: перенаправления — только на адрес клиента (urllib переносит Authorization и на
|
||
чужой хост); контекст TLS — из настроек."""
|
||
handlers = [_SameOriginRedirect(self.base)] # type: List[Any]
|
||
if self.ctx is not None:
|
||
handlers.append(urllib.request.HTTPSHandler(context=self.ctx))
|
||
return urllib.request.build_opener(*handlers)
|
||
|
||
def fetch_bytes(self, path: str, limit: int, timeout=None) -> bytes:
|
||
"""GET с ответом-байтами (содержимое вложения): без разбора JSON, читает не больше limit + 1 байт — длиннее
|
||
limit — TooBig. Перенаправление на другой адрес — RedirectAway без перехода. Ошибки — как у send."""
|
||
req = urllib.request.Request(self.base + path, method="GET")
|
||
for k, v in self.headers.items():
|
||
req.add_header(k, v)
|
||
req.add_header("Accept", "*/*")
|
||
try:
|
||
with self._opener().open(req, timeout=timeout or self.timeout) as resp:
|
||
data = resp.read(limit + 1)
|
||
rest = resp.length # остаток по Content-Length: read(n) при обрыве молча отдаёт меньше
|
||
if rest and len(data) <= limit:
|
||
raise ConnectionError("%s: обрыв при чтении ответа: получено %d байт, не дошло %d" % (self.base, len(data), rest))
|
||
except RedirectAway:
|
||
self._touch()
|
||
raise
|
||
except urllib.error.HTTPError as e:
|
||
self._touch()
|
||
try:
|
||
headers = {k.lower(): v for k, v in e.headers.items()}
|
||
text = e.read().decode("utf-8", "replace")
|
||
except (OSError, ConnectionError, TimeoutError, http.client.HTTPException) as err:
|
||
raise ConnectionError("%s: обрыв при чтении ответа: %s" % (self.base, err)) from None
|
||
finally:
|
||
e.close()
|
||
raise HttpError(e.code, text, headers) from None
|
||
except (urllib.error.URLError, ConnectionError, TimeoutError, OSError, http.client.HTTPException) as e:
|
||
raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None
|
||
self._touch()
|
||
if len(data) > limit:
|
||
raise TooBig("ответ длиннее %d байт" % limit)
|
||
return data
|
||
|
||
def _touch(self) -> None:
|
||
"""Вызывается на каждый полученный HTTP-ответ (любой код)."""
|
||
|
||
|
||
class EvaError(Exception):
|
||
"""JSON-RPC Eva вернула `error`: текст «<метод>: <error>»."""
|
||
|
||
|
||
class EvaClient(Http):
|
||
"""Клиент Eva (JSON-RPC 2.2): повторы, пауза между вызовами, счётчики вызовов и 429, время последнего вызова."""
|
||
PAGE_SIZE = 200
|
||
RETRY_CODES = (429, 502, 503, 504)
|
||
ATTEMPTS = 4
|
||
MAX_RETRY_AFTER = 60
|
||
|
||
def __init__(self, cfg: dict):
|
||
e = cfg["eva"]
|
||
ctx = None
|
||
if not cfg.get("verify_ssl", True):
|
||
ctx = ssl.create_default_context()
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = ssl.CERT_NONE
|
||
elif cfg.get("ca_bundle"):
|
||
ctx = ssl.create_default_context(cafile=cfg["ca_bundle"])
|
||
super().__init__(e["base_url"], {"Authorization": "Bearer " + e["token"]}, cfg.get("timeout", 60), ctx)
|
||
self._token = e["token"]
|
||
self.calls = 0
|
||
self.rate_limited = 0
|
||
self.pacing = 0.0 # с между вызовами; ставит Runner из wiki_pacing
|
||
self.last_call_at = None # type: Optional[float] (time.monotonic)
|
||
self.last_call_iso = None # type: Optional[str]
|
||
self.sleep = time.sleep
|
||
|
||
def _touch(self) -> None:
|
||
self.calls += 1
|
||
self.last_call_at, self.last_call_iso = time.monotonic(), now_iso()
|
||
|
||
def reset_counters(self) -> None:
|
||
self.calls, self.rate_limited = 0, 0
|
||
|
||
def _scrub(self, text: str) -> str:
|
||
return text.replace(self._token, "***") if self._token else text
|
||
|
||
def _retrying(self, fn: Callable[[], Any]) -> Any:
|
||
"""Вызов с паузой между вызовами и повторами: 429/502/503/504 и сетевые сбои, 4 попытки, ожидание 1, 2, 4 с
|
||
(при 429 — Retry-After, не больше 60 с). Тексты ошибок очищаются от токена."""
|
||
for attempt in range(self.ATTEMPTS):
|
||
if self.pacing > 0 and self.last_call_at is not None:
|
||
wait = self.pacing - (time.monotonic() - self.last_call_at)
|
||
if wait > 0:
|
||
self.sleep(wait)
|
||
wait = float(2 ** attempt)
|
||
try:
|
||
return fn()
|
||
except HttpError as e:
|
||
if e.code not in self.RETRY_CODES:
|
||
raise HttpError(e.code, self._scrub(e.text), e.headers) from None
|
||
if e.code == 429:
|
||
self.rate_limited += 1
|
||
try:
|
||
wait = min(max(float(e.headers.get("retry-after", wait)), 0.0), self.MAX_RETRY_AFTER)
|
||
except ValueError:
|
||
pass
|
||
if attempt == self.ATTEMPTS - 1:
|
||
raise HttpError(e.code, self._scrub(e.text), e.headers) from None
|
||
except ConnectionError as e:
|
||
if attempt == self.ATTEMPTS - 1:
|
||
raise ConnectionError(self._scrub(str(e))) from None
|
||
self.sleep(wait)
|
||
raise AssertionError("недостижимо")
|
||
|
||
def call(self, method: str, **kwargs) -> Any: # type: ignore[override]
|
||
"""Метод JSON-RPC: POST <base>/api/?m=<метод>. `error` в ответе — EvaError."""
|
||
body = {"jsonrpc": "2.2", "method": method, "callid": str(uuid.uuid4()), "kwargs": kwargs}
|
||
data = self._retrying(lambda: Http.call(self, "POST", "/api/", body, {"m": method}))
|
||
if isinstance(data, dict) and data.get("error"):
|
||
raise EvaError(self._scrub("%s: %s" % (method, data["error"])))
|
||
return data.get("result") if isinstance(data, dict) else None
|
||
|
||
def get_doc(self, code: str, fields: List[str]) -> Optional[dict]:
|
||
return self.call("CmfDocument.get", filter=[["code", "==", code]], fields=fields)
|
||
|
||
def list_docs(self, flt: list, fields: List[str]) -> List[dict]:
|
||
out, offset = [], 0 # type: List[dict], int
|
||
while True:
|
||
chunk = self.call("CmfDocument.list", filter=flt, fields=fields,
|
||
slice=[offset, offset + self.PAGE_SIZE]) or []
|
||
if isinstance(chunk, dict): # обёрнутый результат
|
||
chunk = chunk.get("items") or chunk.get("result") or []
|
||
out.extend(chunk)
|
||
if len(chunk) < self.PAGE_SIZE:
|
||
return out
|
||
offset += self.PAGE_SIZE
|
||
|
||
def download(self, url: str, limit: int) -> bytes:
|
||
"""Файл Eva потоком: читается не больше limit + 1 байта, длиннее limit — TooBig. Адрес — только своего хоста."""
|
||
if _origin(url) != _origin(self.base) or not url.startswith(self.base):
|
||
raise ValueError("чужой хост")
|
||
return self._retrying(lambda: self.fetch_bytes(url[len(self.base):], limit))
|
||
|
||
|
||
class _LinkParser(HTMLParser):
|
||
"""Собирает img[src] и a[href] в порядке появления: (тег, значение атрибута после unescape, alt/title, текст ссылки)."""
|
||
|
||
def __init__(self):
|
||
HTMLParser.__init__(self, convert_charrefs=True)
|
||
self.items = [] # [tag, ref, hint, text]
|
||
self._open = None # запись открытой <a>, в неё копится текст
|
||
|
||
def handle_starttag(self, tag, attrs):
|
||
a = dict(attrs)
|
||
if tag == "img":
|
||
ref = a.get("src")
|
||
if ref:
|
||
self.items.append(["img", ref, a.get("alt") or a.get("title") or "", ""])
|
||
elif tag == "a":
|
||
ref = a.get("href")
|
||
self._open = None
|
||
if ref:
|
||
self._open = ["a", ref, a.get("title") or "", ""]
|
||
self.items.append(self._open)
|
||
|
||
def handle_startendtag(self, tag, attrs):
|
||
if tag == "img":
|
||
self.handle_starttag(tag, attrs)
|
||
|
||
def handle_endtag(self, tag):
|
||
if tag == "a":
|
||
self._open = None
|
||
|
||
def handle_data(self, data):
|
||
if self._open is not None:
|
||
self._open[3] += data
|
||
|
||
|
||
_PAGE_CODE = re.compile(r"DOC-\d+", re.IGNORECASE)
|
||
_THUMB_TAIL = ".meta/thumbnail.jpg"
|
||
|
||
|
||
def classify_links(html: str, base_url: str) -> Tuple[list, list]:
|
||
"""Вложения и ссылки на страницы из HTML страницы Eva (design 6–7). Порядок появления, повторы не склеиваются.
|
||
|
||
ref — значение атрибута после unescape (design 12). Чужой хост и static/ не передаются."""
|
||
parser = _LinkParser()
|
||
parser.feed(html or "")
|
||
parser.close()
|
||
base = urllib.parse.urlsplit(base_url)
|
||
attachments, links = [], []
|
||
for _tag, ref, hint, text in parser.items:
|
||
parts = urllib.parse.urlsplit(urllib.parse.urljoin(base_url.rstrip("/") + "/", ref.strip()))
|
||
if (parts.scheme, parts.netloc) != (base.scheme, base.netloc):
|
||
continue
|
||
path = parts.path
|
||
if path.startswith("/files/"):
|
||
if path.endswith(_THUMB_TAIL):
|
||
path = path[:-len(_THUMB_TAIL)]
|
||
name = " ".join(hint.split()) or " ".join(text.split()) or urllib.parse.unquote(path.rsplit("/", 1)[-1])
|
||
url = urllib.parse.urlunsplit((parts.scheme, parts.netloc, path, "", ""))
|
||
attachments.append({"ref": ref, "url": url, "name": name})
|
||
continue
|
||
m = _PAGE_CODE.search(path)
|
||
if m:
|
||
links.append({"ref": ref, "code": m.group(0).upper()})
|
||
return attachments, links
|
||
|
||
|
||
class FolderTransport:
|
||
"""Обмен с хабом через общую папку (hub_url = folder:<путь>), когда сеть до хаба закрыта (VPN на виртуалке).
|
||
Запрос — файл req/<id>.json, ответ кладёт реле хаба в resp/<id>.json. Ошибки те же, что у Http.call:
|
||
нет ответа к сроку или нет папки — ConnectionError, код не 2xx — HttpError. Спека: specs/agent, folder-transport."""
|
||
POLL = 0.1
|
||
BAD_JSON_WAIT = 2.0
|
||
|
||
def __init__(self, root: str, authorization: str):
|
||
self.root, self.authorization = root, authorization
|
||
|
||
def __repr__(self):
|
||
return "<FolderTransport %s>" % self.root
|
||
|
||
def call(self, method: str, path: str, body=None, timeout: float = 60) -> Any:
|
||
if not os.path.isdir(self.root):
|
||
raise ConnectionError("нет папки %s" % self.root)
|
||
req_dir, resp_dir = os.path.join(self.root, "req"), os.path.join(self.root, "resp")
|
||
rid = uuid.uuid4().hex
|
||
try:
|
||
os.makedirs(req_dir, exist_ok=True)
|
||
os.makedirs(resp_dir, exist_ok=True)
|
||
deadline = time.time() + timeout
|
||
self._write(req_dir, rid + ".json", {
|
||
"id": rid, "method": method, "path": path, "body": body, "authorization": self.authorization,
|
||
"deadline": datetime.fromtimestamp(deadline, timezone.utc).isoformat()})
|
||
except OSError as e:
|
||
raise ConnectionError("папка %s недоступна: %s" % (self.root, e)) from None
|
||
resp_path = os.path.join(resp_dir, rid + ".json")
|
||
end = time.monotonic() + timeout
|
||
while time.monotonic() < end:
|
||
if os.path.exists(resp_path):
|
||
return self._take(resp_path)
|
||
time.sleep(self.POLL)
|
||
try:
|
||
os.remove(os.path.join(req_dir, rid + ".json"))
|
||
except OSError:
|
||
pass
|
||
raise ConnectionError("не ответил через папку %s за %.0f с" % (self.root, timeout))
|
||
|
||
@staticmethod
|
||
def _write(directory: str, name: str, payload: dict) -> None:
|
||
tmp = os.path.join(directory, "." + name + ".tmp")
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(payload, f, ensure_ascii=False)
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
os.replace(tmp, os.path.join(directory, name))
|
||
|
||
def _take(self, path: str) -> Any:
|
||
end = time.monotonic() + self.BAD_JSON_WAIT
|
||
while True:
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
resp = json.load(f)
|
||
if not isinstance(resp, dict):
|
||
raise ValueError("ответ не объект")
|
||
status, body = resp.get("status"), resp.get("body")
|
||
if type(status) is not int or not (body is None or isinstance(body, str)):
|
||
raise ValueError("неверные типы status/body") # bool — подкласс int, поэтому type(), не isinstance (M-2 agent#8)
|
||
break
|
||
except (ValueError, OSError):
|
||
if time.monotonic() >= end:
|
||
raise ConnectionError("нечитаемый ответ хаба в %s" % path) from None
|
||
time.sleep(self.POLL)
|
||
try:
|
||
os.remove(path)
|
||
except OSError:
|
||
pass
|
||
text = body or "" # status и body проверены выше (M-5 #18)
|
||
if not 200 <= status < 300:
|
||
raise HttpError(status, text)
|
||
try:
|
||
return json.loads(text) if text else {}
|
||
except ValueError as e:
|
||
raise HttpError(status, "ответ не JSON: %s; начало: %s" % (e, text[:200])) from None
|
||
|
||
|
||
class HubClient(Http):
|
||
def __init__(self, cfg: dict):
|
||
auth = "Bearer " + cfg["hub_token"]
|
||
super().__init__(cfg["hub_url"], {"Authorization": auth}, cfg.get("timeout", 60))
|
||
url = cfg["hub_url"]
|
||
self.folder = FolderTransport(url[len("folder:"):], auth) if url.startswith("folder:") else None
|
||
|
||
def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any:
|
||
if self.folder is None:
|
||
return super().call(method, path, body, params, timeout)
|
||
if params:
|
||
path += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
|
||
return self.folder.call(method, path, body, timeout or self.timeout)
|
||
|
||
def claim(self, agent_version, settings_version, last_call_at, update_state=None, launcher_version=None) -> dict:
|
||
return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version,
|
||
"last_call_at": last_call_at, "update_state": update_state,
|
||
"launcher_version": launcher_version}, timeout=max(self.timeout, 45))
|
||
|
||
def release(self) -> dict:
|
||
return self.call("GET", "/agent/release")
|
||
|
||
def settings(self) -> dict:
|
||
return self.call("GET", "/agent/settings")
|
||
|
||
def command(self, command_id) -> dict:
|
||
return self.call("GET", "/agent/commands/%d" % int(command_id))
|
||
|
||
def result(self, command_id, payload: dict) -> dict:
|
||
return self.call("POST", "/agent/commands/%d/result" % int(command_id), payload)
|
||
|
||
def upload_file(self, data: bytes) -> dict:
|
||
"""Файл, скачанный из Eva, — в хранилище хаба: {file (sha256), size}."""
|
||
return self.call("POST", "/agent/files", {"content": base64.b64encode(data).decode("ascii")})
|
||
|
||
def wiki_pages(self, command_id, pages: list) -> dict:
|
||
"""Пачка страниц wiki_sync в стадию хаба (до 50 страниц, тело до 16 МБ)."""
|
||
return self.call("POST", "/agent/wiki/pages", {"command_id": command_id, "pages": pages})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Журнал: только дозапись; started → finished → acked по id команды
|
||
# ---------------------------------------------------------------------------
|
||
class Journal:
|
||
def __init__(self, path: str):
|
||
self.path = path
|
||
|
||
def _torn_tail(self) -> bool:
|
||
"""Файл не пуст и не заканчивается переводом строки (оборванная запись)."""
|
||
try:
|
||
if os.path.getsize(self.path) == 0:
|
||
return False
|
||
except FileNotFoundError:
|
||
return False
|
||
with open(self.path, "rb") as f:
|
||
f.seek(-1, os.SEEK_END)
|
||
return f.read(1) != b"\n"
|
||
|
||
def append(self, record: dict) -> None:
|
||
record = dict(record)
|
||
record.setdefault("at", now_iso())
|
||
data = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8")
|
||
if self._torn_tail():
|
||
data = b"\n" + data
|
||
fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
|
||
with os.fdopen(fd, "ab") as f:
|
||
f.write(data)
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
|
||
def _scan(self):
|
||
"""Возвращает (годные записи, номера битых строк). Годная — dict с command_id."""
|
||
good, bad = [], []
|
||
if not os.path.exists(self.path):
|
||
return good, bad
|
||
with open(self.path, "rb") as f:
|
||
for n, raw in enumerate(f, 1):
|
||
line = raw.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
r = json.loads(line.decode("utf-8"))
|
||
except ValueError:
|
||
r = None
|
||
if isinstance(r, dict) and "command_id" in r:
|
||
good.append(r)
|
||
else:
|
||
bad.append(n)
|
||
return good, bad
|
||
|
||
def records(self) -> List[dict]:
|
||
good, bad = self._scan()
|
||
for n in bad:
|
||
log("журнал: строка %d не читается, пропускаю" % n)
|
||
return good
|
||
|
||
def last_at(self) -> Optional[datetime]:
|
||
"""Самое позднее время записи журнала (UTC). Записи без времени или с непонятным временем пропускаются."""
|
||
latest = None
|
||
for r in self.records():
|
||
try:
|
||
t = datetime.fromisoformat(r["at"])
|
||
except (KeyError, TypeError, ValueError):
|
||
continue
|
||
if t.tzinfo is None:
|
||
continue
|
||
if latest is None or t > latest:
|
||
latest = t
|
||
return latest
|
||
|
||
def claimed(self, command_id) -> None:
|
||
"""Команда получена из claim (до паузы): с этого момента хаб считает её выданной агенту."""
|
||
self.append({"kind": "claimed", "command_id": int(command_id)})
|
||
|
||
def started(self, command_id) -> None:
|
||
self.append({"kind": "started", "command_id": int(command_id)})
|
||
|
||
def finished(self, command_id, status, result, error, jira_calls, rate_limited) -> None:
|
||
self.append({"kind": "finished", "command_id": int(command_id), "status": status, "result": result,
|
||
"error": error, "jira_calls": int(jira_calls or 0), "rate_limited": int(rate_limited or 0)})
|
||
|
||
def acked(self, command_id) -> None:
|
||
self.append({"kind": "acked", "command_id": int(command_id)})
|
||
|
||
def unacked_finished(self) -> List[dict]:
|
||
last, acked = {}, set()
|
||
for r in self.records():
|
||
if r.get("kind") == "finished":
|
||
last[r["command_id"]] = r
|
||
acked.discard(r["command_id"])
|
||
elif r.get("kind") == "acked":
|
||
acked.add(r["command_id"])
|
||
return [r for cid, r in last.items() if cid not in acked]
|
||
|
||
def unfinished_started(self) -> List[int]:
|
||
"""Порядковый учёт: claimed/started открывают id, finished/acked закрывает; порядок — первое открытие."""
|
||
opened = {} # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия
|
||
for r in self.records():
|
||
kind, cid = r.get("kind"), r["command_id"]
|
||
if kind in ("claimed", "started"):
|
||
opened[cid] = True
|
||
elif kind in ("finished", "acked") and cid in opened:
|
||
opened[cid] = False
|
||
return [cid for cid, is_open in opened.items() if is_open]
|
||
|
||
def prune(self, days: int = 30) -> int:
|
||
"""Удалить записи id до его последнего acked, если тот старше days дней (возраст — по acked)
|
||
и после него у id нет started/finished. Возвращает число удалённых годных записей; битые строки
|
||
отбрасываются при перезаписи и учитываются в логе."""
|
||
good, bad = self._scan()
|
||
if bad:
|
||
log("журнал: удаляю битых строк %d" % len(bad))
|
||
cutoff = datetime.now(timezone.utc).timestamp() - days * 86400
|
||
last_ack, last_open = {}, {}
|
||
for i, r in enumerate(good):
|
||
cid = r["command_id"]
|
||
if r.get("kind") == "acked":
|
||
last_ack[cid] = i
|
||
elif r.get("kind") in ("claimed", "started", "finished"):
|
||
last_open[cid] = i
|
||
drop_upto = {} # command_id -> индекс последнего acked: всё до него включительно удаляется
|
||
for cid, i in last_ack.items():
|
||
if last_open.get(cid, -1) > i:
|
||
continue
|
||
try:
|
||
ts = _parse_at(good[i].get("at"))
|
||
except (ValueError, TypeError):
|
||
continue
|
||
if ts < cutoff:
|
||
drop_upto[cid] = i
|
||
keep = [r for i, r in enumerate(good)
|
||
if not (r["command_id"] in drop_upto and i <= drop_upto[r["command_id"]])]
|
||
tmp = self.path + ".tmp"
|
||
try:
|
||
os.remove(tmp)
|
||
except FileNotFoundError:
|
||
pass
|
||
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||
with os.fdopen(fd, "wb") as f:
|
||
for r in keep:
|
||
f.write((json.dumps(r, ensure_ascii=False) + "\n").encode("utf-8"))
|
||
f.flush()
|
||
os.fsync(f.fileno())
|
||
os.replace(tmp, self.path)
|
||
return len(good) - len(keep)
|
||
|
||
|
||
def _parse_at(value) -> float:
|
||
"""Метка времени журнала в эпоху; наивное время трактуется как UTC."""
|
||
dt = datetime.fromisoformat(value)
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.timestamp()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Исполнитель: команда wiki_sync / wiki_fetch (читает Eva, отдаёт хабу)
|
||
# ---------------------------------------------------------------------------
|
||
class HubError(Exception):
|
||
"""Ошибка хаба внутри команды (стадия страниц, загрузка файла): источник — хаб, не Eva (из jira-agent, M-6 #18).
|
||
Повтор — только при недоступности хаба и 5xx; команды wiki Eva только читают, повтор безопасен."""
|
||
|
||
def __init__(self, cause: Exception):
|
||
super().__init__("хаб: %s" % cause)
|
||
self.retryable = isinstance(cause, ConnectionError) or (isinstance(cause, HttpError) and cause.code >= 500)
|
||
|
||
|
||
class Executor:
|
||
def __init__(self, eva: EvaClient, hub: Optional[HubClient], settings: dict):
|
||
self.eva, self.hub, self.s = eva, hub, settings
|
||
self.sleep = time.sleep
|
||
|
||
def run(self, cmd: dict) -> dict:
|
||
self.eva.reset_counters()
|
||
handler = getattr(self, "do_" + str(cmd.get("type")), None)
|
||
if handler is None:
|
||
return self._out("failed", error="неизвестный тип команды %s" % cmd.get("type"))
|
||
try:
|
||
return handler(cmd.get("params") or {}, cmd)
|
||
except HubError as e:
|
||
out = self._out("failed", error=str(e))
|
||
out["retryable"] = e.retryable
|
||
out["peer"] = "хабом"
|
||
return out
|
||
except HttpError as e:
|
||
out = self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500]))
|
||
out["retryable"] = e.code >= 500 # исход неизвестен: Runner повторит
|
||
return out
|
||
except ConnectionError as e:
|
||
out = self._out("failed", error=str(e))
|
||
out["retryable"] = True
|
||
return out
|
||
|
||
def _out(self, status, result=None, error=None) -> dict:
|
||
return {"status": status, "result": result, "error": error,
|
||
"jira_calls": self.eva.calls, "rate_limited": self.eva.rate_limited} # имена полей — протокол хаба
|
||
|
||
def _hub(self, fn, *args):
|
||
"""Вызов хаба из команды: его сбой — HubError, чтобы не выдать его за обрыв связи с Eva."""
|
||
try:
|
||
return fn(*args)
|
||
except (ConnectionError, HttpError) as e:
|
||
raise HubError(e) from None
|
||
|
||
def do_wiki_sync(self, p: dict, cmd: dict) -> dict:
|
||
return self._out("failed", error="wiki_sync: не реализовано")
|
||
|
||
def do_wiki_fetch(self, p: dict, cmd: dict) -> dict:
|
||
return self._out("failed", error="wiki_fetch: не реализовано")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте
|
||
# ---------------------------------------------------------------------------
|
||
class Updater:
|
||
"""Самообновление агента (спека agent «Самообновление агента», design 7–12). Работает только под run.py."""
|
||
RETRY_SECONDS = 300 # временный сбой подготовки — повтор не раньше (R-8)
|
||
HINT_SECONDS = 3600 # подсказка «запускайте через run.py» — не чаще (R-10)
|
||
SELFTEST_TIMEOUT = 120
|
||
VERSION_RE = re.compile(r"^AGENT_VERSION = (\d+)[ \t]*\r?$", re.M)
|
||
SHA_RE = re.compile(r"[0-9a-f]{64}")
|
||
|
||
def __init__(self, hub, base_dir: str, launcher_version: Optional[int], enabled: bool = True,
|
||
config_path: Optional[str] = None):
|
||
self.hub, self.dir, self.launcher_version, self.enabled = hub, base_dir, launcher_version, enabled
|
||
self.config_path = config_path or os.path.join(base_dir, "agent.json")
|
||
self.state_path = os.path.join(base_dir, "update.json")
|
||
self.new_path = os.path.join(base_dir, "agent.py.new")
|
||
self.agent_path = os.path.join(base_dir, "agent.py")
|
||
self.retry_after = {} # (version, sha256) -> clock(), раньше которого не повторять
|
||
self.hints = {} # вид подсказки -> clock() последней записи в лог (не чаще HINT_SECONDS)
|
||
self.clock = time.monotonic
|
||
self.should_stop = lambda: False
|
||
self.run_selftest = self._run_selftest
|
||
|
||
def state(self) -> Optional[dict]:
|
||
return read_update_state(self.state_path)
|
||
|
||
def confirm(self) -> None:
|
||
"""Первый успешный claim новой версии: trial этой версии → ok (запускатель перестаёт ждать отката)."""
|
||
st = self.state()
|
||
if st and st.get("state") == "trial" and st.get("to_version") == AGENT_VERSION:
|
||
try:
|
||
write_update_state(self.state_path, "ok", st.get("from_version"), AGENT_VERSION, st.get("sha256"))
|
||
except OSError as e: # диск полон — не падать; trial остался, повтор на следующем claim (M-3)
|
||
log("не удалось записать update.json: %s" % e)
|
||
return
|
||
log("обновление до v%d подтверждено" % AGENT_VERSION)
|
||
|
||
def _hint(self, kind: str, msg: str) -> None:
|
||
now = self.clock()
|
||
at = self.hints.get(kind)
|
||
if at is None or now - at >= self.HINT_SECONDS:
|
||
self.hints[kind] = now
|
||
log(msg)
|
||
|
||
def check(self, release) -> bool:
|
||
"""Вызывается после claim без команды. True — agent.py.new проверен, update.json = staged: run выходит с 75."""
|
||
if not self.enabled or not isinstance(release, dict):
|
||
return False
|
||
version, sha = release.get("version"), release.get("sha256")
|
||
if type(version) is not int or not isinstance(sha, str) or not self.SHA_RE.fullmatch(sha):
|
||
self._hint("format", "хаб прислал неполную публикацию: version=%.80r, sha256=%.80r" % (version, sha)) # design 10
|
||
return False
|
||
if version <= AGENT_VERSION:
|
||
return False
|
||
st = self.state()
|
||
if st and st.get("state") in ("rejected", "rolled_back") and st.get("to_version") == version \
|
||
and st.get("sha256") == sha:
|
||
return False
|
||
if self.launcher_version is None:
|
||
self._hint("launcher", "доступна v%d; для самообновления запускайте через run.py" % version)
|
||
return False
|
||
until = self.retry_after.get((version, sha))
|
||
if until is not None and self.clock() < until:
|
||
return False
|
||
return self.prepare(version, sha)
|
||
|
||
def prepare(self, version: int, sha: str) -> bool:
|
||
log("доступна v%d — готовлю обновление" % version)
|
||
try:
|
||
rel = self.hub.release()
|
||
except (ConnectionError, HttpError) as e:
|
||
return self._transient(version, sha, "скачивание: %s" % e)
|
||
content = rel.get("content") if isinstance(rel, dict) else None
|
||
if not isinstance(content, str) or rel.get("version") != version or rel.get("sha256") != sha:
|
||
return self._transient(version, sha, "публикация сменилась между запросами")
|
||
data = content.encode("utf-8")
|
||
if hashlib.sha256(data).hexdigest() != sha:
|
||
return self._reject(version, sha, "sha256 не сходится")
|
||
try:
|
||
compile(content, "agent.py.new", "exec")
|
||
except (SyntaxError, ValueError) as e:
|
||
return self._reject(version, sha, "не компилируется: %s" % e)
|
||
if self.VERSION_RE.findall(content) != [str(version)]:
|
||
return self._reject(version, sha, "AGENT_VERSION в файле не %d" % version)
|
||
tmp = self.new_path + ".tmp"
|
||
try: # диск полон, нет прав — временный сбой, не трейсбек (design 9)
|
||
with open(tmp, "wb") as f:
|
||
f.write(data)
|
||
try:
|
||
os.chmod(tmp, os.stat(self.agent_path).st_mode & 0o777)
|
||
except OSError:
|
||
pass
|
||
os.replace(tmp, self.new_path)
|
||
except OSError as e:
|
||
return self._transient(version, sha, "файлы: %s" % e)
|
||
if self.should_stop():
|
||
self._drop_new()
|
||
return False
|
||
try:
|
||
code, out = self.run_selftest(self.new_path)
|
||
except OSError as e:
|
||
return self._transient(version, sha, "selftest: %s" % e)
|
||
if self.should_stop(): # Ctrl-C уронил и selftest — это не отказ версии
|
||
self._drop_new()
|
||
return False
|
||
if code == 0:
|
||
try:
|
||
write_update_state(self.state_path, "staged", AGENT_VERSION, version, sha)
|
||
except OSError as e:
|
||
return self._transient(version, sha, "update.json: %s" % e)
|
||
log("v%d готова, перезапуск через run.py" % version)
|
||
return True
|
||
if code is None:
|
||
return self._transient(version, sha, "selftest: таймаут %d с" % self.SELFTEST_TIMEOUT)
|
||
if code == 3:
|
||
return self._transient(version, sha, "selftest: хаб или Eva недоступны")
|
||
return self._reject(version, sha, "selftest: код %s: %s" % (code, out[-500:]))
|
||
|
||
def _run_selftest(self, path: str):
|
||
try:
|
||
p = subprocess.run([sys.executable, path, "--config", self.config_path, "selftest"], cwd=self.dir,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=self.SELFTEST_TIMEOUT)
|
||
except subprocess.TimeoutExpired:
|
||
return None, ""
|
||
return p.returncode, p.stdout.decode("utf-8", "replace")
|
||
|
||
def _drop_new(self) -> None:
|
||
for path in (self.new_path, self.new_path + ".tmp"):
|
||
try:
|
||
os.remove(path)
|
||
except OSError:
|
||
pass
|
||
|
||
def _transient(self, version, sha, reason) -> bool:
|
||
self._drop_new()
|
||
self.retry_after[(version, sha)] = self.clock() + self.RETRY_SECONDS
|
||
log("обновление до v%d отложено на %d мин: %s" % (version, self.RETRY_SECONDS // 60, reason))
|
||
return False
|
||
|
||
def _reject(self, version, sha, reason) -> bool:
|
||
self._drop_new()
|
||
try:
|
||
write_update_state(self.state_path, "rejected", AGENT_VERSION, version, sha, reason)
|
||
except OSError as e:
|
||
log("не удалось записать update.json: %s" % e)
|
||
log("v%d отвергнута: %s" % (version, reason))
|
||
return False
|
||
|
||
|
||
class Runner:
|
||
RETRY_ATTEMPTS = 3 # повторов команды после обрыва связи с Eva / 5xx (R-10)
|
||
RETRY_MIN_WAIT = 30 # секунд до повтора
|
||
|
||
def __init__(self, cfg: dict, hub: HubClient, eva: EvaClient, journal: Journal):
|
||
self.cfg, self.hub, self.eva, self.journal = cfg, hub, eva, journal
|
||
self.settings = None # type: Optional[dict]
|
||
self.executor = None # type: Optional[Executor]
|
||
self.stop = False
|
||
self.sleep = time.sleep
|
||
self.idle_limit = None # type: Optional[int]
|
||
self.updater = None # type: Optional[Updater]
|
||
self.exit_code = 0 # EXIT_UPDATE — выход для подмены версии
|
||
|
||
@property
|
||
def poll(self) -> float:
|
||
return max(float(self.cfg.get("poll_seconds", 5) or 0), 0)
|
||
|
||
def hub_wait(self, e: Exception, base: float) -> float:
|
||
"""Залогировать сбой хаба и вернуть, сколько ждать. 401/403 — токен неверен: повторять часто бессмысленно."""
|
||
if isinstance(e, HttpError) and e.code in (401, 403):
|
||
log("хаб: HTTP %d — неверный hub_token в agent.json, повтор через 60 с" % e.code)
|
||
return 60
|
||
log("хаб: %s, повтор через %d с" % (e, base))
|
||
return base
|
||
|
||
def load_settings(self) -> None:
|
||
self.settings = self.hub.settings()
|
||
self.executor = Executor(self.eva, self.hub, self.settings)
|
||
self.executor.sleep = self.sleep
|
||
self.eva.pacing = float(self.settings.get("wiki_pacing", 0.2))
|
||
log("настройки хаба v%s" % self.settings.get("settings_version"))
|
||
|
||
def restore_pace(self) -> None:
|
||
"""После перезапуска время последнего вызова Eva неизвестно (живёт в памяти). Берём самое позднее время записи
|
||
журнала: finished пишется сразу после последнего вызова команды, оценка не раньше настоящего вызова (M-9).
|
||
Время в будущем (часы ушли назад) — считаем «сейчас»: пауза полная (M-6 #12). Оценка служит только паузе:
|
||
last_call_iso (уходит хабу в claim) ставит лишь настоящий вызов Eva (M-7 #12)."""
|
||
if self.eva.last_call_at is not None:
|
||
return
|
||
at = self.journal.last_at()
|
||
if at is None:
|
||
return
|
||
ago = (datetime.now(timezone.utc) - at).total_seconds()
|
||
if ago < 0:
|
||
log("журнал: время последней записи в будущем (%s) — пауза перед первой командой полная" % at.isoformat())
|
||
self.eva.last_call_at = time.monotonic()
|
||
return # вторая строка «не позже <будущее>» противоречила бы первой (M-2 #18)
|
||
self.eva.last_call_at = time.monotonic() - ago
|
||
log("темп: последний вызов Eva не позже %s (по журналу)"
|
||
% at.astimezone(timezone.utc).replace(microsecond=0).isoformat())
|
||
|
||
def send_result(self, rec: dict) -> bool:
|
||
"""Отчёт хабу; повторяется каждые 10 с, пока хаб не подтвердит. 404/409/422 — хаб отверг (команда отменена
|
||
или удалена): считаем закрытой, чтобы не зациклиться. При остановке по сигналу делается одна попытка."""
|
||
cid = rec["command_id"]
|
||
payload = {"status": rec["status"], "result": rec.get("result"), "error": rec.get("error"),
|
||
"jira_calls": rec.get("jira_calls", 0), "rate_limited": rec.get("rate_limited", 0),
|
||
"agent_version": AGENT_VERSION}
|
||
while True:
|
||
try:
|
||
self.hub.result(cid, payload)
|
||
self.journal.acked(cid)
|
||
return True
|
||
except HttpError as e:
|
||
if e.code in (404, 409, 422):
|
||
log("хаб отверг отчёт по #%s: %s — считаю закрытым" % (cid, e))
|
||
self.journal.acked(cid)
|
||
return True
|
||
wait = self.hub_wait(e, 10)
|
||
except ConnectionError as e:
|
||
wait = self.hub_wait(e, 10)
|
||
if self.stop:
|
||
log("остановка: отчёт по #%s не подтверждён, будет дослан при следующем старте" % cid)
|
||
return False
|
||
self.sleep(wait)
|
||
|
||
def wait_seconds(self, seconds) -> bool:
|
||
"""Спать заданное время шагами по секунде; False — остановлены во время ожидания."""
|
||
remaining = float(seconds or 0)
|
||
if remaining > 0:
|
||
log(" пауза %.0f с" % remaining)
|
||
while remaining > 0 and not self.stop:
|
||
step = min(remaining, 1.0)
|
||
self.sleep(step)
|
||
remaining -= step
|
||
return not self.stop
|
||
|
||
def wait_pause(self, pause_seconds) -> bool:
|
||
"""Досыпать до зазора от последнего вызова Eva; False — остановлены во время паузы.
|
||
Если вызовов Eva ещё не было — пауза не нужна."""
|
||
pause = float(pause_seconds or 0)
|
||
if pause > 0 and self.eva.last_call_at is not None:
|
||
return self.wait_seconds(pause - (time.monotonic() - self.eva.last_call_at))
|
||
return not self.stop
|
||
|
||
def run_once(self, cmd: dict) -> dict:
|
||
try:
|
||
return self.executor.run(cmd)
|
||
except Exception as e: # баг в обработчике не должен ронять цикл и зацикливать восстановление
|
||
log(" ! необработанная ошибка: %r" % e)
|
||
return {"status": "failed", "result": None,
|
||
"error": "внутренняя ошибка агента: %s: %s" % (type(e).__name__, str(e)[:300]),
|
||
"jira_calls": self.eva.calls, "rate_limited": self.eva.rate_limited}
|
||
|
||
def execute(self, cmd: dict, journal_claimed: bool = True) -> None:
|
||
cid = cmd["id"]
|
||
log("#%s %s %s" % (cid, cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200]))
|
||
if journal_claimed:
|
||
self.journal.claimed(cid) # сразу после claim, до паузы (R-9): остановка не теряет команду
|
||
if not self.wait_pause(cmd.get("pause_seconds")):
|
||
return
|
||
self.journal.started(cid) # started — только после паузы
|
||
calls = limited = attempt = 0
|
||
while True:
|
||
out = self.run_once(cmd)
|
||
calls += out["jira_calls"]
|
||
limited += out["rate_limited"]
|
||
peer = out.pop("peer", "Eva") # с кем оборвалась связь: Eva или хаб
|
||
if not out.pop("retryable", False):
|
||
break
|
||
if attempt >= self.RETRY_ATTEMPTS:
|
||
out = {"status": "failed", "result": None,
|
||
"error": "результат неизвестен (обрыв связи с %s), повтор безопасен: %s" % (peer, out.get("error"))}
|
||
break
|
||
attempt += 1
|
||
log(" ! обрыв связи с %s, повтор %d/%d через проверку: %s" % (peer, attempt, self.RETRY_ATTEMPTS, out.get("error")))
|
||
# F-1: обрыв не обновляет last_call_at, поэтому ждём фиксированно от момента отказа, а не от последнего вызова
|
||
if not self.wait_seconds(max(float(cmd.get("pause_seconds") or 0), self.RETRY_MIN_WAIT)):
|
||
return # остановка: finished не пишем, команду подхватит recover
|
||
out["jira_calls"], out["rate_limited"] = calls, limited
|
||
self.journal.finished(cid, out["status"], out.get("result"), out.get("error"), calls, limited)
|
||
log(" → %s %s" % (out["status"], out.get("error") or json.dumps(out.get("result"), ensure_ascii=False)))
|
||
rec = dict(out)
|
||
rec["command_id"] = cid
|
||
self.send_result(rec)
|
||
|
||
def fetch_command(self, cid):
|
||
"""Команда из хаба по id; None — хаб такой не знает; при недоступности хаба ждём и повторяем."""
|
||
while True:
|
||
try:
|
||
return self.hub.command(cid)
|
||
except HttpError as e:
|
||
if e.code in (404, 409, 422):
|
||
log("#%s: хаб ответил %s" % (cid, e))
|
||
return None
|
||
wait = self.hub_wait(e, 10)
|
||
except ConnectionError as e:
|
||
wait = self.hub_wait(e, 10)
|
||
if self.stop:
|
||
raise InterruptedError(cid)
|
||
self.sleep(wait)
|
||
|
||
def recover(self) -> None:
|
||
for rec in self.journal.unacked_finished():
|
||
log("досылаю отчёт по #%s" % rec["command_id"])
|
||
self.send_result(rec)
|
||
for cid in self.journal.unfinished_started():
|
||
if self.stop:
|
||
return
|
||
try:
|
||
cmd = self.fetch_command(cid)
|
||
except InterruptedError:
|
||
return
|
||
if cmd is None:
|
||
self.journal.acked(cid)
|
||
continue
|
||
if cmd.get("status") != "claimed":
|
||
log("#%s уже %s в хабе — пропускаю" % (cid, cmd.get("status")))
|
||
self.journal.acked(cid)
|
||
continue
|
||
log("#%s выдана до перезапуска — выполняю через проверку" % cid)
|
||
self.execute(cmd, journal_claimed=False) # claimed/started для неё уже в журнале (FM-4); пауза — из хаба (M-9)
|
||
|
||
def reload_settings_before(self, cmd) -> bool:
|
||
"""Настройки сменились: перечитать ДО выполнения. False — остановка, команда не должна выполняться
|
||
(если она уже выдана хабом, фиксируем claimed — подхватит recover)."""
|
||
while True:
|
||
try:
|
||
self.load_settings()
|
||
return True
|
||
except (ConnectionError, HttpError) as e:
|
||
wait = self.hub_wait(e, max(self.poll, 1))
|
||
if self.stop:
|
||
if cmd:
|
||
self.journal.claimed(cmd["id"])
|
||
return False
|
||
if not cmd:
|
||
self.sleep(wait)
|
||
return False
|
||
self.sleep(wait)
|
||
|
||
def loop(self) -> None:
|
||
self.restore_pace()
|
||
idle, recovered = 0, False
|
||
while not self.stop:
|
||
if self.settings is None:
|
||
try:
|
||
self.load_settings()
|
||
except (ConnectionError, HttpError) as e:
|
||
self.sleep(self.hub_wait(e, max(self.poll, 1)))
|
||
continue
|
||
if not recovered:
|
||
self.recover()
|
||
recovered = not self.stop
|
||
continue
|
||
try:
|
||
upd = self.updater
|
||
resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.eva.last_call_iso,
|
||
upd.state() if upd else None, upd.launcher_version if upd else None)
|
||
except (ConnectionError, HttpError) as e:
|
||
self.sleep(self.hub_wait(e, max(self.poll, 1)))
|
||
continue
|
||
if self.updater:
|
||
self.updater.confirm()
|
||
cmd = resp.get("command")
|
||
if resp.get("settings_version") != self.settings.get("settings_version"):
|
||
if not self.reload_settings_before(cmd):
|
||
continue
|
||
if cmd:
|
||
idle = 0
|
||
self.execute(cmd)
|
||
continue
|
||
if self.updater and not self.stop and self.updater.check(resp.get("agent_release")):
|
||
if not self.stop: # Ctrl-C во время подготовки — обычная остановка; staged доделает run.py
|
||
self.exit_code = EXIT_UPDATE
|
||
return
|
||
idle += 1
|
||
if self.idle_limit is not None and idle >= self.idle_limit:
|
||
return
|
||
self.sleep(self.poll)
|
||
|
||
|
||
def acquire_lock(journal_path: str):
|
||
"""Исключительная блокировка <журнал>.lock: один процесс агента на журнал. Хаб довыдаёт команду без отчёта при
|
||
следующем claim, поэтому второй процесс выполнял бы ту же команду одновременно с первым. None — уже занято."""
|
||
f = open(journal_path + ".lock", "a")
|
||
try:
|
||
fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except OSError:
|
||
f.close()
|
||
return None
|
||
return f
|
||
|
||
|
||
def cmd_run(cfg: dict, journal_path: str, config_path: Optional[str] = None) -> int:
|
||
lock = acquire_lock(journal_path)
|
||
if lock is None:
|
||
log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path)
|
||
return EXIT_LOCKED
|
||
try:
|
||
return _run(cfg, journal_path, config_path)
|
||
finally:
|
||
lock.close()
|
||
|
||
|
||
def cmd_prune(journal_path: str) -> int:
|
||
"""Удалить подтверждённые записи старше 30 дней. Под той же блокировкой, что run: prune перезаписывает файл, и запись,
|
||
добавленная работающим агентом (finished — неотправленный отчёт), пропала бы (T2.3)."""
|
||
lock = acquire_lock(journal_path)
|
||
if lock is None:
|
||
log("агент запущен с журналом %s — prune невозможен, сначала остановите агента" % journal_path)
|
||
return 1
|
||
try:
|
||
removed = Journal(journal_path).prune(30)
|
||
log("журнал: удалено записей %d" % removed)
|
||
return 0
|
||
finally:
|
||
lock.close()
|
||
|
||
|
||
def launcher_version() -> Optional[int]:
|
||
"""Версия запускателя из окружения (run.py ставит JIRA_AGENT_LAUNCHER — имя общее с jira-agent); без запускателя — None."""
|
||
try:
|
||
return int(os.environ["JIRA_AGENT_LAUNCHER"])
|
||
except (KeyError, ValueError):
|
||
return None
|
||
|
||
|
||
def _run(cfg: dict, journal_path: str, config_path: Optional[str] = None) -> int:
|
||
runner = Runner(cfg, HubClient(cfg), EvaClient(cfg), Journal(journal_path))
|
||
runner.updater = Updater(runner.hub, HERE, launcher_version(), cfg.get("auto_update", True) is not False, config_path)
|
||
runner.updater.should_stop = lambda: runner.stop
|
||
|
||
def on_signal(signum, frame):
|
||
try:
|
||
if runner.stop:
|
||
log("повторный сигнал — выхожу немедленно")
|
||
else:
|
||
log("сигнал %s: доделываю текущую команду и выхожу" % signum)
|
||
except RuntimeError: # print внутри обработчика, прервавшего другой print (reentrant call)
|
||
pass
|
||
if runner.stop:
|
||
sys.exit(130)
|
||
runner.stop = True
|
||
|
||
signal.signal(signal.SIGINT, on_signal)
|
||
signal.signal(signal.SIGTERM, on_signal)
|
||
log("eva-agent v%d: хаб %s, Eva %s, журнал %s" % (AGENT_VERSION, cfg["hub_url"], cfg["eva"]["base_url"], journal_path))
|
||
runner.loop()
|
||
log("выход для обновления (код %d)" % EXIT_UPDATE if runner.exit_code == EXIT_UPDATE else "остановлен")
|
||
return runner.exit_code
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CLI
|
||
# ---------------------------------------------------------------------------
|
||
def cmd_check(cfg: dict) -> int:
|
||
ok = True
|
||
hub = HubClient(cfg)
|
||
via = ""
|
||
if hub.folder is not None:
|
||
hub.timeout = 10 # реле молчит дольше 10 с — хаб недоступен
|
||
via = " через папку %s" % hub.folder.root
|
||
try:
|
||
s = hub.settings()
|
||
log("хаб: доступен%s, настройки v%s" % (via, s.get("settings_version")))
|
||
except (ConnectionError, HttpError) as e:
|
||
log("хаб: НЕДОСТУПЕН — %s" % e)
|
||
ok = False
|
||
# проверка Eva — в задаче 6 плана
|
||
return 0 if ok else 1
|
||
|
||
|
||
def cmd_selftest(cfg: dict) -> int:
|
||
"""Самопроверка версии перед обновлением: её запускает предыдущая версия. Контракт кодов MUST сохраняться:
|
||
0 — хаб (и Eva, когда проверка добавится) доступны, 3 — что-то недоступно (временно); любой другой код — файл негоден (design 10)."""
|
||
log("eva-agent v%d" % AGENT_VERSION)
|
||
return 0 if cmd_check(cfg) == 0 else 3
|
||
|
||
|
||
def main(argv=None) -> int:
|
||
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
ap.add_argument("--config", default=os.path.join(HERE, "agent.json"))
|
||
ap.add_argument("--journal", default=os.path.join(HERE, "journal.jsonl"))
|
||
ap.add_argument("cmd", nargs="?", default="run", choices=["run", "check", "selftest", "prune"])
|
||
args = ap.parse_args(argv)
|
||
cfg = load_config(args.config)
|
||
if args.cmd == "check":
|
||
return cmd_check(cfg)
|
||
if args.cmd == "selftest":
|
||
return cmd_selftest(cfg)
|
||
if args.cmd == "prune":
|
||
return cmd_prune(args.journal)
|
||
return cmd_run(cfg, args.journal, args.config)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|