From 48a9b203799dba73b08ba8ffcb53d0522096459d Mon Sep 17 00:00:00 2001 From: JiriUhlir <149317995+JiriUhlir@users.noreply.github.com> Date: Fri, 29 May 2026 10:35:56 +0200 Subject: [PATCH] workers --- app/db/migrations.py | 13 +++ app/db/workers.py | 89 +++++++++++++++++++++ app/main.py | 3 +- app/routes/workers.py | 172 ++++++++++++++++++++++++++++++++++++++++ app/templates/layout.py | 1 + 5 files changed, 277 insertions(+), 1 deletion(-) create mode 100644 app/db/workers.py create mode 100644 app/routes/workers.py diff --git a/app/db/migrations.py b/app/db/migrations.py index e03fe81..4b49928 100644 --- a/app/db/migrations.py +++ b/app/db/migrations.py @@ -38,10 +38,23 @@ def run_migrations(): ) """ ) + con.execute( + """ + CREATE TABLE IF NOT EXISTS workers ( + worker_id TEXT PRIMARY KEY, + status TEXT, + last_seen_at TEXT, + current_job_id INTEGER, + started_at TEXT, + metadata_json TEXT + ) + """ + ) con.execute("CREATE INDEX IF NOT EXISTS idx_jobs_status_created ON jobs(status, created_at)") con.execute("CREATE INDEX IF NOT EXISTS idx_jobs_target_status ON jobs(target_type, target_id, status)") con.execute("CREATE INDEX IF NOT EXISTS idx_job_logs_job_id ON job_logs(job_id, id)") + con.execute("CREATE INDEX IF NOT EXISTS idx_workers_last_seen ON workers(last_seen_at)") try: con.execute("ALTER TABLE jobs ADD COLUMN created_by_display_name TEXT") diff --git a/app/db/workers.py b/app/db/workers.py new file mode 100644 index 0000000..4c446e2 --- /dev/null +++ b/app/db/workers.py @@ -0,0 +1,89 @@ +from datetime import datetime, timedelta, timezone + +from app.db.database import get_connection +from app.db.migrations import run_migrations + + +def parse_timestamp(value: str | None) -> datetime | None: + if not value: + return None + + normalized = value.strip().replace("Z", "+00:00") + formats = [ + "%Y-%m-%d %H:%M:%S", + "%Y-%m-%d %H:%M:%S.%f", + "%Y-%m-%dT%H:%M:%S", + "%Y-%m-%dT%H:%M:%S.%f", + ] + + try: + parsed = datetime.fromisoformat(normalized) + return parsed.astimezone(timezone.utc).replace(tzinfo=None) if parsed.tzinfo else parsed + except ValueError: + pass + + for fmt in formats: + try: + return datetime.strptime(normalized, fmt) + except ValueError: + continue + + return None + + +def is_worker_online(last_seen_at: str | None) -> bool: + last_seen = parse_timestamp(last_seen_at) + if not last_seen: + return False + + return last_seen >= datetime.utcnow() - timedelta(seconds=30) + + +def with_online_status(worker: dict): + worker["online"] = is_worker_online(worker.get("last_seen_at")) + return worker + + +def get_workers(): + run_migrations() + con = get_connection() + + rows = con.execute( + """ + SELECT + worker_id, + status, + last_seen_at, + current_job_id, + started_at, + metadata_json + FROM workers + ORDER BY worker_id + """ + ).fetchall() + + con.close() + return [with_online_status(dict(row)) for row in rows] + + +def get_worker(worker_id: str): + run_migrations() + con = get_connection() + + row = con.execute( + """ + SELECT + worker_id, + status, + last_seen_at, + current_job_id, + started_at, + metadata_json + FROM workers + WHERE worker_id = ? + """, + (worker_id,), + ).fetchone() + + con.close() + return with_online_status(dict(row)) if row else None diff --git a/app/main.py b/app/main.py index 5ea3c09..4fffa03 100644 --- a/app/main.py +++ b/app/main.py @@ -5,7 +5,7 @@ from fastapi.staticfiles import StaticFiles from starlette.middleware.sessions import SessionMiddleware from .config import read_env_value -from .routes import apps, audit, auth, backups, deployments, health, jobs +from .routes import apps, audit, auth, backups, deployments, health, jobs, workers def create_app() -> FastAPI: @@ -28,6 +28,7 @@ def create_app() -> FastAPI: app.include_router(backups.router) app.include_router(deployments.router) app.include_router(jobs.router) + app.include_router(workers.router) app.include_router(audit.router) return app diff --git a/app/routes/workers.py b/app/routes/workers.py new file mode 100644 index 0000000..c2db0d5 --- /dev/null +++ b/app/routes/workers.py @@ -0,0 +1,172 @@ +import html +import json +from urllib.parse import quote + +from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi.responses import HTMLResponse + +from app.auth import require_user +from app.db.audit import log_audit_event +from app.db.workers import get_worker, get_workers +from app.templates.layout import page + +router = APIRouter() + + +def render_worker_badge(online: bool) -> str: + if online: + return 'ONLINE' + return 'OFFLINE' + + +def pretty_json(value: str | None) -> str: + if not value: + return "{}" + + try: + return json.dumps(json.loads(value), ensure_ascii=False, indent=2, sort_keys=True) + except (TypeError, ValueError): + return value + + +@router.get("/workers", response_class=HTMLResponse) +def workers_page(request: Request, user=Depends(require_user)): + workers = get_workers() + online_count = sum(1 for worker in workers if worker.get("online")) + total_count = len(workers) + offline_count = total_count - online_count + + log_audit_event( + user, + action="workers.view", + target_type="workers", + metadata={ + "online": online_count, + "offline": offline_count, + "total": total_count, + }, + ) + + rows = "" + for worker in workers: + worker_id_raw = worker.get("worker_id", "") or "" + worker_id = html.escape(worker_id_raw) + worker_url_id = quote(worker_id_raw, safe="") + status = html.escape(worker.get("status", "") or "") + last_seen_at = html.escape(worker.get("last_seen_at", "") or "") + started_at = html.escape(worker.get("started_at", "") or "") + badge = render_worker_badge(bool(worker.get("online"))) + current_job_id = worker.get("current_job_id") + current_job = "" + if current_job_id is not None: + current_job_label = html.escape(str(current_job_id)) + current_job = f'#{current_job_label}' + + rows += f""" + + + {worker_id} + + {badge}
{status} + {last_seen_at} + {current_job} + {started_at} + Detail + + """ + + if not rows: + rows = 'Zatím nejsou evidovaní žádní workeři.' + + return page( + "Workers", + f""" +
+

Workers

+

Přehled worker procesů obsluhujících AppFactory joby.

+
+ +
+
Online Workers{online_count}
+
Offline Workers{offline_count}
+
Total Workers{total_count}
+
+ +
+

Workers

+ + + + + + + + + + {rows} +
Worker IDStatusLast SeenCurrent JobStarted AtAkce
+
+ """, + user=user, + ) + + +@router.get("/workers/{worker_id}", response_class=HTMLResponse) +def worker_detail_page(worker_id: str, request: Request, user=Depends(require_user)): + worker = get_worker(worker_id) + if not worker: + raise HTTPException(status_code=404, detail="Worker not found") + + log_audit_event( + user, + action="workers.view", + target_type="worker", + target_id=worker_id, + ) + + worker_id_html = html.escape(worker.get("worker_id", "") or "") + status = html.escape(worker.get("status", "") or "") + last_seen_at = html.escape(worker.get("last_seen_at", "") or "") + started_at = html.escape(worker.get("started_at", "") or "") + badge = render_worker_badge(bool(worker.get("online"))) + current_job_id = worker.get("current_job_id") + current_job = "" + if current_job_id is not None: + current_job_label = html.escape(str(current_job_id)) + current_job = f'#{current_job_label}' + + metadata = html.escape(pretty_json(worker.get("metadata_json"))) + metadata_block = "" + if worker.get("metadata_json"): + metadata_block = f""" +
+

Metadata

+ 
{metadata}
+
+ """ + + return page( + f"Worker {worker_id_html}", + f""" +
+

Worker {worker_id_html}

+

+ ← Zpět na workers +

+
+ +
+

Souhrn

+ + + + + + +
Worker ID{worker_id_html}
Status{badge}
{status}
Last Seen{last_seen_at}
Current Job{current_job}
Started At{started_at}
+
+ + {metadata_block} + """, + user=user, + ) diff --git a/app/templates/layout.py b/app/templates/layout.py index 17447b9..43cfd3f 100644 --- a/app/templates/layout.py +++ b/app/templates/layout.py @@ -16,6 +16,7 @@ def page(title: str, body: str, user=None) -> str: Nová aplikace Nasazení Joby + Operations -> Workers Zálohy Audit