feat(agent): самообновление в паузе — скачивание, sha256, selftest, выход 75 (#19)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
140f921ce5
commit
3aa8823929
129
src/agent.py
129
src/agent.py
@ -950,6 +950,10 @@ class Executor:
|
||||
# ---------------------------------------------------------------------------
|
||||
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)
|
||||
|
||||
def __init__(self, hub, base_dir: str, launcher_version: Optional[int], enabled: bool = True,
|
||||
config_path: Optional[str] = None):
|
||||
@ -958,6 +962,11 @@ class Updater:
|
||||
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.hint_at = None # type: Optional[float]
|
||||
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)
|
||||
@ -969,6 +978,101 @@ class Updater:
|
||||
write_update_state(self.state_path, "ok", st.get("from_version"), AGENT_VERSION, st.get("sha256"))
|
||||
log("обновление до v%d подтверждено" % AGENT_VERSION)
|
||||
|
||||
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
|
||||
try:
|
||||
version, sha = int(release.get("version")), str(release.get("sha256"))
|
||||
except (TypeError, ValueError):
|
||||
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:
|
||||
now = self.clock()
|
||||
if self.hint_at is None or now - self.hint_at >= self.HINT_SECONDS:
|
||||
self.hint_at = now
|
||||
log("доступна 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"
|
||||
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)
|
||||
if self.should_stop():
|
||||
self._drop_new()
|
||||
return False
|
||||
code, out = self.run_selftest(self.new_path)
|
||||
if self.should_stop(): # Ctrl-C уронил и selftest — это не отказ версии
|
||||
self._drop_new()
|
||||
return False
|
||||
if code == 0:
|
||||
write_update_state(self.state_path, "staged", AGENT_VERSION, version, sha)
|
||||
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: хаб или Jira недоступны")
|
||||
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 FileNotFoundError:
|
||||
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()
|
||||
write_update_state(self.state_path, "rejected", AGENT_VERSION, version, sha, reason)
|
||||
log("v%d отвергнута: %s" % (version, reason))
|
||||
return False
|
||||
|
||||
|
||||
class Runner:
|
||||
RETRY_ATTEMPTS = 3 # повторов команды после обрыва связи с Jira / 5xx (R-10)
|
||||
@ -1190,6 +1294,9 @@ class Runner:
|
||||
idle = 0
|
||||
self.execute(cmd)
|
||||
continue
|
||||
if self.updater and not self.stop and self.updater.check(resp.get("agent_release")):
|
||||
self.exit_code = EXIT_UPDATE
|
||||
return
|
||||
idle += 1
|
||||
if self.idle_limit is not None and idle >= self.idle_limit:
|
||||
return
|
||||
@ -1208,13 +1315,13 @@ def acquire_lock(journal_path: str):
|
||||
return f
|
||||
|
||||
|
||||
def cmd_run(cfg: dict, journal_path: str) -> int:
|
||||
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 1
|
||||
try:
|
||||
return _run(cfg, journal_path)
|
||||
return _run(cfg, journal_path, config_path)
|
||||
finally:
|
||||
lock.close()
|
||||
|
||||
@ -1234,8 +1341,18 @@ def cmd_prune(journal_path: str) -> int:
|
||||
lock.close()
|
||||
|
||||
|
||||
def _run(cfg: dict, journal_path: str) -> int:
|
||||
def launcher_version() -> Optional[int]:
|
||||
"""Версия запускателя из окружения (run.py ставит JIRA_AGENT_LAUNCHER); без запускателя — 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), JiraClient(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:
|
||||
@ -1253,8 +1370,8 @@ def _run(cfg: dict, journal_path: str) -> int:
|
||||
signal.signal(signal.SIGTERM, on_signal)
|
||||
log("jira-agent v%d: хаб %s, Jira %s, журнал %s" % (AGENT_VERSION, cfg["hub_url"], cfg["jira"]["base_url"], journal_path))
|
||||
runner.loop()
|
||||
log("остановлен")
|
||||
return 0
|
||||
log("выход для обновления (код %d)" % EXIT_UPDATE if runner.exit_code == EXIT_UPDATE else "остановлен")
|
||||
return runner.exit_code
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -1302,7 +1419,7 @@ def main(argv=None) -> int:
|
||||
return cmd_selftest(cfg)
|
||||
if args.cmd == "prune":
|
||||
return cmd_prune(args.journal)
|
||||
return cmd_run(cfg, args.journal)
|
||||
return cmd_run(cfg, args.journal, args.config)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@ -66,6 +66,27 @@ class RunnerTest(unittest.TestCase):
|
||||
self.assertIsNone(self.hub.claim_bodies[0]["update_state"])
|
||||
self.assertIsNone(self.hub.claim_bodies[0]["launcher_version"])
|
||||
|
||||
def test_idle_with_release_exits_for_update(self):
|
||||
rel = {"version": 6, "sha256": "s", "content": "x"}
|
||||
self.hub.release = rel
|
||||
self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1)
|
||||
self.runner.updater.check = lambda r: r == {"version": 6, "sha256": "s"}
|
||||
self.runner.idle_limit = 5
|
||||
self.runner.loop()
|
||||
self.assertEqual(self.runner.exit_code, agent.EXIT_UPDATE)
|
||||
self.assertEqual(self.hub.claims, 1)
|
||||
|
||||
def test_command_first_then_update(self):
|
||||
self.hub.release = {"version": 6, "sha256": "s", "content": "x"}
|
||||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||||
self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1)
|
||||
seen = []
|
||||
self.runner.updater.check = lambda r: (seen.append(r), True)[1]
|
||||
self.runner.loop()
|
||||
self.assertEqual(self.hub.results[1]["status"], "done") # команда выполнена и подтверждена
|
||||
self.assertEqual(len(seen), 1) # проверка — только после claim без команды
|
||||
self.assertEqual(self.runner.exit_code, agent.EXIT_UPDATE)
|
||||
|
||||
def test_restart_pace_from_old_finished(self):
|
||||
at = (datetime.now(timezone.utc) - timedelta(seconds=10)).replace(microsecond=0).isoformat()
|
||||
self.journal.append({"kind": "finished", "command_id": 99, "status": "done", "result": None, "error": None,
|
||||
|
||||
@ -1,9 +1,11 @@
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from _common import agent
|
||||
from _common import SETTINGS, agent
|
||||
from fake_hub import FakeHub
|
||||
|
||||
|
||||
class UpdateStateTest(unittest.TestCase):
|
||||
@ -49,5 +51,161 @@ class UpdateStateTest(unittest.TestCase):
|
||||
self.assertEqual(u.state()["state"], "rolled_back")
|
||||
|
||||
|
||||
def make_release(version, body="print('ok')\n"):
|
||||
content = "AGENT_VERSION = %d\n%s" % (version, body)
|
||||
return {"version": version, "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), "content": content}
|
||||
|
||||
|
||||
class UpdaterCheckTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.dir = tempfile.TemporaryDirectory()
|
||||
self.hub = FakeHub(SETTINGS)
|
||||
cfg = {"hub_url": self.hub.base_url, "hub_token": self.hub.token, "timeout": 5}
|
||||
self.u = agent.Updater(agent.HubClient(cfg), self.dir.name, 1)
|
||||
self.selftests = []
|
||||
self.code = 0
|
||||
self.u.run_selftest = lambda path: (self.selftests.append(path), (self.code, "вывод selftest"))[1]
|
||||
self.now = [1000.0]
|
||||
self.u.clock = lambda: self.now[0]
|
||||
|
||||
def tearDown(self):
|
||||
self.hub.stop()
|
||||
self.dir.cleanup()
|
||||
|
||||
def ref(self, rel):
|
||||
return {"version": rel["version"], "sha256": rel["sha256"]}
|
||||
|
||||
def test_good_version_is_staged(self):
|
||||
rel = make_release(6)
|
||||
self.hub.release = rel
|
||||
self.assertTrue(self.u.check(self.ref(rel)))
|
||||
with open(self.u.new_path, encoding="utf-8") as f:
|
||||
self.assertEqual(f.read(), rel["content"])
|
||||
st = self.u.state()
|
||||
self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("staged", 5, 6, rel["sha256"]))
|
||||
self.assertEqual(self.selftests, [self.u.new_path])
|
||||
|
||||
def test_not_newer_or_disabled_or_no_release(self):
|
||||
self.hub.release = make_release(5)
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertFalse(self.u.check(None))
|
||||
self.hub.release = make_release(6)
|
||||
self.u.enabled = False
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertEqual(self.hub.release_requests, 0)
|
||||
|
||||
def test_sha_mismatch_rejected_once(self):
|
||||
rel = make_release(6)
|
||||
self.hub.release = dict(rel, content=rel["content"] + "# подмена\n")
|
||||
ref = self.ref(rel)
|
||||
self.hub.release["sha256"] = rel["sha256"]
|
||||
self.assertFalse(self.u.check(ref))
|
||||
st = self.u.state()
|
||||
self.assertEqual((st["state"], st["to_version"], st["error"]), ("rejected", 6, "sha256 не сходится"))
|
||||
self.assertFalse(os.path.exists(self.u.new_path))
|
||||
self.assertFalse(self.u.check(ref))
|
||||
self.assertEqual(self.hub.release_requests, 1) # отвергнутую версию больше не качает
|
||||
|
||||
def test_syntax_error_and_wrong_version_rejected(self):
|
||||
bad = make_release(6, body="def (:\n")
|
||||
self.hub.release = bad
|
||||
self.assertFalse(self.u.check(self.ref(bad)))
|
||||
self.assertIn("не компилируется", self.u.state()["error"])
|
||||
wrong = make_release(7)
|
||||
wrong = dict(wrong, version=8)
|
||||
self.hub.release = wrong
|
||||
self.assertFalse(self.u.check(self.ref(wrong)))
|
||||
self.assertIn("AGENT_VERSION", self.u.state()["error"])
|
||||
self.assertEqual(self.selftests, [])
|
||||
|
||||
def test_selftest_bad_code_rejected(self):
|
||||
self.hub.release = make_release(6)
|
||||
self.code = 1
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
st = self.u.state()
|
||||
self.assertEqual(st["state"], "rejected")
|
||||
self.assertTrue(st["error"].startswith("selftest: код 1"))
|
||||
self.assertIn("вывод selftest", st["error"])
|
||||
self.assertFalse(os.path.exists(self.u.new_path))
|
||||
|
||||
def test_selftest_unavailable_is_transient(self):
|
||||
self.hub.release = make_release(6)
|
||||
self.code = 3
|
||||
ref = self.ref(self.hub.release)
|
||||
self.assertFalse(self.u.check(ref))
|
||||
self.assertIsNone(self.u.state()) # update.json не тронут
|
||||
self.assertFalse(os.path.exists(self.u.new_path))
|
||||
self.now[0] += 299
|
||||
self.assertFalse(self.u.check(ref))
|
||||
self.assertEqual(len(self.selftests), 1) # раньше 5 минут не повторяет
|
||||
self.now[0] += 2
|
||||
self.code = 0
|
||||
self.assertTrue(self.u.check(ref))
|
||||
|
||||
def test_selftest_timeout_is_transient(self):
|
||||
self.hub.release = make_release(6)
|
||||
self.code = None
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertIsNone(self.u.state())
|
||||
|
||||
def test_download_failure_is_transient(self):
|
||||
rel = make_release(6) # в claim есть, а /agent/release — 404
|
||||
self.assertFalse(self.u.check(self.ref(rel)))
|
||||
self.assertIsNone(self.u.state())
|
||||
|
||||
def test_release_changed_between_requests_is_transient(self):
|
||||
old, new = make_release(6), make_release(7)
|
||||
self.hub.release = new
|
||||
self.assertFalse(self.u.check(self.ref(old)))
|
||||
self.assertIsNone(self.u.state())
|
||||
self.assertEqual(self.selftests, [])
|
||||
|
||||
def test_rolled_back_same_sha_not_retried(self):
|
||||
rel = make_release(6)
|
||||
self.hub.release = rel
|
||||
agent.write_update_state(self.u.state_path, "rolled_back", 5, 6, rel["sha256"], "код 1 через 3 с")
|
||||
self.assertFalse(self.u.check(self.ref(rel)))
|
||||
fixed = make_release(6, body="print('fixed')\n")
|
||||
self.hub.release = fixed
|
||||
self.assertTrue(self.u.check(self.ref(fixed))) # тот же номер, другой sha — пробуем
|
||||
|
||||
def test_without_launcher_only_hint(self):
|
||||
self.u.launcher_version = None
|
||||
self.hub.release = make_release(6)
|
||||
logs = []
|
||||
orig = agent.log
|
||||
agent.log = logs.append
|
||||
try:
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.now[0] += 3600
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
finally:
|
||||
agent.log = orig
|
||||
self.assertEqual(logs.count("доступна v6; для самообновления запускайте через run.py"), 2)
|
||||
self.assertEqual(self.hub.release_requests, 0)
|
||||
|
||||
def test_stop_during_selftest_is_not_reject(self):
|
||||
self.hub.release = make_release(6)
|
||||
stop = [False]
|
||||
self.u.should_stop = lambda: stop[0]
|
||||
|
||||
def selftest(path):
|
||||
stop[0] = True # Ctrl-C пришёл и selftest упал с 1
|
||||
return 1, "KeyboardInterrupt"
|
||||
self.u.run_selftest = selftest
|
||||
self.assertFalse(self.u.check(self.ref(self.hub.release)))
|
||||
self.assertIsNone(self.u.state())
|
||||
self.assertFalse(os.path.exists(self.u.new_path))
|
||||
|
||||
def test_real_selftest_runs_new_file(self):
|
||||
u = agent.Updater(None, self.dir.name, 1, config_path=os.path.join(self.dir.name, "agent.json"))
|
||||
path = os.path.join(self.dir.name, "agent.py.new")
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
f.write("import sys\nprint('args', sys.argv[1:])\nsys.exit(3)\n")
|
||||
code, out = u._run_selftest(path)
|
||||
self.assertEqual(code, 3)
|
||||
self.assertIn("'--config', '%s', 'selftest'" % os.path.join(self.dir.name, "agent.json"), out)
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user