244 lines
12 KiB
Python
244 lines
12 KiB
Python
"""Реле через общую папку (specs/hub-api, «Реле через общую папку»): файлы-запросы агента из <relay_dir>/req/
|
||
пересылаются в HTTP этого же хаба, ответы кладутся в <relay_dir>/resp/. HTTP API, обработчики и токены не меняются;
|
||
реле для них — ещё один клиент. Нужно, когда VPN на виртуалке закрывает сеть до хаба (design.md folder-transport)."""
|
||
import http.client
|
||
import json
|
||
import os
|
||
import threading
|
||
import time
|
||
import traceback
|
||
import urllib.error
|
||
import urllib.request
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
from . import settings
|
||
|
||
CLAIM_PATH = "/agent/claim"
|
||
CLAIM_MIN_LEFT = 35 # с: claim пересылается, только если до deadline осталось не меньше (хаб держит claim ≤ 30 с)
|
||
SETTINGS_EVERY = 5 # с: как часто перечитывать relay_dir
|
||
CLEANUP_EVERY = 60 # с: как часто убирать мусор
|
||
RESP_TTL = 600 # с: ответ, за которым агент не пришёл
|
||
REQ_GRACE = 60 # с после deadline: запрос, который никто не взял
|
||
TMP_TTL = 600 # с: брошенные временные файлы (имя на точку)
|
||
BAD_JSON_WAIT = 2 # с: сколько перечитывать нечитаемый запрос
|
||
FORWARD_TIMEOUT = 90 # с: ожидание ответа своего HTTP (claim держит до 30 с)
|
||
METHODS = ("GET", "POST", "PUT")
|
||
|
||
|
||
def say(msg: str) -> None:
|
||
print(f"[{time.strftime('%d/%b/%Y %H:%M:%S')}] {msg}", flush=True)
|
||
|
||
|
||
def parse_deadline(value) -> float | None:
|
||
"""ISO 8601 → секунды эпохи; без зоны считается UTC; мусор → None."""
|
||
if not isinstance(value, str):
|
||
return None
|
||
try:
|
||
dt = datetime.fromisoformat(value)
|
||
except ValueError:
|
||
return None
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.timestamp()
|
||
|
||
|
||
def write_atomic(directory: Path, name: str, payload: dict) -> None:
|
||
"""Записать JSON под временным именем на точку и переименовать: читатель видит файл только целиком."""
|
||
tmp = Path(directory) / f".{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, Path(directory) / name)
|
||
|
||
|
||
class Relay:
|
||
def __init__(self, hub, port: int, host: str = "127.0.0.1", poll: float = 0.1):
|
||
self.hub, self.port, self.host, self.poll = hub, port, host, poll
|
||
self.root: Path | None = None
|
||
self._settings_at: float | None = None
|
||
self._cleanup_at = 0.0
|
||
self._ready = False
|
||
self._warned: str | None = None
|
||
self._cleaned: set[Path] = set() # папки, чей work/ вычищен после старта реле
|
||
self.workers: list[threading.Thread] = []
|
||
|
||
# --- цикл ---------------------------------------------------------------------
|
||
def run(self, stop: threading.Event) -> None:
|
||
while not stop.is_set():
|
||
try:
|
||
self.step()
|
||
except Exception: # noqa: BLE001 — сбой одного шага не останавливает реле
|
||
traceback.print_exc()
|
||
stop.wait(self.poll)
|
||
|
||
def step(self) -> None:
|
||
now = time.time()
|
||
if self._settings_at is None or now - self._settings_at >= SETTINGS_EVERY:
|
||
self._settings_at = now
|
||
self._reload()
|
||
if self.root is None:
|
||
return
|
||
try:
|
||
if not self.root.is_dir():
|
||
self._warn(f"relay: папки {self.root} нет — жду")
|
||
self._ready = False
|
||
return
|
||
if not self._ready:
|
||
self._prepare()
|
||
for path in sorted((self.root / "req").iterdir()):
|
||
if path.name.startswith(".") or not path.name.endswith(".json"):
|
||
continue
|
||
self._take(path)
|
||
except OSError as e: # нет доступа (TCC), подпапку удалили — подготовить заново на следующем шаге
|
||
self._ready = False
|
||
self._warn(f"relay: нет доступа к папке {self.root} — {e.strerror or e}; жду")
|
||
return
|
||
if now - self._cleanup_at >= CLEANUP_EVERY:
|
||
self._cleanup_at = now
|
||
self.cleanup(now)
|
||
|
||
def _warn(self, msg: str) -> None:
|
||
"""Одно предупреждение на причину: без потока одинаковых строк каждые 0,1 с."""
|
||
if msg != self._warned:
|
||
say(msg)
|
||
self._warned = msg
|
||
|
||
def join(self, timeout: float = 5) -> None:
|
||
for w in list(self.workers):
|
||
w.join(timeout)
|
||
|
||
# --- папка ----------------------------------------------------------------------
|
||
def _reload(self) -> None:
|
||
conn = self.hub.conn()
|
||
try:
|
||
value = settings.get(conn, "relay_dir")
|
||
finally:
|
||
conn.close()
|
||
root = Path(value) if value else None
|
||
if root != self.root:
|
||
self.root, self._ready, self._warned = root, False, None
|
||
say(f"relay: папка {root}" if root else "relay: выключено")
|
||
|
||
def _prepare(self) -> None:
|
||
for sub in ("req", "resp", "work"):
|
||
(self.root / sub).mkdir(exist_ok=True)
|
||
if self.root not in self._cleaned:
|
||
# первая подготовка этой папки после старта: взятые прошлым процессом запросы уже никто не обработает
|
||
for stale in (self.root / "work").iterdir():
|
||
stale.unlink(missing_ok=True)
|
||
say(f"relay: удалён незавершённый запрос {stale.name} (старт реле)")
|
||
self._cleaned.add(self.root)
|
||
else:
|
||
# пропал подкаталог или не было доступа: взятые запросы ещё обрабатываются своими потоками — work/ не трогаем
|
||
say(f"relay: папка {self.root} подготовлена заново")
|
||
self._ready, self._warned = True, None
|
||
|
||
def _take(self, path: Path) -> None:
|
||
target = self.root / "work" / path.name
|
||
try:
|
||
os.rename(path, target)
|
||
except FileNotFoundError:
|
||
if path.exists(): # пропала work/ — step() подготовит папку заново
|
||
raise
|
||
return # агент успел удалить свой просроченный запрос
|
||
worker = threading.Thread(target=self.process, args=(self.root, target), daemon=True)
|
||
worker.start()
|
||
self.workers = [w for w in self.workers if w.is_alive()] + [worker]
|
||
|
||
# --- обработка запроса -------------------------------------------------------------
|
||
def process(self, root: Path, work: Path) -> None:
|
||
started = time.monotonic()
|
||
try:
|
||
req = self._read(work)
|
||
if req is None:
|
||
say(f"relay: нечитаемый запрос {work.name} — выброшен")
|
||
return
|
||
method, path = req.get("method"), req.get("path")
|
||
deadline = parse_deadline(req.get("deadline"))
|
||
left = deadline - time.time() if deadline is not None else -1.0
|
||
if left <= 0 or (method == "POST" and path == CLAIM_PATH and left < CLAIM_MIN_LEFT):
|
||
say(f"relay {method} {path} просрочен (осталось {left:.0f} с) — выброшен")
|
||
return
|
||
(root / "resp").mkdir(exist_ok=True) # до пересылки: выданный claim не должен уйти в никуда
|
||
status, body = self.forward(method, path, req.get("body"), req.get("authorization"))
|
||
write_atomic(root / "resp", work.name, {"id": req.get("id"), "status": status, "body": body})
|
||
say(f"relay {method} {path} {status} {time.monotonic() - started:.1f} с")
|
||
except OSError as e: # папка пропала на ходу — агент дождётся своего срока
|
||
say(f"relay: сбой файла {work.name}: {e}")
|
||
finally:
|
||
work.unlink(missing_ok=True)
|
||
|
||
def _read(self, work: Path) -> dict | None:
|
||
end = time.monotonic() + BAD_JSON_WAIT
|
||
while True:
|
||
try:
|
||
with open(work, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
return data if isinstance(data, dict) else None
|
||
except (ValueError, OSError):
|
||
if time.monotonic() >= end:
|
||
return None
|
||
time.sleep(0.1)
|
||
|
||
def forward(self, method, path, body, authorization) -> tuple[int, str]:
|
||
if method not in METHODS or not isinstance(path, str) or not path.startswith("/"):
|
||
return 400, _error("bad_request", f"неверный метод или путь: {method} {path}")
|
||
data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
|
||
try:
|
||
req = urllib.request.Request(f"http://{self.host}:{self.port}{path}", data=data, method=method)
|
||
if authorization:
|
||
req.add_header("Authorization", str(authorization))
|
||
if data is not None:
|
||
req.add_header("Content-Type", "application/json; charset=utf-8")
|
||
with urllib.request.urlopen(req, timeout=FORWARD_TIMEOUT) as resp:
|
||
return resp.status, resp.read().decode("utf-8", "replace")
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
return e.code, e.read().decode("utf-8", "replace")
|
||
finally:
|
||
e.close()
|
||
except ValueError as e: # недопустимый заголовок (перевод строки) или URL
|
||
return 400, _error("bad_request", str(e))
|
||
except (urllib.error.URLError, http.client.HTTPException, OSError) as e:
|
||
return 502, _error("relay", f"хаб недоступен изнутри: {e}")
|
||
|
||
# --- уборка ------------------------------------------------------------------------
|
||
def cleanup(self, now: float) -> None:
|
||
if self.root is None or not self.root.is_dir():
|
||
return
|
||
for p in _entries(self.root / "resp"):
|
||
ttl = TMP_TTL if p.name.startswith(".") else RESP_TTL
|
||
if _age(p, now) > ttl:
|
||
p.unlink(missing_ok=True)
|
||
for p in _entries(self.root / "req"):
|
||
if p.name.startswith("."):
|
||
if _age(p, now) > TMP_TTL:
|
||
p.unlink(missing_ok=True)
|
||
continue
|
||
try:
|
||
deadline = parse_deadline(json.loads(p.read_text(encoding="utf-8")).get("deadline"))
|
||
except (OSError, ValueError, AttributeError):
|
||
deadline = None
|
||
if deadline is None or now - deadline > REQ_GRACE:
|
||
p.unlink(missing_ok=True)
|
||
|
||
|
||
def _error(error: str, detail: str) -> str:
|
||
return json.dumps({"error": error, "detail": detail}, ensure_ascii=False)
|
||
|
||
|
||
def _entries(directory: Path) -> list[Path]:
|
||
try:
|
||
return list(directory.iterdir())
|
||
except FileNotFoundError:
|
||
return []
|
||
|
||
|
||
def _age(p: Path, now: float) -> float:
|
||
try:
|
||
return now - p.stat().st_mtime
|
||
except FileNotFoundError:
|
||
return 0.0
|