@@ -396,85 +472,111 @@ export default function AdminTaskMonitorPage() {
);
}
- const overview = overviewQuery.data;
+ const workersOverview = workersOverviewQuery.data;
return (
-
- 任务列表上限
- setTaskLimit(normalizePositiveInt(value, DEFAULT_TASK_LIMIT, 1, 500))}
- />
-
-
- 历史任务扫描上限
- setHistoryLimit(normalizePositiveInt(value, DEFAULT_HISTORY_LIMIT, 0, 500))}
- />
-
+ setWorkerKeyword(event.target.value)}
+ style={{ width: 220 }}
+ />
+ setQueueKeyword(event.target.value)}
+ style={{ width: 220 }}
+ />
+ setTaskKeyword(event.target.value)}
+ style={{ width: 240 }}
+ />
+
- {overviewQuery.error && (
+ {workersOverviewQuery.error && (
+ )}
+ {allTasksQuery.error && (
+
)}
- {!overview && !overviewQuery.isFetching && (
+ {!workersOverview && !workersOverviewQuery.isFetching && (
)}
- {overview && (
+ {workersOverview && (
<>
-
+
-
+
-
+
-
+
- {overview.task_state_buckets.length > 0 ? (
- overview.task_state_buckets.map((item) => (
- {`${item.label}: ${item.count}`}
+ {stateBuckets.length > 0 ? (
+ stateBuckets.map((item) => (
+ {`${item.state}: ${item.count}`}
))
) : (
暂无状态分布数据
@@ -482,36 +584,25 @@ export default function AdminTaskMonitorPage() {
- Broker: {overview.broker_url || "-"}}>
-
+
+
rowKey={(record) => record.worker}
columns={workerColumns}
- dataSource={overview.workers}
+ dataSource={filteredWorkers}
pagination={false}
locale={{ emptyText: "暂无 Worker 数据" }}
- scroll={{ x: 1200 }}
- />
-
-
- Result Backend: {overview.result_backend || "-"}}>
-
- rowKey={(record) => record.name}
- columns={queueColumns}
- dataSource={overview.queues}
- pagination={false}
- locale={{ emptyText: "暂无 Queue 数据" }}
- scroll={{ x: 760 }}
+ scroll={{ x: 1500 }}
/>
-
- rowKey={(record) => record.task_id}
+
+ rowKey={(record) => record.key}
columns={taskColumns}
- dataSource={overview.tasks}
- pagination={false}
+ dataSource={filteredTaskRows}
+ pagination={{ pageSize: 50, showSizeChanger: true }}
locale={{ emptyText: "暂无任务数据" }}
- scroll={{ x: 2200 }}
+ scroll={{ x: 2600 }}
/>
diff --git a/web/src/app/admin/workers/page.tsx b/web/src/app/admin/workers/page.tsx
new file mode 100644
index 0000000..e46d97c
--- /dev/null
+++ b/web/src/app/admin/workers/page.tsx
@@ -0,0 +1,627 @@
+"use client";
+
+import Link from "next/link";
+import dayjs from "dayjs";
+import { useMemo, useState } from "react";
+import type { ComponentType } from "react";
+import { useQuery } from "@tanstack/react-query";
+import {
+ Alert,
+ Button,
+ Card,
+ Col,
+ Drawer,
+ Empty,
+ Input,
+ Row,
+ Select,
+ Space,
+ Spin,
+ Statistic,
+ Switch,
+ Table,
+ Tag,
+ Typography,
+ type CardProps,
+ type TableColumnsType,
+} from "antd";
+
+import { useAuth } from "@/components/auth-provider";
+import { readApiError } from "@/lib/api";
+
+const { Text } = Typography;
+const AntCard = Card as unknown as ComponentType;
+
+const DEFAULT_RECENT_LIMIT = 100;
+
+type WorkerMonitorWorkerItem = {
+ worker: string;
+ status: string;
+ queue_names: string[];
+ concurrency: number;
+ prefetch_count: number;
+ processed_count: number;
+ active_count: number;
+ reserved_count: number;
+ scheduled_count: number;
+ registered_count: number;
+ last_heartbeat_at: string | null;
+};
+
+type WorkerMonitorOverviewResponse = {
+ generated_at: string;
+ summary: {
+ total: number;
+ online: number;
+ offline: number;
+ };
+ workers: WorkerMonitorWorkerItem[];
+};
+
+type WorkerMonitorTaskItem = {
+ task_id: string;
+ name: string;
+ state: string;
+ source: string;
+ queue_name: string | null;
+ worker: string;
+ args_text: string | null;
+ kwargs_text: string | null;
+ eta: string | null;
+ received_at: string | null;
+ started_at: string | null;
+ finished_at: string | null;
+ runtime_seconds: number | null;
+ result_text: string | null;
+ exception_text: string | null;
+};
+
+type WorkerMonitorTaskOverviewResponse = {
+ generated_at: string;
+ worker: string;
+ summary: {
+ active: number;
+ reserved: number;
+ scheduled: number;
+ recent: number;
+ };
+ active_tasks: WorkerMonitorTaskItem[];
+ reserved_tasks: WorkerMonitorTaskItem[];
+ scheduled_tasks: WorkerMonitorTaskItem[];
+ recent_tasks: WorkerMonitorTaskItem[];
+};
+
+function formatDateTime(value: string | null | undefined): string {
+ if (!value) {
+ return "-";
+ }
+ const parsed = dayjs(value);
+ if (!parsed.isValid()) {
+ return "-";
+ }
+ return parsed.format("YYYY-MM-DD HH:mm:ss");
+}
+
+function renderTaskStateTag(state: string) {
+ const normalized = (state || "").toUpperCase();
+ const color =
+ normalized === "STARTED"
+ ? "processing"
+ : normalized === "RECEIVED"
+ ? "blue"
+ : normalized === "SCHEDULED"
+ ? "purple"
+ : normalized === "RETRY"
+ ? "orange"
+ : normalized === "SUCCESS"
+ ? "green"
+ : normalized === "FAILURE"
+ ? "red"
+ : normalized === "REVOKED"
+ ? "default"
+ : "geekblue";
+ return {normalized || "UNKNOWN"};
+}
+
+function renderWorkerStatusTag(status: string) {
+ return (status || "").toUpperCase() === "ONLINE" ? 在线 : 离线;
+}
+
+function containsText(source: string | null | undefined, keyword: string): boolean {
+ if (!keyword) {
+ return true;
+ }
+ return (source || "").toLowerCase().includes(keyword.toLowerCase());
+}
+
+function parseStatusFilter(value: string | undefined): "all" | "online" | "offline" {
+ if (value === "online" || value === "offline") {
+ return value;
+ }
+ return "all";
+}
+
+export default function AdminWorkersPage() {
+ const { user, initializing, fetchWithAuth, hasPermission } = useAuth();
+ const canRead = hasPermission("celery.read") || hasPermission("celery.manage");
+
+ const [autoRefresh, setAutoRefresh] = useState(true);
+ const [workerKeyword, setWorkerKeyword] = useState("");
+ const [queueKeyword, setQueueKeyword] = useState("");
+ const [statusFilter, setStatusFilter] = useState<"all" | "online" | "offline">("all");
+ const [selectedWorker, setSelectedWorker] = useState(null);
+
+ const overviewQuery = useQuery({
+ queryKey: ["worker-monitor-overview"],
+ enabled: Boolean(user) && canRead,
+ queryFn: async () => {
+ const response = await fetchWithAuth("/api/v1/admin/flower/workers?forceRefresh=false");
+ if (!response.ok) {
+ throw new Error(await readApiError(response));
+ }
+ return (await response.json()) as WorkerMonitorOverviewResponse;
+ },
+ refetchInterval: autoRefresh ? 5_000 : false,
+ staleTime: 15_000,
+ });
+
+ const workerTasksPath = useMemo(() => {
+ if (!selectedWorker) {
+ return null;
+ }
+ const params = new URLSearchParams();
+ params.set("worker", selectedWorker);
+ params.set("recentLimit", String(DEFAULT_RECENT_LIMIT));
+ params.set("forceRefresh", "false");
+ return `/api/v1/admin/flower/worker-tasks?${params.toString()}`;
+ }, [selectedWorker]);
+
+ const workerTasksQuery = useQuery({
+ queryKey: ["worker-monitor-tasks", workerTasksPath],
+ enabled: Boolean(user) && canRead && Boolean(workerTasksPath),
+ queryFn: async () => {
+ if (!workerTasksPath) {
+ throw new Error("missing worker path");
+ }
+ const response = await fetchWithAuth(workerTasksPath);
+ if (!response.ok) {
+ throw new Error(await readApiError(response));
+ }
+ return (await response.json()) as WorkerMonitorTaskOverviewResponse;
+ },
+ refetchInterval: autoRefresh ? 5_000 : false,
+ staleTime: 15_000,
+ });
+
+ const workerColumns = useMemo>(
+ () => [
+ {
+ title: "Worker",
+ dataIndex: "worker",
+ key: "worker",
+ width: 280,
+ },
+ {
+ title: "状态",
+ dataIndex: "status",
+ key: "status",
+ width: 90,
+ render: (value: string) => renderWorkerStatusTag(value),
+ },
+ {
+ title: "队列",
+ dataIndex: "queue_names",
+ key: "queue_names",
+ render: (value: string[]) => (value.length > 0 ? value.join(", ") : "-"),
+ },
+ {
+ title: "并发",
+ dataIndex: "concurrency",
+ key: "concurrency",
+ width: 80,
+ },
+ {
+ title: "Prefetch",
+ dataIndex: "prefetch_count",
+ key: "prefetch_count",
+ width: 90,
+ },
+ {
+ title: "已注册任务",
+ dataIndex: "registered_count",
+ key: "registered_count",
+ width: 110,
+ },
+ {
+ title: "累计处理",
+ dataIndex: "processed_count",
+ key: "processed_count",
+ width: 110,
+ },
+ {
+ title: "最近心跳",
+ dataIndex: "last_heartbeat_at",
+ key: "last_heartbeat_at",
+ width: 170,
+ render: (value: string | null) => formatDateTime(value),
+ },
+ {
+ title: "Active/Reserved/Scheduled",
+ key: "runtime_counts",
+ width: 190,
+ render: (_: unknown, record) => `${record.active_count}/${record.reserved_count}/${record.scheduled_count}`,
+ },
+ {
+ title: "操作",
+ key: "action",
+ width: 120,
+ fixed: "right",
+ render: (_: unknown, record) => (
+
+ ),
+ },
+ ],
+ [],
+ );
+
+ const taskColumns = useMemo>(
+ () => [
+ {
+ title: "Task ID",
+ dataIndex: "task_id",
+ key: "task_id",
+ width: 260,
+ render: (value: string) => {value},
+ },
+ {
+ title: "任务名",
+ dataIndex: "name",
+ key: "name",
+ width: 220,
+ render: (value: string) => value || "-",
+ },
+ {
+ title: "状态",
+ dataIndex: "state",
+ key: "state",
+ width: 110,
+ render: (value: string) => renderTaskStateTag(value),
+ },
+ {
+ title: "来源",
+ dataIndex: "source",
+ key: "source",
+ width: 90,
+ render: (value: string) => value || "-",
+ },
+ {
+ title: "队列",
+ dataIndex: "queue_name",
+ key: "queue_name",
+ width: 120,
+ render: (value: string | null) => value || "-",
+ },
+ {
+ title: "Worker",
+ dataIndex: "worker",
+ key: "worker",
+ width: 220,
+ render: (value: string) => value || "-",
+ },
+ {
+ title: "ETA",
+ dataIndex: "eta",
+ key: "eta",
+ width: 170,
+ render: (value: string | null) => formatDateTime(value),
+ },
+ {
+ title: "接收",
+ dataIndex: "received_at",
+ key: "received_at",
+ width: 170,
+ render: (value: string | null) => formatDateTime(value),
+ },
+ {
+ title: "开始",
+ dataIndex: "started_at",
+ key: "started_at",
+ width: 170,
+ render: (value: string | null) => formatDateTime(value),
+ },
+ {
+ title: "完成",
+ dataIndex: "finished_at",
+ key: "finished_at",
+ width: 170,
+ render: (value: string | null) => formatDateTime(value),
+ },
+ {
+ title: "运行时长",
+ dataIndex: "runtime_seconds",
+ key: "runtime_seconds",
+ width: 110,
+ render: (value: number | null) => (value === null ? "-" : `${value.toFixed(3)}s`),
+ },
+ {
+ title: "Args",
+ dataIndex: "args_text",
+ key: "args_text",
+ width: 220,
+ render: (value: string | null) => (value ? {value} : "-"),
+ },
+ {
+ title: "Kwargs",
+ dataIndex: "kwargs_text",
+ key: "kwargs_text",
+ width: 220,
+ render: (value: string | null) => (value ? {value} : "-"),
+ },
+ {
+ title: "Result",
+ dataIndex: "result_text",
+ key: "result_text",
+ width: 220,
+ render: (value: string | null) =>
+ value ? (
+
+ {value}
+
+ ) : (
+ "-"
+ ),
+ },
+ {
+ title: "Exception",
+ dataIndex: "exception_text",
+ key: "exception_text",
+ width: 220,
+ render: (value: string | null) =>
+ value ? (
+
+ {value}
+
+ ) : (
+ "-"
+ ),
+ },
+ ],
+ [],
+ );
+
+ const filteredWorkers = useMemo(() => {
+ const rows = overviewQuery.data?.workers || [];
+ return rows.filter((item) => {
+ const online = (item.status || "").toUpperCase() === "ONLINE";
+ if (statusFilter === "online" && !online) {
+ return false;
+ }
+ if (statusFilter === "offline" && online) {
+ return false;
+ }
+ if (!containsText(item.worker, workerKeyword.trim())) {
+ return false;
+ }
+ if (queueKeyword.trim()) {
+ const text = item.queue_names.join(", ");
+ if (!containsText(text, queueKeyword.trim())) {
+ return false;
+ }
+ }
+ return true;
+ });
+ }, [overviewQuery.data?.workers, queueKeyword, statusFilter, workerKeyword]);
+
+ if (initializing || (overviewQuery.isLoading && !overviewQuery.data && canRead && Boolean(user))) {
+ return (
+
+
+
+ );
+ }
+
+ if (!user) {
+ return (
+
+ 请先登录后再访问Worker监控页面。
+
+ 返回首页
+
+
+ );
+ }
+
+ if (!canRead) {
+ return (
+
+ 你没有访问该页面的权限（需要 `celery.read` 或 `celery.manage`）。
+
+ 返回首页
+
+
+ );
+ }
+
+ const overview = overviewQuery.data;
+ const taskOverview = workerTasksQuery.data;
+
+ return (
+
+
+
+ setWorkerKeyword(event.target.value)}
+ style={{ width: 220 }}
+ />
+ setQueueKeyword(event.target.value)}
+ style={{ width: 220 }}
+ />
+
+
+
+ {overviewQuery.error && (
+
+ )}
+
+ {!overview && !overviewQuery.isFetching && (
+
+
+
+ )}
+
+ {overview && (
+ <>
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ rowKey={(record) => record.worker}
+ columns={workerColumns}
+ dataSource={filteredWorkers}
+ pagination={{ pageSize: 10, showSizeChanger: true }}
+ locale={{ emptyText: "暂无Worker数据" }}
+ scroll={{ x: 1600 }}
+ />
+
+
+ )}
+
+ setSelectedWorker(null)}
+ extra={
+
+
+ {taskOverview ? `采集时间：${formatDateTime(taskOverview.generated_at)}` : "-"}
+
+ }
+ >
+ {workerTasksQuery.isLoading && !taskOverview ? : null}
+ {workerTasksQuery.error ? (
+
+ ) : null}
+ {!workerTasksQuery.isLoading && !taskOverview ? : null}
+ {taskOverview ? (
+
+
+ 运行中: {taskOverview.summary.active}
+ 保留中: {taskOverview.summary.reserved}
+ 已调度: {taskOverview.summary.scheduled}
+ 最近任务: {taskOverview.summary.recent}
+
+
+
+
+ rowKey={(record) => `active-${record.task_id}`}
+ columns={taskColumns}
+ dataSource={taskOverview.active_tasks}
+ pagination={false}
+ locale={{ emptyText: "暂无运行中任务" }}
+ scroll={{ x: 2200 }}
+ />
+
+
+
+
+ rowKey={(record) => `reserved-${record.task_id}`}
+ columns={taskColumns}
+ dataSource={taskOverview.reserved_tasks}
+ pagination={false}
+ locale={{ emptyText: "暂无保留任务" }}
+ scroll={{ x: 2200 }}
+ />
+
+
+
+
+ rowKey={(record) => `scheduled-${record.task_id}`}
+ columns={taskColumns}
+ dataSource={taskOverview.scheduled_tasks}
+ pagination={false}
+ locale={{ emptyText: "暂无定时任务" }}
+ scroll={{ x: 2200 }}
+ />
+
+
+
+
+ rowKey={(record) => `recent-${record.task_id}`}
+ columns={taskColumns}
+ dataSource={taskOverview.recent_tasks}
+ pagination={false}
+ locale={{ emptyText: "暂无最近任务" }}
+ scroll={{ x: 2200 }}
+ />
+
+
+ ) : null}
+
+
+ );
+}