add monitor scrappers
This commit is contained in:
lekss361
parent
bfee5ea916
commit
7c4197000f
3 changed files with 358 additions and 1 deletions
|
|
@ -67,6 +67,96 @@ def trigger_kn_sweep(
|
|||
}
|
||||
|
||||
|
||||
@router.get("/queue")
|
||||
def queue_status(
|
||||
x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Snapshot of Celery worker(s) state: active tasks, reserved (queued on
|
||||
worker, not yet started), scheduled (with ETA), beat-schedule, and queue
|
||||
depth in Redis. If no worker is running, all worker-side fields are empty.
|
||||
"""
|
||||
_check_token(x_admin_token)
|
||||
|
||||
from app.workers.celery_app import celery_app
|
||||
|
||||
inspect = celery_app.control.inspect(timeout=2.0)
|
||||
|
||||
def _flatten(per_worker: dict[str, list[dict[str, Any]]] | None) -> list[dict[str, Any]]:
|
||||
if not per_worker:
|
||||
return []
|
||||
out: list[dict[str, Any]] = []
|
||||
for worker, tasks in per_worker.items():
|
||||
for t in tasks or []:
|
||||
out.append(
|
||||
{
|
||||
"worker": worker,
|
||||
"task_id": t.get("id"),
|
||||
"name": t.get("name"),
|
||||
"args": t.get("args"),
|
||||
"kwargs": t.get("kwargs"),
|
||||
"time_start": t.get("time_start"),
|
||||
"eta": t.get("eta"),
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
try:
|
||||
active = _flatten(inspect.active())
|
||||
except Exception:
|
||||
active = []
|
||||
try:
|
||||
reserved = _flatten(inspect.reserved())
|
||||
except Exception:
|
||||
reserved = []
|
||||
try:
|
||||
scheduled = _flatten(inspect.scheduled())
|
||||
except Exception:
|
||||
scheduled = []
|
||||
|
||||
# Pending in broker queue (not yet picked up by any worker).
|
||||
queue_depth: int | None = None
|
||||
try:
|
||||
with celery_app.connection_or_acquire() as conn:
|
||||
with conn.channel() as channel:
|
||||
# Default queue name is 'celery'.
|
||||
queue_depth = channel.client.llen("celery")
|
||||
except Exception:
|
||||
queue_depth = None
|
||||
|
||||
workers: list[str] = []
|
||||
try:
|
||||
ping_resp = inspect.ping() or {}
|
||||
workers = list(ping_resp.keys())
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return {
|
||||
"workers": workers,
|
||||
"queue_depth": queue_depth,
|
||||
"active": active,
|
||||
"reserved": reserved,
|
||||
"scheduled": scheduled,
|
||||
}
|
||||
|
||||
|
||||
class RevokeRequest(BaseModel):
|
||||
task_id: str = Field(..., min_length=1, max_length=255)
|
||||
terminate: bool = False # SIGTERM running task vs just unmark from queue
|
||||
|
||||
|
||||
@router.post("/revoke")
|
||||
def revoke_task(
|
||||
payload: RevokeRequest,
|
||||
x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Cancel a queued or running task by id."""
|
||||
_check_token(x_admin_token)
|
||||
from app.workers.celery_app import celery_app
|
||||
|
||||
celery_app.control.revoke(payload.task_id, terminate=payload.terminate, signal="SIGTERM")
|
||||
return {"task_id": payload.task_id, "revoked": True, "terminate": payload.terminate}
|
||||
|
||||
|
||||
@router.get("/failures")
|
||||
def list_failures(
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
|
|
|
|||
|
|
@ -25,6 +25,24 @@ interface TriggerResp {
|
|||
developers: string[] | null;
|
||||
}
|
||||
|
||||
interface QueueTask {
|
||||
worker: string;
|
||||
task_id: string;
|
||||
name: string | null;
|
||||
args: unknown[] | null;
|
||||
kwargs: Record<string, unknown> | null;
|
||||
time_start: number | null;
|
||||
eta: string | null;
|
||||
}
|
||||
|
||||
interface QueueStatus {
|
||||
workers: string[];
|
||||
queue_depth: number | null;
|
||||
active: QueueTask[];
|
||||
reserved: QueueTask[];
|
||||
scheduled: QueueTask[];
|
||||
}
|
||||
|
||||
interface FailureRow {
|
||||
failure_id: number;
|
||||
run_id: number | null;
|
||||
|
|
@ -89,6 +107,27 @@ export default function ScrapeAdminPage() {
|
|||
},
|
||||
});
|
||||
|
||||
const queue = useQuery({
|
||||
queryKey: ["scrape-queue", token],
|
||||
queryFn: () =>
|
||||
apiFetch<QueueStatus>("/api/v1/admin/scrape/queue", {
|
||||
headers: { "X-Admin-Token": token },
|
||||
}),
|
||||
enabled: !!token,
|
||||
refetchInterval: 3000,
|
||||
});
|
||||
|
||||
const revokeMutation = useMutation({
|
||||
mutationFn: (task_id: string) =>
|
||||
apiFetch<{ revoked: boolean }>("/api/v1/admin/scrape/revoke", {
|
||||
method: "POST",
|
||||
headers: { "X-Admin-Token": token },
|
||||
body: JSON.stringify({ task_id, terminate: false }),
|
||||
}),
|
||||
onSuccess: () =>
|
||||
queryClient.invalidateQueries({ queryKey: ["scrape-queue"] }),
|
||||
});
|
||||
|
||||
const failures = useQuery({
|
||||
queryKey: ["scrape-failures", failuresRunId, token],
|
||||
queryFn: () => {
|
||||
|
|
@ -269,6 +308,219 @@ export default function ScrapeAdminPage() {
|
|||
) : null}
|
||||
</section>
|
||||
|
||||
<section style={{ ...cardStyle, marginTop: 16 }}>
|
||||
<div
|
||||
style={{
|
||||
display: "flex",
|
||||
alignItems: "center",
|
||||
justifyContent: "space-between",
|
||||
marginBottom: 12,
|
||||
}}
|
||||
>
|
||||
<h2 style={{ margin: 0, fontSize: 16 }}>
|
||||
Очередь и активные{" "}
|
||||
{queue.data ? (
|
||||
<span style={{ fontWeight: 400, color: "#5b6066", fontSize: 13 }}>
|
||||
· workers: {queue.data.workers.length || "0"} · в очереди:{" "}
|
||||
{queue.data.queue_depth ?? "?"} · активно:{" "}
|
||||
{queue.data.active.length} · reserved:{" "}
|
||||
{queue.data.reserved.length}
|
||||
</span>
|
||||
) : null}
|
||||
</h2>
|
||||
</div>
|
||||
{!token ? (
|
||||
<p style={{ color: "#5b6066", fontSize: 13 }}>
|
||||
Введите Admin Token чтобы видеть состояние очереди.
|
||||
</p>
|
||||
) : queue.isLoading ? (
|
||||
<p style={{ color: "#5b6066" }}>Загрузка…</p>
|
||||
) : queue.isError ? (
|
||||
<p style={{ color: "#b3261e" }}>Ошибка: {String(queue.error)}</p>
|
||||
) : queue.data &&
|
||||
queue.data.workers.length === 0 &&
|
||||
(queue.data.queue_depth ?? 0) === 0 ? (
|
||||
<p style={{ color: "#5b6066", fontSize: 13 }}>
|
||||
Нет активных worker-ов и очередь пуста. Запусти{" "}
|
||||
<code
|
||||
style={{
|
||||
background: "#f3f4f6",
|
||||
padding: "1px 5px",
|
||||
borderRadius: 3,
|
||||
}}
|
||||
>
|
||||
docker compose up -d worker beat
|
||||
</code>{" "}
|
||||
если ожидаешь scheduled-прогон.
|
||||
</p>
|
||||
) : queue.data ? (
|
||||
<>
|
||||
{queue.data.active.length > 0 ? (
|
||||
<>
|
||||
<h3
|
||||
style={{
|
||||
fontSize: 14,
|
||||
margin: "8px 0 6px",
|
||||
color: "#0a7a3a",
|
||||
}}
|
||||
>
|
||||
▶ Активные ({queue.data.active.length})
|
||||
</h3>
|
||||
<table style={queueTable}>
|
||||
<thead>
|
||||
<tr style={{ background: "#f0fdf4" }}>
|
||||
{["Task", "Worker", "Аргументы", "Старт", "Действие"].map(
|
||||
(h) => (
|
||||
<th key={h} style={th}>
|
||||
{h}
|
||||
</th>
|
||||
),
|
||||
)}
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{queue.data.active.map((t) => (
|
||||
<tr
|
||||
key={t.task_id}
|
||||
style={{ borderBottom: "1px solid #eef0f3" }}
|
||||
>
|
||||
<td style={td}>
|
||||
<div style={{ fontWeight: 600 }}>
|
||||
{t.name?.split(".").pop() ?? "—"}
|
||||
</div>
|
||||
<div
|
||||
style={{
|
||||
fontFamily: "monospace",
|
||||
fontSize: 11,
|
||||
color: "#73767e",
|
||||
}}
|
||||
>
|
||||
{t.task_id?.slice(0, 8)}…
|
||||
</div>
|
||||
</td>
|
||||
<td style={td}>{t.worker}</td>
|
||||
<td
|
||||
style={{
|
||||
...td,
|
||||
fontFamily: "monospace",
|
||||
fontSize: 11,
|
||||
}}
|
||||
>
|
||||
{JSON.stringify(t.args)}{" "}
|
||||
{JSON.stringify(t.kwargs).slice(0, 80)}
|
||||
</td>
|
||||
<td style={td}>
|
||||
{t.time_start
|
||||
? new Date(t.time_start * 1000).toLocaleTimeString(
|
||||
"ru",
|
||||
)
|
||||
: "—"}
|
||||
</td>
|
||||
<td style={td}>
|
||||
<button
|
||||
onClick={() => revokeMutation.mutate(t.task_id)}
|
||||
style={dangerBtn}
|
||||
>
|
||||
Отменить
|
||||
</button>
|
||||
</td>
|
||||
</tr>
|
||||
))}
|
||||
</tbody>
|
||||
</table>
|
||||
</>
|
||||
) : null}
|
||||
{queue.data.reserved.length > 0 ? (
|
||||
<>
|
||||
<h3
|
||||
style={{
|
||||
fontSize: 14,
|
||||
margin: "12px 0 6px",
|
||||
color: "#9a6700",
|
||||
}}
|
||||
>
|
||||
⏸ Reserved (приняты worker-ом, не стартовали) (
|
||||
{queue.data.reserved.length})
|
||||
</h3>
|
||||
<table style={queueTable}>
|
||||
<thead>
|
||||
<tr style={{ background: "#fffbeb" }}>
|
||||
{["Task", "Worker", "Аргументы", "Действие"].map((h) => (
|
||||
<th key={h} style={th}>
|
||||
{h}
|
||||
</th>
|
||||
))}
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{queue.data.reserved.map((t) => (
|
||||
<tr
|
||||
key={t.task_id}
|
||||
style={{ borderBottom: "1px solid #eef0f3" }}
|
||||
>
|
||||
<td style={td}>
|
||||
<div style={{ fontWeight: 600 }}>
|
||||
{t.name?.split(".").pop() ?? "—"}
|
||||
</div>
|
||||
<div
|
||||
style={{
|
||||
fontFamily: "monospace",
|
||||
fontSize: 11,
|
||||
color: "#73767e",
|
||||
}}
|
||||
>
|
||||
{t.task_id?.slice(0, 8)}…
|
||||
</div>
|
||||
</td>
|
||||
<td style={td}>{t.worker}</td>
|
||||
<td
|
||||
style={{
|
||||
...td,
|
||||
fontFamily: "monospace",
|
||||
fontSize: 11,
|
||||
}}
|
||||
>
|
||||
{JSON.stringify(t.args)}{" "}
|
||||
{JSON.stringify(t.kwargs).slice(0, 80)}
|
||||
</td>
|
||||
<td style={td}>
|
||||
<button
|
||||
onClick={() => revokeMutation.mutate(t.task_id)}
|
||||
style={dangerBtn}
|
||||
>
|
||||
Отменить
|
||||
</button>
|
||||
</td>
|
||||
</tr>
|
||||
))}
|
||||
</tbody>
|
||||
</table>
|
||||
</>
|
||||
) : null}
|
||||
{queue.data.scheduled.length > 0 ? (
|
||||
<>
|
||||
<h3
|
||||
style={{
|
||||
fontSize: 14,
|
||||
margin: "12px 0 6px",
|
||||
color: "#5b6066",
|
||||
}}
|
||||
>
|
||||
📅 Scheduled (с ETA) ({queue.data.scheduled.length})
|
||||
</h3>
|
||||
<ul style={{ marginTop: 0, fontSize: 13 }}>
|
||||
{queue.data.scheduled.map((t) => (
|
||||
<li key={t.task_id}>
|
||||
{t.name} → ETA {t.eta} — {JSON.stringify(t.args)}
|
||||
</li>
|
||||
))}
|
||||
</ul>
|
||||
</>
|
||||
) : null}
|
||||
</>
|
||||
) : null}
|
||||
</section>
|
||||
|
||||
<section style={{ ...cardStyle, marginTop: 16 }}>
|
||||
<div
|
||||
style={{
|
||||
|
|
@ -515,6 +767,21 @@ const secondaryBtn = {
|
|||
fontSize: 13,
|
||||
cursor: "pointer",
|
||||
};
|
||||
const dangerBtn = {
|
||||
padding: "4px 10px",
|
||||
background: "#fef2f2",
|
||||
color: "#991b1b",
|
||||
border: "1px solid #fecaca",
|
||||
borderRadius: 4,
|
||||
fontSize: 12,
|
||||
cursor: "pointer",
|
||||
};
|
||||
const queueTable = {
|
||||
width: "100%",
|
||||
borderCollapse: "collapse" as const,
|
||||
fontSize: 13,
|
||||
marginBottom: 4,
|
||||
};
|
||||
const th = {
|
||||
padding: "8px 10px",
|
||||
textAlign: "left" as const,
|
||||
|
|
|
|||
File diff suppressed because one or more lines are too long
Loading…
Add table
Reference in a new issue