Merge pull request 'feat(status): раздел «Статус» — контейнеры, карточка, проверки' (#44) from feat/14-status into dev
CD / deploy (push) Successful in 34s
CI / ruff · mypy · pytest (push) Successful in 1m14s

Reviewed-on: #44
This commit was merged in pull request #44.
This commit is contained in:
2026-10-17 11:49:54 +03:00
25 changed files with 1594 additions and 13 deletions
+5
View File
@@ -24,6 +24,11 @@ GIT_SSH_KEY_B64= # (secret) для Docker: тот же ключ
GIT_UPLOADS_URL=git@ssh.sourcecraft.dev:rogachevegorushka/tgit-uploads.git
# GIT_CLONE_DIR=data/uploads # локальный клон репозитория загрузок (в Docker — volume /data/uploads)
# --- Docker: раздел «Статус» (#14) ---
# В Compose задан сам: http://docker-proxy:2375. Без него (make run) «Статус» отвечает «Docker недоступен».
# DOCKER_URL=
# DOCKER_LABEL=tgit.managed=true
# --- LLM (#17) ---
LLM_PROVIDER=mock # mock | openai_compat
LLM_BASE_URL=
+1 -1
View File
@@ -15,7 +15,7 @@ concurrency:
jobs:
deploy:
name: deploy → ${{ github.ref_name == 'main' && 'prod' || 'dev' }}
name: deploy # Gitea doesn't render expressions in job names
runs-on: ubuntu-latest
env:
DEPLOY_ENV: ${{ github.ref_name == 'main' && 'prod' || 'dev' }}
+2 -2
View File
@@ -99,8 +99,8 @@ class Row:
def human_size(n: int) -> str:
size = float(n)
for unit in ("Б", "КБ", "МБ"):
if size < 1024 or unit == "МБ":
for unit in ("Б", "КБ", "МБ", "ГБ"):
if size < 1024 or unit == "ГБ":
return f"{size:.0f} {unit}" if unit == "Б" else f"{size:.1f} {unit}"
size /= 1024
raise AssertionError("unreachable")
-1
View File
@@ -16,7 +16,6 @@ def build_router(texts: Texts, access: AccessService) -> Router:
router = Router(name="common")
sections = section_by_button(texts)
section_text = {
Section.STATUS: texts.section.status,
Section.LOGS: texts.section.logs,
Section.ALERTS: texts.section.alerts,
}
+216
View File
@@ -0,0 +1,216 @@
"""«🩺 Статус»: managed containers, a card per container, on-demand checks (SPEC S-1…S-4)."""
import logging
from datetime import datetime
from enum import StrEnum
from aiogram import F, Router
from aiogram.exceptions import TelegramBadRequest
from aiogram.filters.callback_data import CallbackData
from aiogram.fsm.context import FSMContext
from aiogram.types import CallbackQuery, InlineKeyboardMarkup, Message
from aiogram.utils.keyboard import InlineKeyboardBuilder
from app.bot.menu import Nav, NavAction
from app.bot.routers.commit import human_size
from app.db.models import utcnow
from app.runtime.checks import Checker, CheckName, ContainerReport
from app.runtime.docker import Container, Details, DockerClient, DockerError, Stats
from app.texts import StatusTexts, Texts
log = logging.getLogger(__name__)
MAX_TEXT = 4000 # Telegram's limit is 4096
class StatusAct(StrEnum):
LIST = "l"
CARD = "c"
CHECK = "k"
class StatusCb(CallbackData, prefix="s"):
a: StatusAct
id: str = "" # 12-char container id prefix: callback data is limited to 64 bytes
def duration(t: StatusTexts, seconds: float) -> str:
s = int(max(seconds, 0))
d, s = divmod(s, 86400)
h, s = divmod(s, 3600)
m, s = divmod(s, 60)
if d:
return f"{d} {t.days} {h} {t.hours}"
if h:
return f"{h} {t.hours} {m} {t.minutes}"
if m:
return f"{m} {t.minutes}"
return f"{s} {t.seconds}"
def _clip(text: str) -> str:
return text if len(text) <= MAX_TEXT else text[: MAX_TEXT - 1] + "…"
def _fmt_time(dt: datetime | None, none: str) -> str:
return dt.strftime("%d.%m %H:%M") if dt else none
def render_list(texts: Texts, containers: list[Container], now: datetime) -> tuple[str, InlineKeyboardMarkup]:
t = texts.status
lines = [t.title.format(count=len(containers), time=now.strftime("%H:%M:%S")), ""]
if not containers:
lines.append(t.empty)
for c in containers:
state = c.state if c.health is None else f"{c.state}, {c.health}"
lines.append(t.line.format(light=c.light.value, name=c.name, state=state))
kb = InlineKeyboardBuilder()
for c in containers:
kb.button(text=f"{c.light.value} {c.name}", callback_data=StatusCb(a=StatusAct.CARD, id=c.short_id))
if containers:
kb.button(text=t.check_button, callback_data=StatusCb(a=StatusAct.CHECK))
kb.button(text=t.refresh, callback_data=StatusCb(a=StatusAct.LIST))
kb.button(text=texts.menu.back, callback_data=Nav(action=NavAction.BACK))
kb.adjust(1)
return _clip("\n".join(lines)), kb.as_markup()
def render_card(texts: Texts, d: Details, stats: Stats | None, now: datetime) -> tuple[str, InlineKeyboardMarkup]:
t = texts.status
running = d.state == "running"
uptime = duration(t, (now - d.started_at).total_seconds()) if running and d.started_at else t.none
if stats is None:
cpu = mem = limit = mem_pct = t.none
else:
cpu = f"{stats.cpu_pct:.1f}%"
mem = human_size(stats.mem_bytes)
limit = human_size(stats.mem_limit) if stats.mem_limit else t.no_limit
mem_pct = f"{stats.mem_pct:.0f}%"
text = t.card.format(
light=d.light.value,
name=d.name,
service=d.service or t.none,
image=d.image or t.none,
state=d.state if running else f"{d.state} (exit {d.exit_code}{', OOM' if d.oom_killed else ''})",
health=d.health or t.none,
uptime=uptime,
restarts=d.restart_count,
cpu=cpu,
mem=mem,
limit=limit,
mem_pct=mem_pct,
sha=(d.sha or t.none)[:12],
deployed=_fmt_time(d.deployed_at, t.none),
)
kb = InlineKeyboardBuilder()
kb.button(text=t.check_button, callback_data=StatusCb(a=StatusAct.CHECK, id=d.id[:12]))
kb.button(text=t.refresh, callback_data=StatusCb(a=StatusAct.CARD, id=d.id[:12]))
kb.button(text=t.to_list, callback_data=StatusCb(a=StatusAct.LIST))
kb.adjust(2, 1)
return text, kb.as_markup()
def render_report(
texts: Texts, reports: list[ContainerReport], now: datetime, back_id: str = ""
) -> tuple[str, InlineKeyboardMarkup]:
"""back_id: the report was run from a card — offer the way back to it."""
t = texts.status
titles = {
CheckName.RUNNING: t.check_running,
CheckName.HEALTH: t.check_health,
CheckName.RESTARTS: t.check_restarts,
CheckName.CPU: t.check_cpu,
CheckName.MEM: t.check_mem,
CheckName.LOG_ERRORS: t.check_log_errors,
CheckName.HTTP: t.check_http,
}
ok = sum(r.ok for r in reports)
lines = [t.report_title.format(ok=ok, total=len(reports), time=now.strftime("%H:%M:%S"))]
for report in reports:
lines.append("")
lines.append(t.report_container.format(light="✅" if report.ok else "❌", name=report.details.name))
for r in report.results:
detail = f" · {r.detail}" if r.detail else ""
lines.append(t.report_line.format(verdict=r.verdict.value, title=titles[r.name], detail=detail))
kb = InlineKeyboardBuilder()
kb.button(text=t.refresh, callback_data=StatusCb(a=StatusAct.CHECK, id=back_id))
if back_id:
kb.button(text=t.to_card, callback_data=StatusCb(a=StatusAct.CARD, id=back_id))
kb.button(text=t.to_list, callback_data=StatusCb(a=StatusAct.LIST))
kb.adjust(1)
return _clip("\n".join(lines)), kb.as_markup()
def build_router(texts: Texts, docker: DockerClient | None, checker: Checker | None) -> Router:
router = Router(name="status")
t = texts.status
async def list_screen() -> tuple[str, InlineKeyboardMarkup]:
if docker is None:
raise DockerError("docker is not configured (DOCKER_URL)")
return render_list(texts, await docker.containers(), utcnow())
async def card_screen(short_id: str) -> tuple[str, InlineKeyboardMarkup]:
if docker is None:
raise DockerError("docker is not configured (DOCKER_URL)")
cid = await docker.resolve(short_id)
details = await docker.inspect(cid)
stats = None
if details.state == "running":
try:
stats = await docker.stats(cid)
except DockerError:
log.warning("stats %s unavailable", short_id)
return render_card(texts, details, stats, utcnow())
@router.message(F.text == texts.menu.status)
async def open_status(message: Message, state: FSMContext) -> None:
await state.clear()
try:
text, kb = await list_screen()
except DockerError as e:
log.warning("status: %s", e)
await message.answer(t.unavailable)
return
await message.answer(text, reply_markup=kb)
@router.callback_query(StatusCb.filter())
async def on_action(query: CallbackQuery, callback_data: StatusCb) -> None:
if not isinstance(query.message, Message):
await query.answer()
return
msg = query.message
try:
match callback_data.a:
case StatusAct.LIST:
text, kb = await list_screen()
case StatusAct.CARD:
text, kb = await card_screen(callback_data.id)
case StatusAct.CHECK:
if docker is None or checker is None:
raise DockerError("docker is not configured (DOCKER_URL)")
await query.answer(t.checking)
if callback_data.id:
reports = [await checker.check(await docker.resolve(callback_data.id))]
else:
reports = await checker.check_all()
text, kb = render_report(texts, reports, utcnow(), callback_data.id)
await _edit(msg, text, kb)
return
except DockerError as e:
log.warning("status: %s", e)
await query.answer(t.unavailable, show_alert=True)
return
await _edit(msg, text, kb)
await query.answer()
return router
async def _edit(msg: Message, text: str, kb: InlineKeyboardMarkup) -> None:
# «🔄 Обновить» with nothing changed: Telegram rejects an identical edit — not an error for the user
try:
await msg.edit_text(text, reply_markup=kb)
except TelegramBadRequest as e:
if "message is not modified" not in str(e):
raise
+13
View File
@@ -84,6 +84,18 @@ class GitSettings(BaseSettings):
timeout: float = Field(default=60.0, gt=0)
class DockerSettings(BaseSettings):
model_config = SettingsConfigDict(**_ENV, env_prefix="DOCKER_")
# docker-socket-proxy (GET only, ADR-006); empty — the «Статус» section reports Docker as unavailable
url: str = ""
# Only containers with this label are shown: the host runs other projects too
label: str = "tgit.managed=true"
timeout: float = Field(default=10.0, gt=0)
snapshot_interval: float = Field(default=15.0, gt=0)
max_log_bytes: int = Field(default=5 * 1024 * 1024, gt=0)
class RuntimeSettings(BaseSettings):
model_config = _ENV
@@ -105,4 +117,5 @@ class Settings(BaseSettings):
llm: LLMSettings = Field(default_factory=LLMSettings)
sourcecraft: SourceCraftSettings = Field(default_factory=SourceCraftSettings)
git: GitSettings = Field(default_factory=GitSettings)
docker: DockerSettings = Field(default_factory=DockerSettings)
runtime: RuntimeSettings = Field(default_factory=RuntimeSettings)
+28 -4
View File
@@ -11,7 +11,7 @@ from app import __version__
from app.access.service import AccessService
from app.bot.fsm import make_storage
from app.bot.middlewares.access import AccessMiddleware
from app.bot.routers import commit, common, members
from app.bot.routers import commit, common, members, status
from app.config import Settings
from app.db.maintenance import cleanup_loop
from app.db.migrate import upgrade
@@ -21,13 +21,22 @@ from app.logs_setup import setup_logging
from app.repo.drafts import DraftService
from app.repo.git import UploadsRepo
from app.repo.publisher import Publisher
from app.runtime.checks import Checker, load_checks
from app.runtime.docker import DockerClient
from app.runtime.snapshots import SnapshotStore, snapshot_loop
from app.texts import Texts, load_texts
log = logging.getLogger("tgit")
def build_dispatcher(
texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo, commit_url_template: str = ""
texts: Texts,
sessions: SessionMaker,
access: AccessService,
repo: UploadsRepo,
commit_url_template: str = "",
docker: DockerClient | None = None,
checker: Checker | None = None,
) -> Dispatcher:
dp = Dispatcher(storage=make_storage(sessions))
gate = AccessMiddleware(access, texts)
@@ -38,31 +47,42 @@ def build_dispatcher(
dp.include_router(
commit.build_router(texts, repo, DraftService(sessions), Publisher(repo, sessions), commit_url_template)
)
dp.include_router(status.build_router(texts, docker, checker))
dp.include_router(common.build_router(texts, access))
return dp
async def run(settings: Settings) -> None:
texts = load_texts(settings.runtime.config_dir)
checks_cfg = load_checks(settings.runtime.config_dir)
engine = make_engine(settings.database.url)
await upgrade(engine)
sessions = make_sessionmaker(engine)
access = AccessService(sessions, settings.access)
await access.bootstrap_admins()
dp = build_dispatcher(texts, sessions, access, UploadsRepo(settings.git), settings.git.commit_url_template)
# No DOCKER_URL (e.g. `make run` without Docker): «Статус» answers «Docker недоступен», the rest works
docker = DockerClient(settings.docker) if settings.docker.url else None
store = SnapshotStore(sessions)
checker = Checker(docker, store, checks_cfg) if docker else None
dp = build_dispatcher(
texts, sessions, access, UploadsRepo(settings.git), settings.git.commit_url_template, docker, checker
)
rt = settings.runtime
background = [
asyncio.create_task(heartbeat(rt.heartbeat_file, rt.heartbeat_interval)),
asyncio.create_task(cleanup_loop(sessions)),
]
if docker:
background.append(asyncio.create_task(snapshot_loop(docker, store, settings.docker.snapshot_interval)))
proxy = settings.telegram.proxy_url
session = AiohttpSession(proxy=proxy.get_secret_value()) if proxy else None
bot = Bot(settings.telegram.bot_token.get_secret_value(), session=session)
log.info(
"tgit %s starting, %d bootstrap admin(s), telegram %s",
"tgit %s starting, %d bootstrap admin(s), telegram %s, docker %s",
__version__,
len(settings.access.admin_ids),
"via proxy" if proxy else "direct",
"on" if docker else "off",
)
try:
await dp.start_polling(bot, handle_signals=True)
@@ -73,6 +93,10 @@ async def run(settings: Settings) -> None:
with contextlib.suppress(asyncio.CancelledError):
await task
await bot.session.close()
if docker:
await docker.aclose()
if checker:
await checker.aclose()
await engine.dispose()
log.info("tgit stopped")
+1
View File
@@ -0,0 +1 @@
"""Containers: read-only Docker API via docker-socket-proxy, checks and snapshots (ARCHITECTURE §2.6)."""
+223
View File
@@ -0,0 +1,223 @@
"""«🩺 Запустить проверки» (SPEC S-3): independent checks per container, each with its own timeout.
A check that raises or hangs becomes ❌ with the reason; the others still run and the report is always complete.
Details are short and language-neutral (numbers, codes); the titles live in `config/texts.yaml`.
"""
import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import StrEnum
from pathlib import Path
import httpx
import regex
import yaml
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.db.models import utcnow
from app.runtime.docker import Details, DockerClient, Stats
from app.runtime.snapshots import SnapshotStore
REGEX_TIMEOUT = 1.0 # seconds per log line: a pathological pattern can't stall the bot
class ChecksConfig(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
window: int = Field(default=300, gt=0)
cpu_pct: float = Field(default=90, gt=0)
mem_pct: float = Field(default=90, gt=0, le=100)
error_pattern: str
http_timeout: float = Field(default=5, gt=0)
check_timeout: float = Field(default=15, gt=0)
@field_validator("error_pattern")
@classmethod
def _compiles(cls, value: str) -> str:
try:
regex.compile(value)
except regex.error as e: # not a ValueError: pydantic wouldn't turn it into a validation error
raise ValueError(f"error_pattern: {e}") from e
return value
def load_checks(config_dir: Path) -> ChecksConfig:
with (config_dir / "checks.yaml").open(encoding="utf-8") as f:
return ChecksConfig.model_validate(yaml.safe_load(f))
class CheckName(StrEnum):
RUNNING = "running"
HEALTH = "health"
RESTARTS = "restarts"
CPU = "cpu"
MEM = "mem"
LOG_ERRORS = "log_errors"
HTTP = "http"
class Verdict(StrEnum):
OK = "✅"
FAIL = "❌"
SKIP = "➖" # not applicable: no healthcheck, not running, no health_url
@dataclass(frozen=True)
class CheckResult:
name: CheckName
verdict: Verdict
detail: str = ""
@dataclass(frozen=True)
class ContainerReport:
details: Details
results: list[CheckResult]
@property
def ok(self) -> bool:
return all(r.verdict is not Verdict.FAIL for r in self.results)
@dataclass(frozen=True)
class Context:
details: Details
docker: DockerClient
store: SnapshotStore
cfg: ChecksConfig
http: httpx.AsyncClient
now: datetime
stats: Stats | None # None when not running
@property
def since(self) -> datetime:
return self.now - timedelta(seconds=self.cfg.window)
async def check_running(ctx: Context) -> CheckResult:
d = ctx.details
if d.state == "running":
return CheckResult(CheckName.RUNNING, Verdict.OK, d.state)
detail = f"{d.state}, exit {d.exit_code}" + (", OOM" if d.oom_killed else "")
return CheckResult(CheckName.RUNNING, Verdict.FAIL, detail)
async def check_health(ctx: Context) -> CheckResult:
health = ctx.details.health
if health is None:
return CheckResult(CheckName.HEALTH, Verdict.SKIP, "no healthcheck")
return CheckResult(CheckName.HEALTH, Verdict.OK if health == "healthy" else Verdict.FAIL, health)
async def check_restarts(ctx: Context) -> CheckResult:
now_count = ctx.details.restart_count
before = await ctx.store.oldest_since(ctx.details.name, ctx.since)
if before is None:
return CheckResult(CheckName.RESTARTS, Verdict.SKIP, f"{now_count}, no history")
grown = now_count - before.restart_count
if grown > 0:
return CheckResult(CheckName.RESTARTS, Verdict.FAIL, f"{before.restart_count} → {now_count}")
return CheckResult(CheckName.RESTARTS, Verdict.OK, str(now_count))
async def check_cpu(ctx: Context) -> CheckResult:
if ctx.stats is None:
return CheckResult(CheckName.CPU, Verdict.SKIP)
pct = ctx.stats.cpu_pct
verdict = Verdict.OK if pct < ctx.cfg.cpu_pct else Verdict.FAIL
return CheckResult(CheckName.CPU, verdict, f"{pct:.1f}% / {ctx.cfg.cpu_pct:g}%")
async def check_mem(ctx: Context) -> CheckResult:
if ctx.stats is None:
return CheckResult(CheckName.MEM, Verdict.SKIP)
pct = ctx.stats.mem_pct
verdict = Verdict.OK if pct < ctx.cfg.mem_pct else Verdict.FAIL
return CheckResult(CheckName.MEM, verdict, f"{pct:.1f}% / {ctx.cfg.mem_pct:g}%")
def count_errors(text: str, pattern: str) -> tuple[int, str]:
"""Matching lines and the first of them (trimmed)."""
compiled = regex.compile(pattern)
hits, first = 0, ""
for line in text.splitlines():
try:
matched = compiled.search(line, timeout=REGEX_TIMEOUT) is not None
except TimeoutError:
matched = False
if matched:
hits += 1
first = first or line.strip()[:120]
return hits, first
async def check_log_errors(ctx: Context) -> CheckResult:
text = await ctx.docker.logs(ctx.details.id, since=ctx.since)
hits, first = await asyncio.to_thread(count_errors, text, ctx.cfg.error_pattern)
if hits:
return CheckResult(CheckName.LOG_ERRORS, Verdict.FAIL, f"{hits}: {first}")
return CheckResult(CheckName.LOG_ERRORS, Verdict.OK, "0")
async def check_http(ctx: Context) -> CheckResult:
url = ctx.details.health_url
if url is None:
return CheckResult(CheckName.HTTP, Verdict.SKIP, "no health_url")
try:
response = await ctx.http.get(url, timeout=ctx.cfg.http_timeout)
except httpx.HTTPError as e:
return CheckResult(CheckName.HTTP, Verdict.FAIL, type(e).__name__)
ok = response.status_code < 400
return CheckResult(CheckName.HTTP, Verdict.OK if ok else Verdict.FAIL, f"HTTP {response.status_code}")
Check = Callable[[Context], Awaitable[CheckResult]]
CHECKS: list[tuple[CheckName, Check]] = [
(CheckName.RUNNING, check_running),
(CheckName.HEALTH, check_health),
(CheckName.RESTARTS, check_restarts),
(CheckName.CPU, check_cpu),
(CheckName.MEM, check_mem),
(CheckName.LOG_ERRORS, check_log_errors),
(CheckName.HTTP, check_http),
]
async def _guarded(name: CheckName, check: Check, ctx: Context) -> CheckResult:
try:
return await asyncio.wait_for(check(ctx), timeout=ctx.cfg.check_timeout)
except TimeoutError:
return CheckResult(name, Verdict.FAIL, "timeout")
except Exception as e: # one broken check must not spoil the report
return CheckResult(name, Verdict.FAIL, f"{type(e).__name__}: {str(e)[:80]}")
class Checker:
def __init__(
self, docker: DockerClient, store: SnapshotStore, cfg: ChecksConfig, http: httpx.AsyncClient | None = None
) -> None:
self._docker = docker
self._store = store
self._cfg = cfg
self._http = http or httpx.AsyncClient(follow_redirects=False)
async def check(self, container_id: str, now: datetime | None = None) -> ContainerReport:
details = await self._docker.inspect(container_id)
stats = None
if details.state == "running":
try:
stats = await asyncio.wait_for(self._docker.stats(container_id), timeout=self._cfg.check_timeout)
except Exception: # cpu/mem become ➖, the rest still runs
stats = None
ctx = Context(details, self._docker, self._store, self._cfg, self._http, now or utcnow(), stats)
results = await asyncio.gather(*(_guarded(name, fn, ctx) for name, fn in CHECKS))
return ContainerReport(details, list(results))
async def check_all(self, now: datetime | None = None) -> list[ContainerReport]:
containers = await self._docker.containers()
return list(await asyncio.gather(*(self.check(c.id, now) for c in containers)))
async def aclose(self) -> None:
await self._http.aclose()
+283
View File
@@ -0,0 +1,283 @@
"""Read-only Docker Engine API client. The bot talks to docker-socket-proxy (GET only), never to the socket itself.
Only containers carrying the managed label are visible: the host also runs other projects (Vobot, meditron),
and an id outside the list is refused before any request goes out.
"""
import json
import re
import struct
from dataclasses import dataclass
from datetime import UTC, datetime
from enum import StrEnum
from types import TracebackType
from typing import Any, Self
import httpx
from app.config import DockerSettings
SHA_LABEL = "tgit.sha"
DEPLOYED_LABEL = "tgit.deployed_at"
HEALTH_URL_LABEL = "tgit.health_url"
SERVICE_LABEL = "com.docker.compose.service"
_FRACTION = re.compile(r"(\.\d{6})\d+") # Docker gives nanoseconds, datetime takes microseconds
class DockerError(Exception):
pass
class Light(StrEnum):
"""S-1: 🟢 healthy / 🟡 starting / 🔴 unhealthy or crashed / ⚪ stopped."""
GREEN = "🟢"
YELLOW = "🟡"
RED = "🔴"
WHITE = "⚪"
def parse_time(value: str | None) -> datetime | None:
if not value or value.startswith("0001-"): # Docker's zero time: never started / no deploy label
return None
try:
return datetime.fromisoformat(_FRACTION.sub(r"\1", value).replace("Z", "+00:00")).astimezone(UTC)
except ValueError:
return None
def light(state: str, health: str | None, exit_code: int = 0) -> Light:
if state == "running":
if health == "unhealthy":
return Light.RED
if health == "starting":
return Light.YELLOW
return Light.GREEN
if state in ("restarting", "dead") or (state == "exited" and exit_code != 0):
return Light.RED
if state == "created":
return Light.YELLOW
return Light.WHITE
@dataclass(frozen=True)
class Container:
"""A list entry (GET /containers/json): enough for the overview, no extra request per container."""
id: str
name: str
service: str
image: str
state: str # created | running | paused | restarting | removing | exited | dead
health: str | None # starting | healthy | unhealthy | None — no healthcheck
exit_code: int
labels: dict[str, str]
@property
def short_id(self) -> str:
return self.id[:12]
@property
def light(self) -> Light:
return light(self.state, self.health, self.exit_code)
_HEALTH_IN_STATUS = re.compile(r"\((healthy|unhealthy|health: starting)\)")
_EXIT_IN_STATUS = re.compile(r"^Exited \((-?\d+)\)")
def _health_from_status(status: str) -> str | None:
"""`Up 2 hours (healthy)` → healthy. The list endpoint has no structured health field."""
m = _HEALTH_IN_STATUS.search(status)
if m is None:
return None
return "starting" if m.group(1) == "health: starting" else m.group(1)
@dataclass(frozen=True)
class Details:
"""GET /containers/{id}/json — the card (S-2) and most checks."""
id: str
name: str
service: str
image: str
state: str
health: str | None
exit_code: int
restart_count: int
started_at: datetime | None
finished_at: datetime | None
oom_killed: bool
labels: dict[str, str]
@property
def light(self) -> Light:
return light(self.state, self.health, self.exit_code)
@property
def sha(self) -> str | None:
return self.labels.get(SHA_LABEL) or None
@property
def deployed_at(self) -> datetime | None:
return parse_time(self.labels.get(DEPLOYED_LABEL))
@property
def health_url(self) -> str | None:
return self.labels.get(HEALTH_URL_LABEL) or None
@dataclass(frozen=True)
class Stats:
cpu_pct: float # of one CPU × online CPUs, as `docker stats` shows it
mem_bytes: int
mem_limit: int # the cgroup limit; the host's RAM when the container has none
@property
def mem_pct(self) -> float:
return 100.0 * self.mem_bytes / self.mem_limit if self.mem_limit else 0.0
def parse_stats(data: dict[str, Any]) -> Stats:
cpu, pre = data.get("cpu_stats") or {}, data.get("precpu_stats") or {}
cpu_delta = (cpu.get("cpu_usage") or {}).get("total_usage", 0) - (pre.get("cpu_usage") or {}).get("total_usage", 0)
sys_delta = cpu.get("system_cpu_usage", 0) - pre.get("system_cpu_usage", 0)
cpus = cpu.get("online_cpus") or len((cpu.get("cpu_usage") or {}).get("percpu_usage") or []) or 1
cpu_pct = 100.0 * cpu_delta / sys_delta * cpus if cpu_delta > 0 and sys_delta > 0 else 0.0
mem = data.get("memory_stats") or {}
extra = mem.get("stats") or {}
# Page cache isn't pressure: cgroup v2 reports inactive_file, v1 total_inactive_file (docker CLI does the same)
cache = extra.get("inactive_file", extra.get("total_inactive_file", 0))
usage = max(int(mem.get("usage", 0)) - int(cache), 0)
return Stats(cpu_pct=round(cpu_pct, 1), mem_bytes=usage, mem_limit=int(mem.get("limit", 0)))
def demux_logs(raw: bytes) -> str:
"""Split Docker's multiplexed stream (8-byte header per frame) — containers without a TTY send it this way."""
if raw[:1] not in (b"\x00", b"\x01", b"\x02") or len(raw) < 8 or raw[1:4] != b"\x00\x00\x00":
return raw.decode(errors="replace") # TTY container: plain bytes
out: list[bytes] = []
pos = 0
while pos + 8 <= len(raw):
(size,) = struct.unpack(">I", raw[pos + 4 : pos + 8])
out.append(raw[pos + 8 : pos + 8 + size])
pos += 8 + size
return b"".join(out).decode(errors="replace")
class DockerClient:
def __init__(self, settings: DockerSettings, transport: httpx.AsyncBaseTransport | None = None) -> None:
self._s = settings
self._http = httpx.AsyncClient(base_url=settings.url, timeout=settings.timeout, transport=transport)
self._known: set[str] = set() # ids from the last listing: the only ones card/logs/stats may touch
async def __aenter__(self) -> Self:
return self
async def __aexit__(
self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None
) -> None:
await self.aclose()
async def aclose(self) -> None:
await self._http.aclose()
async def containers(self) -> list[Container]:
"""All managed containers, stopped ones too, sorted by name."""
filters = json.dumps({"label": [self._s.label]})
data = await self._get_json("/containers/json", params={"all": "1", "filters": filters})
result = []
for c in data:
labels = c.get("Labels") or {}
names = c.get("Names") or [c["Id"][:12]]
result.append(
Container(
id=c["Id"],
name=names[0].lstrip("/"),
service=labels.get(SERVICE_LABEL, ""),
image=c.get("Image", ""),
state=c.get("State", ""),
health=_health_from_status(c.get("Status", "")),
exit_code=int(m.group(1)) if (m := _EXIT_IN_STATUS.match(c.get("Status", ""))) else 0,
labels=labels,
)
)
self._known = {c.id for c in result}
return sorted(result, key=lambda c: c.name)
async def resolve(self, short_id: str) -> str:
"""Full id of a managed container by its 12-char prefix (callback data carries only that)."""
if not self._known:
await self.containers()
for full in self._known:
if full.startswith(short_id):
return full
await self.containers() # maybe recreated since the last listing
for full in self._known:
if full.startswith(short_id):
return full
raise DockerError(f"container {short_id} is not managed by tgit")
async def inspect(self, container_id: str) -> Details:
self._guard(container_id)
d = await self._get_json(f"/containers/{container_id}/json")
state = d.get("State") or {}
labels = (d.get("Config") or {}).get("Labels") or {}
return Details(
id=d["Id"],
name=d.get("Name", "").lstrip("/"),
service=labels.get(SERVICE_LABEL, ""),
image=(d.get("Config") or {}).get("Image", ""),
state=state.get("Status", ""),
health=(state.get("Health") or {}).get("Status"),
exit_code=int(state.get("ExitCode", 0)),
restart_count=int(d.get("RestartCount", 0)),
started_at=parse_time(state.get("StartedAt")),
finished_at=parse_time(state.get("FinishedAt")),
oom_killed=bool(state.get("OOMKilled", False)),
labels=labels,
)
async def stats(self, container_id: str) -> Stats:
"""One sample with a previous one to diff against (stream=false waits ~1 s for it)."""
self._guard(container_id)
return parse_stats(await self._get_json(f"/containers/{container_id}/stats", params={"stream": "false"}))
async def logs(self, container_id: str, since: datetime | None = None, tail: int | None = None) -> str:
self._guard(container_id)
params = {"stdout": "1", "stderr": "1"}
if since is not None:
params["since"] = str(int(since.timestamp()))
if tail is not None:
params["tail"] = str(tail)
response = await self._request(f"/containers/{container_id}/logs", params)
raw = response.content[: self._s.max_log_bytes]
return demux_logs(raw)
async def ping(self) -> bool:
try:
return (await self._request("/_ping", {})).text == "OK"
except DockerError:
return False
def _guard(self, container_id: str) -> None:
if container_id not in self._known:
raise DockerError(f"container {container_id[:12]} is not managed by tgit")
async def _get_json(self, path: str, params: dict[str, str] | None = None) -> Any:
response = await self._request(path, params or {})
try:
return response.json()
except ValueError as e:
raise DockerError(f"{path}: not JSON") from e
async def _request(self, path: str, params: dict[str, str]) -> httpx.Response:
try:
response = await self._http.get(path, params=params)
except httpx.TransportError as e: # timeouts included
raise DockerError(f"docker unreachable: {type(e).__name__}") from e
if response.status_code >= 400:
raise DockerError(f"{path}: HTTP {response.status_code} {response.text[:200]}")
return response
+87
View File
@@ -0,0 +1,87 @@
"""Every 15 s: restart count, CPU and RAM of each running managed container → `container_snapshots`.
The history answers «are restarts growing over the last 5 minutes» (S-3) and feeds alerts (#15).
Old rows are removed by `app.db.maintenance`.
"""
import asyncio
import logging
from datetime import datetime
from sqlalchemy import select
from app.db.models import ContainerSnapshot, utcnow
from app.db.session import SessionMaker
from app.runtime.docker import DockerClient, DockerError
log = logging.getLogger(__name__)
class SnapshotStore:
def __init__(self, sessions: SessionMaker) -> None:
self._sessions = sessions
async def add(self, rows: list[ContainerSnapshot]) -> None:
if rows:
async with self._sessions.begin() as session:
session.add_all(rows)
async def oldest_since(self, container: str, since: datetime) -> ContainerSnapshot | None:
async with self._sessions() as session:
return await session.scalar(
select(ContainerSnapshot)
.where(ContainerSnapshot.container == container, ContainerSnapshot.at >= since)
.order_by(ContainerSnapshot.at)
.limit(1)
)
async def latest(self, container: str) -> ContainerSnapshot | None:
async with self._sessions() as session:
return await session.scalar(
select(ContainerSnapshot)
.where(ContainerSnapshot.container == container)
.order_by(ContainerSnapshot.at.desc())
.limit(1)
)
async def _snapshot(docker: DockerClient, container_id: str, at: datetime) -> ContainerSnapshot | None:
try:
details, stats = await asyncio.gather(docker.inspect(container_id), docker.stats(container_id))
except DockerError as e:
log.warning("snapshot %s: %s", container_id[:12], e)
return None
return ContainerSnapshot(
container=details.name,
at=at,
restart_count=details.restart_count,
cpu_pct=stats.cpu_pct,
mem_pct=round(stats.mem_pct, 1),
health=details.health,
)
async def collect_once(docker: DockerClient, store: SnapshotStore, now: datetime | None = None) -> int:
at = now or utcnow()
running = [c for c in await docker.containers() if c.state == "running"]
rows = await asyncio.gather(*(_snapshot(docker, c.id, at) for c in running))
snapshots = [r for r in rows if r is not None]
await store.add(snapshots)
return len(snapshots)
async def snapshot_loop(docker: DockerClient, store: SnapshotStore, interval: float) -> None:
failing = False
while True:
try:
await collect_once(docker, store)
if failing:
log.info("snapshots: docker reachable again")
failing = False
except DockerError as e:
if not failing: # one line per outage, not one every 15 s
log.warning("snapshots: %s", e)
failing = True
except Exception: # a broken round must not kill the bot
log.exception("snapshots failed")
await asyncio.sleep(interval)
+30 -1
View File
@@ -33,7 +33,6 @@ class HelpTexts(_Strict):
class SectionTexts(_Strict):
commit: str
status: str
logs: str
alerts: str
members: str
@@ -124,6 +123,35 @@ class CommitTexts(_Strict):
push_failed: str
class StatusTexts(_Strict):
title: str
line: str
empty: str
unavailable: str
card: str
none: str
no_limit: str
check_button: str
refresh: str
to_list: str
to_card: str
checking: str
report_title: str
report_container: str
report_line: str
check_running: str
check_health: str
check_restarts: str
check_cpu: str
check_mem: str
check_log_errors: str
check_http: str
days: str
hours: str
minutes: str
seconds: str
class Texts(_Strict):
menu: MenuTexts
start: StartTexts
@@ -131,6 +159,7 @@ class Texts(_Strict):
section: SectionTexts
members: MembersTexts
commit: CommitTexts
status: StatusTexts
def load_texts(config_dir: Path) -> Texts:
+13
View File
@@ -0,0 +1,13 @@
# Проверки раздела «🩺 Статус» (SPEC S-3). Читает app/runtime/checks.py; неизвестный ключ — ошибка при старте.
# Окно для «рестарты не растут» и «нет ошибок в логе», секунды
window: 300
# Пороги, % от лимита контейнера (CPU — от числа доступных ядер × 100)
cpu_pct: 90
mem_pct: 90
# Строка лога считается ошибкой, если совпала с шаблоном (синтаксис модуля regex)
error_pattern: '(?i)\b(error|exception|traceback|critical|fatal|panic)\b|"level":\s*"(ERROR|CRITICAL)"'
# HTTP-проба health_url (метка tgit.health_url): таймаут, секунды
http_timeout: 5
# Таймаут одной проверки — упавшая или зависшая проверка не роняет отчёт
check_timeout: 15
+34 -1
View File
@@ -30,7 +30,6 @@ help:
section:
commit: "📝 Коммит\n\nРаздел в разработке."
status: "🩺 Статус\n\nРаздел в разработке."
logs: "📜 Логи\n\nРаздел в разработке."
alerts: "🚨 Алерты\n\nРаздел в разработке."
members: "👥 Участники\n\nРаздел в разработке."
@@ -134,3 +133,37 @@ commit:
push_empty: "В main уже всё так же — коммитить нечего. Черновик очищен."
push_limit: "Лимит — 20 пушей в час. Попробуйте позже."
push_failed: "Не удалось запушить: репозиторий недоступен. Черновик сохранён — попробуйте позже."
status:
title: "🩺 Статус · {count} конт. · {time} UTC"
line: "{light} {name} · {state}"
empty: "Контейнеров tgit не найдено."
unavailable: "Docker недоступен, попробуйте через минуту."
card: |-
{light} {name}
Сервис: {service} · образ: {image}
Состояние: {state} · health: {health}
Аптайм: {uptime} · рестартов: {restarts}
CPU: {cpu} · RAM: {mem} из {limit} ({mem_pct})
Ревизия: {sha} · деплой: {deployed}
none: "—"
no_limit: "без лимита"
check_button: "🩺 Запустить проверки"
refresh: "🔄 Обновить"
to_list: "⬅️ К списку"
to_card: "⬅️ К контейнеру"
checking: "🩺 Проверяю контейнеры…"
report_title: "🩺 Проверки · без ошибок {ok} из {total} · {time} UTC"
report_container: "{light} {name}"
report_line: " {verdict} {title}{detail}"
check_running: "запущен"
check_health: "Docker health"
check_restarts: "рестарты за 5 мин не растут"
check_cpu: "CPU ниже порога"
check_mem: "RAM ниже порога"
check_log_errors: "нет ошибок в логе за 5 мин"
check_http: "HTTP-проба"
days: "д"
hours: "ч"
minutes: "мин"
seconds: "с"
+21
View File
@@ -6,15 +6,21 @@ services:
environment:
DATABASE_URL: postgresql+asyncpg://tgit:${POSTGRES_PASSWORD:-tgit}@postgres:5432/tgit
GIT_CLONE_DIR: /data/uploads
DOCKER_URL: http://docker-proxy:2375
volumes:
- uploads:/data
depends_on:
postgres:
condition: service_healthy
docker-proxy:
condition: service_started
restart: unless-stopped
mem_limit: 384m # ~190 MB idle; git clone/push and aiogram spikes need headroom
labels:
tgit.managed: "true"
# Shown on the «Статус» card; infra/deploy/remote.sh sets both, a local `make up` leaves them empty
tgit.sha: ${TGIT_SHA:-}
tgit.deployed_at: ${TGIT_DEPLOYED_AT:-}
logging:
options: { max-size: "10m", max-file: "3" }
@@ -38,6 +44,21 @@ services:
logging:
options: { max-size: "10m", max-file: "3" }
# The only holder of the Docker socket: GET on /containers/* and nothing else (ADR-006). The bot never sees the socket.
docker-proxy:
image: tecnativa/docker-socket-proxy:0.3.0
environment:
CONTAINERS: 1 # list, inspect, stats, logs
POST: 0 # no create/start/stop/exec — restarts go through tgit-runner (#13)
volumes:
- /var/run/docker.sock:/var/run/docker.sock:ro
restart: unless-stopped
mem_limit: 64m
labels:
tgit.managed: "true"
logging:
options: { max-size: "10m", max-file: "3" }
volumes:
pgdata:
uploads:
+5 -2
View File
@@ -84,8 +84,11 @@ pydantic-модели ответов. Методы: `list_runs`, `get_run`, `get
- **Карточка** — `inspect` (state, health, restart count, started_at, метки `tgit.sha`, `tgit.deployed_at`) + `stats`.
- **Проверки** (`runtime/checks.py`) — набор независимых функций `Check -> CheckResult(ok, detail)`:
`running`, `docker_health`, `restarts_stable` (сравнение с 5 мин назад, история из БД), `cpu_ok`, `mem_ok`,
`no_errors_in_logs` (5 мин, шаблон из `config/checks.yaml`), `http_probe` (если в `services/<name>/tgit.yaml`
объявлен `health_url`; проба из сети сервиса через runner). Результат — отчёт-таблица.
`no_errors_in_logs` (5 мин, шаблон из `config/checks.yaml`), `http_probe` (если у контейнера есть метка
`tgit.health_url`; пока нет runner (#13) — проба из сети бота). Каждая проверка — со своим таймаутом: упавшая или
зависшая даёт ❌ с причиной, остальные выполняются. Пороги и окно — в `config/checks.yaml`.
- **Изоляция** — прокси отдаёт контейнеры всего хоста, поэтому бот сам фильтрует по метке `tgit.managed=true` и
отказывает в `inspect`/`stats`/`logs` по id не из этого списка ещё до запроса.
- **Мониторинг** — поток `events` (`die`, `oom`, `health_status`, `restart`) и опрос `stats` раз в 15 с → события для `alerts`.
### 2.7 `app/logs` — grep-движок
+3
View File
@@ -17,6 +17,9 @@ if [ -f "release-$SHA.tgz" ]; then
fi
cd app
# Labels on the bot container, shown on its «Статус» card
export TGIT_SHA=$SHA TGIT_DEPLOYED_AT
TGIT_DEPLOYED_AT=$(date -u +%Y-%m-%dT%H:%M:%SZ)
docker compose -p "$PROJECT" up -d --build --remove-orphans </dev/null
status=""
+1
View File
@@ -27,6 +27,7 @@ dev = [
"pytest-cov>=5.0",
"respx>=0.21",
"aiosqlite>=0.20",
"types-regex>=2026.9.29.20261017",
]
[build-system]
+1 -1
View File
@@ -46,7 +46,7 @@ async def test_outsider_stale_button_is_swallowed(tg: TgClient) -> None:
assert len(tg.session.sent(AnswerCallbackQuery)) == 1
@pytest.mark.parametrize("section", ["status", "logs", "alerts"])
@pytest.mark.parametrize("section", ["logs", "alerts"])
async def test_menu_button_opens_section_with_nav(tg: TgClient, texts: Texts, section: str) -> None:
await tg.send(ADMIN, getattr(texts.menu, section))
+143
View File
@@ -0,0 +1,143 @@
from collections.abc import AsyncIterator
import httpx
import pytest
from aiogram.methods import AnswerCallbackQuery, EditMessageText, SendMessage
from app.access.service import AccessService
from app.bot.routers.status import StatusAct, StatusCb, duration
from app.db.session import SessionMaker
from app.main import build_dispatcher
from app.repo.git import UploadsRepo
from app.runtime.checks import Checker, load_checks
from app.runtime.docker import DockerClient
from app.runtime.snapshots import SnapshotStore
from app.texts import Texts
from tests.conftest import ADMIN, ROOT
from tests.fakes import TgClient
from tests.runtime.fake_docker import FakeDocker
@pytest.fixture
def fake() -> FakeDocker:
fake = FakeDocker()
fake.add("tgit-dev-bot-1", service="bot", labels={"tgit.sha": "22d9f0c5abcdef0123"})
fake.add("unhealthy", service="unhealthy", health="unhealthy")
fake.add("vobot-api-1", managed=False)
return fake
@pytest.fixture
async def docker(fake: FakeDocker) -> AsyncIterator[DockerClient]:
async with fake.client() as client:
yield client
@pytest.fixture
def tg(
texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo, docker: DockerClient
) -> TgClient:
checker = Checker(
docker,
SnapshotStore(sessions),
load_checks(ROOT / "config"),
httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200))),
)
return TgClient(build_dispatcher(texts, sessions, access, repo, docker=docker, checker=checker))
def short(fake: FakeDocker, name: str) -> str:
return next(c.id[:12] for c in fake.containers.values() if c.name == name)
async def test_list_shows_managed_containers_with_lights(tg: TgClient, texts: Texts) -> None:
await tg.send(ADMIN, texts.menu.status)
[reply] = tg.session.sent(SendMessage)
assert "🟢 tgit-dev-bot-1 · running, healthy" in reply.text
assert "🔴 unhealthy · running, unhealthy" in reply.text
assert "vobot" not in reply.text
labels = [b.text for row in reply.reply_markup.inline_keyboard for b in row]
assert texts.status.check_button in labels and texts.status.refresh in labels
async def test_card(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
await tg.send(ADMIN, texts.menu.status)
await tg.press(ADMIN, StatusCb(a=StatusAct.CARD, id=short(fake, "tgit-dev-bot-1")).pack())
[card] = tg.session.sent(EditMessageText)
assert card.text.startswith("🟢 tgit-dev-bot-1")
assert "Ревизия: 22d9f0c5abcd" in card.text
assert "CPU: 5.0% · RAM: 64.0 МБ из 256.0 МБ (25%)" in card.text
assert "рестартов: 0" in card.text
async def test_card_of_stopped_container(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
fake.add("stopped", state="exited", health=None, exit_code=1)
await tg.press(ADMIN, StatusCb(a=StatusAct.CARD, id=short(fake, "stopped")).pack())
[card] = tg.session.sent(EditMessageText)
assert card.text.startswith("🔴 stopped")
assert "exited (exit 1)" in card.text
assert f"CPU: {texts.status.none}" in card.text
async def test_run_all_checks(tg: TgClient, texts: Texts) -> None:
await tg.press(ADMIN, StatusCb(a=StatusAct.CHECK).pack())
[report] = tg.session.sent(EditMessageText)
assert report.text.startswith("🩺 Проверки · без ошибок 1 из 2")
assert "✅ tgit-dev-bot-1" in report.text
assert "❌ unhealthy" in report.text
assert f"❌ {texts.status.check_health} · unhealthy" in report.text
[ack, *_] = tg.session.sent(AnswerCallbackQuery)
assert ack.text == texts.status.checking
async def test_checks_of_one_container_offer_way_back(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
await tg.press(ADMIN, StatusCb(a=StatusAct.CHECK, id=short(fake, "unhealthy")).pack())
[report] = tg.session.sent(EditMessageText)
assert "без ошибок 0 из 1" in report.text
labels = [b.text for row in report.reply_markup.inline_keyboard for b in row]
assert texts.status.to_card in labels
async def test_foreign_or_vanished_container_is_refused(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
await tg.press(ADMIN, StatusCb(a=StatusAct.CARD, id=short(fake, "vobot-api-1")).pack())
assert not tg.session.sent(EditMessageText)
[alert] = tg.session.sent(AnswerCallbackQuery)
assert alert.text == texts.status.unavailable and alert.show_alert
async def test_docker_down(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
fake.down = True
await tg.send(ADMIN, texts.menu.status)
assert [r.text for r in tg.session.sent(SendMessage)] == [texts.status.unavailable]
async def test_without_docker_configured(
texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo
) -> None:
tg = TgClient(build_dispatcher(texts, sessions, access, repo))
await tg.send(ADMIN, texts.menu.status)
await tg.press(ADMIN, StatusCb(a=StatusAct.CHECK).pack())
assert [r.text for r in tg.session.sent(SendMessage)] == [texts.status.unavailable]
assert tg.session.sent(AnswerCallbackQuery)[0].text == texts.status.unavailable
async def test_refresh_list(tg: TgClient, texts: Texts) -> None:
await tg.press(ADMIN, StatusCb(a=StatusAct.LIST).pack())
[edit] = tg.session.sent(EditMessageText)
assert edit.text.startswith("🩺 Статус · 2 конт.")
@pytest.mark.parametrize(
("seconds", "expected"), [(5, "5 с"), (125, "2 мин"), (3 * 3600 + 120, "3 ч 2 мин"), (2 * 86400 + 3600, "2 д 1 ч")]
)
def test_duration(texts: Texts, seconds: float, expected: str) -> None:
assert duration(texts.status, seconds) == expected
View File
+160
View File
@@ -0,0 +1,160 @@
"""An in-memory Docker Engine API behind httpx.MockTransport, GET-only like docker-socket-proxy."""
import hashlib
import json
import struct
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Any
from urllib.parse import parse_qs
import httpx
from app.config import DockerSettings
from app.runtime.docker import DockerClient
STARTED = datetime(2026, 10, 17, 8, 0, tzinfo=UTC)
MiB = 1024 * 1024
def frame(text: str, stream: int = 1) -> bytes:
data = text.encode()
return struct.pack(">BxxxI", stream, len(data)) + data
@dataclass
class FakeContainer:
id: str
name: str
service: str = ""
state: str = "running"
health: str | None = "healthy"
exit_code: int = 0
restart_count: int = 0
oom: bool = False
managed: bool = True
labels: dict[str, str] = field(default_factory=dict)
logs: list[tuple[datetime, str]] = field(default_factory=list) # (when, line)
cpu_pct: float = 5.0
mem_bytes: int = 64 * MiB
mem_limit: int = 256 * MiB
stats_status: int = 200
def all_labels(self) -> dict[str, str]:
base = {"com.docker.compose.service": self.service} if self.service else {}
return {**base, **({"tgit.managed": "true"} if self.managed else {}), **self.labels}
def status_line(self) -> str:
if self.state == "running":
return "Up 2 hours" + {
"healthy": " (healthy)",
"unhealthy": " (unhealthy)",
"starting": " (health: starting)",
}.get(self.health or "", "")
if self.state == "exited":
return f"Exited ({self.exit_code}) 5 minutes ago"
return self.state.capitalize()
class FakeDocker:
def __init__(self) -> None:
self.containers: dict[str, FakeContainer] = {}
self.requests: list[httpx.Request] = []
self.down = False
def add(self, name: str, **kw: Any) -> FakeContainer:
cid = hashlib.sha256(name.encode()).hexdigest()
c = FakeContainer(id=cid, name=name, **kw)
self.containers[cid] = c
return c
def client(self, **settings: Any) -> DockerClient:
return DockerClient(
DockerSettings(url="http://docker-proxy:2375", **settings), transport=httpx.MockTransport(self)
)
def __call__(self, request: httpx.Request) -> httpx.Response:
self.requests.append(request)
if self.down:
raise httpx.ConnectError("connection refused", request=request)
if request.method != "GET":
return httpx.Response(403, text="forbidden by proxy")
path = request.url.path
q = {k: v[0] for k, v in parse_qs(request.url.query.decode()).items()}
if path == "/_ping":
return httpx.Response(200, text="OK")
if path == "/containers/json":
return httpx.Response(200, json=self._list(q))
parts = path.strip("/").split("/")
if len(parts) == 3 and parts[0] == "containers" and parts[1] in self.containers:
c = self.containers[parts[1]]
match parts[2]:
case "json":
return httpx.Response(200, json=self._inspect(c))
case "stats":
if c.stats_status != 200:
return httpx.Response(c.stats_status, text="stats broken")
return httpx.Response(200, json=self._stats(c))
case "logs":
since = datetime.fromtimestamp(int(q.get("since", "0")), UTC)
body = b"".join(frame(line + "\n", 2 if "ERR" in line else 1) for at, line in c.logs if at >= since)
return httpx.Response(200, content=body)
return httpx.Response(404, json={"message": "No such container"})
def _list(self, q: dict[str, str]) -> list[dict[str, Any]]:
wanted = json.loads(q.get("filters", "{}")).get("label", [])
out = []
for c in self.containers.values():
labels = c.all_labels()
if all(labels.get(k) == v for k, _, v in (w.partition("=") for w in wanted)):
out.append(
{
"Id": c.id,
"Names": [f"/{c.name}"],
"Image": f"{c.name}:latest",
"State": c.state,
"Status": c.status_line(),
"Labels": labels,
}
)
return out
def _inspect(self, c: FakeContainer) -> dict[str, Any]:
state: dict[str, Any] = {
"Status": c.state,
"ExitCode": c.exit_code,
"OOMKilled": c.oom,
"StartedAt": STARTED.strftime("%Y-%m-%dT%H:%M:%S.123456789Z"),
"FinishedAt": "0001-01-01T00:00:00Z",
}
if c.health is not None:
state["Health"] = {"Status": c.health}
return {
"Id": c.id,
"Name": f"/{c.name}",
"RestartCount": c.restart_count,
"State": state,
"Config": {"Image": f"{c.name}:latest", "Labels": c.all_labels()},
}
@staticmethod
def _stats(c: FakeContainer) -> dict[str, Any]:
# 2 CPUs, system delta 1e9 → cpu_pct = delta/1e9*2*100
delta = int(c.cpu_pct / 200 * 1e9)
return {
"cpu_stats": {
"cpu_usage": {"total_usage": 10_000 + delta},
"system_cpu_usage": 2_000_000_000,
"online_cpus": 2,
},
"precpu_stats": {"cpu_usage": {"total_usage": 10_000}, "system_cpu_usage": 1_000_000_000},
"memory_stats": {
"usage": c.mem_bytes + 10 * MiB,
"limit": c.mem_limit,
"stats": {"inactive_file": 10 * MiB},
},
}
def now_minus(minutes: float) -> datetime:
return datetime.now(UTC) - timedelta(minutes=minutes)
+203
View File
@@ -0,0 +1,203 @@
import asyncio
import contextlib
from collections.abc import AsyncIterator
from datetime import UTC, datetime, timedelta
import httpx
import pytest
from app.db.models import ContainerSnapshot
from app.db.session import SessionMaker
from app.runtime import checks
from app.runtime.checks import Checker, CheckName, ChecksConfig, Verdict, count_errors, load_checks
from app.runtime.docker import DockerClient, Light
from app.runtime.snapshots import SnapshotStore, collect_once, snapshot_loop
from tests.conftest import ROOT
from tests.runtime.fake_docker import FakeDocker, now_minus
CFG = load_checks(ROOT / "config")
def probe(request: httpx.Request) -> httpx.Response:
if request.url.host == "broken":
raise httpx.ConnectError("refused", request=request)
return httpx.Response(200 if request.url.path == "/health" else 503)
@pytest.fixture
def fake() -> FakeDocker:
fake = FakeDocker()
fake.add("hello", service="hello", labels={"tgit.health_url": "http://hello:8000/health"})
fake.add(
"unhealthy", service="unhealthy", health="unhealthy", labels={"tgit.health_url": "http://unhealthy:8000/x"}
)
fake.add("crashed", state="exited", health=None, exit_code=137, oom=True)
return fake
@pytest.fixture
async def docker(fake: FakeDocker) -> AsyncIterator[DockerClient]:
async with fake.client() as client:
await client.containers() # the bot always lists before it inspects
yield client
@pytest.fixture
def store(sessions: SessionMaker) -> SnapshotStore:
return SnapshotStore(sessions)
@pytest.fixture
def checker(docker: DockerClient, store: SnapshotStore) -> Checker:
return Checker(docker, store, CFG, httpx.AsyncClient(transport=httpx.MockTransport(probe)))
def verdicts(report: checks.ContainerReport) -> dict[CheckName, Verdict]:
return {r.name: r.verdict for r in report.results}
async def test_config_loads_and_bad_pattern_rejected() -> None:
assert CFG.window == 300
with pytest.raises(ValueError):
ChecksConfig(error_pattern="(unclosed")
async def test_healthy_vs_unhealthy_vs_crashed(checker: Checker) -> None:
reports = {r.details.name: r for r in await checker.check_all()}
hello = verdicts(reports["hello"])
assert hello[CheckName.RUNNING] is Verdict.OK
assert hello[CheckName.HEALTH] is Verdict.OK
assert hello[CheckName.HTTP] is Verdict.OK
assert hello[CheckName.RESTARTS] is Verdict.SKIP # no history yet
assert reports["hello"].ok
bad = verdicts(reports["unhealthy"])
assert bad[CheckName.HEALTH] is Verdict.FAIL
assert bad[CheckName.HTTP] is Verdict.FAIL
assert not reports["unhealthy"].ok
crashed = reports["crashed"]
assert verdicts(crashed)[CheckName.RUNNING] is Verdict.FAIL
assert "exit 137, OOM" in crashed.results[0].detail
assert verdicts(crashed)[CheckName.CPU] is Verdict.SKIP
assert crashed.details.light is Light.RED
async def test_restarts_compared_with_history(
checker: Checker, fake: FakeDocker, docker: DockerClient, store: SnapshotStore
) -> None:
hello = next(c for c in fake.containers.values() if c.name == "hello")
assert await collect_once(docker, store, now=datetime.now(UTC) - timedelta(minutes=3)) == 2 # running ones only
hello.restart_count = 2
report = await checker.check(hello.id)
result = next(r for r in report.results if r.name is CheckName.RESTARTS)
assert result.verdict is Verdict.FAIL
assert result.detail == "0 → 2"
async def test_old_history_outside_window_is_ignored(
checker: Checker, fake: FakeDocker, sessions: SessionMaker
) -> None:
hello = next(c for c in fake.containers.values() if c.name == "hello")
async with sessions.begin() as s:
s.add(
ContainerSnapshot(container="hello", at=now_minus(30), restart_count=0, cpu_pct=1, mem_pct=1, health=None)
)
hello.restart_count = 5
report = await checker.check(hello.id)
assert verdicts(report)[CheckName.RESTARTS] is Verdict.SKIP
async def test_thresholds_and_log_errors(checker: Checker, fake: FakeDocker) -> None:
hello = next(c for c in fake.containers.values() if c.name == "hello")
hello.cpu_pct, hello.mem_bytes = 150.0, 250 * 1024 * 1024
hello.logs = [
(now_minus(20), "Traceback from long ago"),
(now_minus(1), "all good"),
(now_minus(1), '{"level": "ERROR", "msg": "boom"}'),
]
report = await checker.check(hello.id)
v = verdicts(report)
assert v[CheckName.CPU] is Verdict.FAIL
assert v[CheckName.MEM] is Verdict.FAIL
log_result = next(r for r in report.results if r.name is CheckName.LOG_ERRORS)
assert log_result.verdict is Verdict.FAIL
assert log_result.detail.startswith("1: ") # the old Traceback is outside the window
async def test_broken_or_slow_check_does_not_spoil_report(
checker: Checker, fake: FakeDocker, monkeypatch: pytest.MonkeyPatch
) -> None:
async def boom(ctx: checks.Context) -> checks.CheckResult:
raise RuntimeError("kaputt")
async def slow(ctx: checks.Context) -> checks.CheckResult:
await asyncio.sleep(10)
raise AssertionError
monkeypatch.setattr(checks, "CHECKS", [(CheckName.HEALTH, boom), (CheckName.HTTP, slow), *checks.CHECKS[:1]])
fast = Checker(checker._docker, checker._store, CFG.model_copy(update={"check_timeout": 0.05}), checker._http)
hello = next(c for c in fake.containers.values() if c.name == "hello")
report = await fast.check(hello.id)
by_name = {r.name: r for r in report.results}
assert by_name[CheckName.HEALTH].detail == "RuntimeError: kaputt"
assert by_name[CheckName.HTTP].detail == "timeout"
assert by_name[CheckName.RUNNING].verdict is Verdict.OK
async def test_stats_failure_skips_cpu_and_mem(checker: Checker, fake: FakeDocker) -> None:
hello = next(c for c in fake.containers.values() if c.name == "hello")
hello.stats_status = 500
report = await checker.check(hello.id)
assert verdicts(report)[CheckName.CPU] is Verdict.SKIP
assert verdicts(report)[CheckName.RUNNING] is Verdict.OK
async def test_http_probe_connection_error(fake: FakeDocker, checker: Checker) -> None:
hello = next(c for c in fake.containers.values() if c.name == "hello")
hello.labels["tgit.health_url"] = "http://broken/health"
report = await checker.check(hello.id)
result = next(r for r in report.results if r.name is CheckName.HTTP)
assert (result.verdict, result.detail) == (Verdict.FAIL, "ConnectError")
async def test_snapshot_store_latest(docker: DockerClient, store: SnapshotStore) -> None:
await collect_once(docker, store, now=now_minus(1))
await collect_once(docker, store)
latest = await store.latest("hello")
assert latest is not None and latest.cpu_pct == 5.0 and latest.health == "healthy"
def test_count_errors_reports_first_hit() -> None:
hits, first = count_errors("ok\nERROR one\nERROR two\n", r"ERROR")
assert (hits, first) == (2, "ERROR one")
async def test_snapshot_loop_survives_docker_outage(
fake: FakeDocker, docker: DockerClient, store: SnapshotStore
) -> None:
fake.down = True
task = asyncio.create_task(snapshot_loop(docker, store, interval=0.01))
await asyncio.sleep(0.05)
fake.down = False
await asyncio.sleep(0.05)
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
assert await store.latest("hello") is not None
+110
View File
@@ -0,0 +1,110 @@
import pytest
from app.runtime.docker import DockerError, Light, demux_logs, light, parse_stats, parse_time
from tests.runtime.fake_docker import FakeDocker, MiB, frame, now_minus
@pytest.fixture
def fake() -> FakeDocker:
fake = FakeDocker()
fake.add(
"tgit-dev-bot-1", service="bot", labels={"tgit.sha": "526eb56aa", "tgit.deployed_at": "2026-10-17T07:52:00Z"}
)
fake.add("tgit-dev-postgres-1", service="postgres")
fake.add("vobot-api-1", managed=False) # someone else's project on the same host
return fake
async def test_lists_only_managed_containers(fake: FakeDocker) -> None:
async with fake.client() as docker:
names = [c.name for c in await docker.containers()]
assert names == ["tgit-dev-bot-1", "tgit-dev-postgres-1"]
assert all(r.method == "GET" for r in fake.requests)
async def test_foreign_container_is_refused_without_a_request(fake: FakeDocker) -> None:
foreign = next(c for c in fake.containers.values() if not c.managed)
async with fake.client() as docker:
await docker.containers()
fake.requests.clear()
with pytest.raises(DockerError):
await docker.inspect(foreign.id)
with pytest.raises(DockerError):
await docker.logs(foreign.id)
with pytest.raises(DockerError):
await docker.resolve(foreign.id[:12])
assert all("/containers/json" in str(r.url) for r in fake.requests) # only re-listings, never the foreign id
async def test_inspect_and_stats(fake: FakeDocker) -> None:
async with fake.client() as docker:
bot = next(c for c in await docker.containers() if c.service == "bot")
details = await docker.inspect(await docker.resolve(bot.short_id))
stats = await docker.stats(bot.id)
assert details.light is Light.GREEN
assert details.sha == "526eb56aa"
assert details.deployed_at is not None and details.deployed_at.hour == 7
assert details.started_at is not None and details.started_at.microsecond == 123456 # nanoseconds trimmed
assert stats.cpu_pct == 5.0
assert stats.mem_bytes == 64 * MiB # page cache subtracted
assert stats.mem_pct == 25.0
async def test_logs_are_demultiplexed_and_since_filtered(fake: FakeDocker) -> None:
bot = next(c for c in fake.containers.values() if c.service == "bot")
bot.logs = [(now_minus(10), "old line"), (now_minus(1), "fresh line"), (now_minus(0.5), "ERR on stderr")]
async with fake.client() as docker:
await docker.containers()
text = await docker.logs(bot.id, since=now_minus(5))
assert text == "fresh line\nERR on stderr\n"
async def test_unreachable_docker_is_a_docker_error(fake: FakeDocker) -> None:
fake.down = True
async with fake.client() as docker:
assert not await docker.ping()
with pytest.raises(DockerError, match="unreachable"):
await docker.containers()
async def test_ping(fake: FakeDocker) -> None:
async with fake.client() as docker:
assert await docker.ping()
def test_demux_plain_tty_stream_passes_through() -> None:
assert demux_logs(b"plain text\n") == "plain text\n"
assert demux_logs(frame("a\n") + frame("b\n", 2)) == "a\nb\n"
@pytest.mark.parametrize(
("state", "health", "code", "expected"),
[
("running", "healthy", 0, Light.GREEN),
("running", None, 0, Light.GREEN),
("running", "starting", 0, Light.YELLOW),
("running", "unhealthy", 0, Light.RED),
("restarting", None, 0, Light.RED),
("exited", None, 137, Light.RED),
("exited", None, 0, Light.WHITE),
("created", None, 0, Light.YELLOW),
],
)
def test_light(state: str, health: str | None, code: int, expected: Light) -> None:
assert light(state, health, code) is expected
def test_parse_time_edge_cases() -> None:
assert parse_time(None) is None
assert parse_time("0001-01-01T00:00:00Z") is None
assert parse_time("garbage") is None
def test_parse_stats_without_previous_sample() -> None:
stats = parse_stats({"cpu_stats": {"cpu_usage": {"total_usage": 5}}, "memory_stats": {}})
assert stats.cpu_pct == 0.0
assert stats.mem_pct == 0.0
Generated
+11
View File
@@ -807,6 +807,7 @@ dev = [
{ name = "respx" },
{ name = "ruff" },
{ name = "types-pyyaml" },
{ name = "types-regex" },
]
[package.metadata]
@@ -834,6 +835,7 @@ dev = [
{ name = "respx", specifier = ">=0.21" },
{ name = "ruff", specifier = ">=0.6" },
{ name = "types-pyyaml", specifier = ">=6.0" },
{ name = "types-regex", specifier = ">=2026.9.29.20261017" },
]
[[package]]
@@ -845,6 +847,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/15/c0/fc0644b7ddcfb969e95845837143cb5173ddd6e06ee4ba5fc493cd9329b7/types_pyyaml-6.0.12.20260906-py3-none-any.whl", hash = "sha256:bca893ff0d51df5c9053137d5d0e6ccd36e939a196356f1d5c16372422f5137b", size = 21282, upload-time = "2026-09-06T06:35:34.372Z" },
]
[[package]]
name = "types-regex"
version = "2026.9.29.20261017"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/a2/bb/15d3c5f2f21ebf040430970e57bf8c438701e21ecd155607860e1dcff4f2/types_regex-2026.9.29.20261017.tar.gz", hash = "sha256:1eb0d21f77331e0f87268eeb1a75c4ac8025665a211c6e2e8ba084504b1c299f", size = 13971, upload-time = "2026-10-17T06:20:04.659Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/bd/2d/4a7cde6131caf872b33d25a66ffd938ebd10758e84527eb3c456cb558e77/types_regex-2026.9.29.20261017-py3-none-any.whl", hash = "sha256:d0bc5d8f44d2039f40f52b272e672266ba6cc1710956c6c7f5045000fea433b8", size = 11194, upload-time = "2026-10-17T06:20:03.623Z" },
]
[[package]]
name = "typing-extensions"
version = "4.16.0"