реворк задач

This commit is contained in:
dv 2026-05-28 17:29:51 +03:00
parent 735d8b2736
commit de0a28ad62
3 changed files with 88 additions and 6 deletions

View file

@ -28,6 +28,23 @@ from app.services.sse import SSEEvent, sse_manager
_TASK_DEVICE_CONCURRENCY = settings.TASK_DEVICE_CONCURRENCY _TASK_DEVICE_CONCURRENCY = settings.TASK_DEVICE_CONCURRENCY
async def _quick_ping(ip: str, attempts: int = 4) -> bool:
"""Quick reachability check: ping -c 1 -W 1, up to `attempts` times."""
for _ in range(attempts):
try:
proc = await asyncio.create_subprocess_exec(
"ping", "-c", "1", "-W", "1", ip,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await asyncio.wait_for(proc.wait(), timeout=3)
if proc.returncode == 0:
return True
except (asyncio.TimeoutError, OSError):
continue
return False
async def _emit(task_id: str, message: str, level: str = "info") -> None: async def _emit(task_id: str, message: str, level: str = "info") -> None:
"""Write a progress line to DB and push to SSE subscribers.""" """Write a progress line to DB and push to SSE subscribers."""
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
@ -126,6 +143,41 @@ async def _run_on_device(
task_row = await db.get(Task, task_id) task_row = await db.get(Task, task_id)
if task_row and task_row.cancel_requested: if task_row and task_row.cancel_requested:
return None # None = skipped due to cancellation return None # None = skipped due to cancellation
# Pre-ping reachability check (skip for ping action itself)
if action.name != "ping":
reachable = await _quick_ping(device.ip)
if not reachable:
await _emit(task_id, f"{device.ip}: unreachable, skipping", "warn")
async with AsyncSessionLocal() as db:
async with db.begin():
tr = TaskResult(
task_id=task_id,
device_id=device.id,
status="skipped",
executed_at=datetime.now(timezone.utc),
duration_ms=0,
stderr="Device unreachable (ping pre-check failed)",
)
db.add(tr)
await sse_manager.publish(SSEEvent(
type="task_result",
task_id=task_id,
data={
"device_id": device.id,
"device_ip": device.ip,
"device_hostname": device.hostname,
"status": "skipped",
"duration_ms": 0,
"stdout": None,
"stderr": "Device unreachable (ping pre-check failed)",
"error": None,
},
))
return None
async with AsyncSessionLocal() as db:
async with db.begin():
tr = TaskResult( tr = TaskResult(
task_id=task_id, task_id=task_id,
device_id=device.id, device_id=device.id,
@ -284,6 +336,11 @@ async def _run_task(task_id: str) -> None:
return return
await _emit(task_id, f"Running on {len(targets)} device(s)...", "info") await _emit(task_id, f"Running on {len(targets)} device(s)...", "info")
await sse_manager.publish(SSEEvent(
type="task_meta",
task_id=task_id,
data={"total": len(targets)},
))
sem = asyncio.Semaphore(_TASK_DEVICE_CONCURRENCY) sem = asyncio.Semaphore(_TASK_DEVICE_CONCURRENCY)
# Start heartbeat writer # Start heartbeat writer

View file

@ -15,6 +15,7 @@ interface SSEHandlers {
error: string | null error: string | null
}) => void }) => void
onTaskStatus?: (data: { status: string; total: number; ok: number; failed: number }) => void onTaskStatus?: (data: { status: string; total: number; ok: number; failed: number }) => void
onTaskMeta?: (data: { total: number }) => void
} }
export function useSSE(taskId: string | null | undefined, handlers: SSEHandlers) { export function useSSE(taskId: string | null | undefined, handlers: SSEHandlers) {
@ -35,6 +36,7 @@ export function useSSE(taskId: string | null | undefined, handlers: SSEHandlers)
if (ev.event === 'task_event') handlersRef.current.onTaskEvent?.(data) if (ev.event === 'task_event') handlersRef.current.onTaskEvent?.(data)
else if (ev.event === 'task_result') handlersRef.current.onTaskResult?.(data) else if (ev.event === 'task_result') handlersRef.current.onTaskResult?.(data)
else if (ev.event === 'task_status') handlersRef.current.onTaskStatus?.(data) else if (ev.event === 'task_status') handlersRef.current.onTaskStatus?.(data)
else if (ev.event === 'task_meta') handlersRef.current.onTaskMeta?.(data)
} catch { } catch {
// ignore parse errors // ignore parse errors
} }

View file

@ -16,6 +16,7 @@ import {
Descriptions, Descriptions,
Modal, Modal,
Popconfirm, Popconfirm,
Progress,
Row, Row,
Space, Space,
Spin, Spin,
@ -83,7 +84,7 @@ export function TaskDetailPage() {
const [log, setLog] = useState<LogLine[]>([]) const [log, setLog] = useState<LogLine[]>([])
const [liveResults, setLiveResults] = useState<Record<number, DeviceResultState>>({}) const [liveResults, setLiveResults] = useState<Record<number, DeviceResultState>>({})
const logEndRef = useRef<HTMLDivElement>(null) const [totalDevices, setTotalDevices] = useState<number | null>(null)
const [stdoutModal, setStdoutModal] = useState<string | null>(null) const [stdoutModal, setStdoutModal] = useState<string | null>(null)
const seenLogKeys = useRef(new Set<string>()) const seenLogKeys = useRef(new Set<string>())
@ -118,12 +119,9 @@ export function TaskDetailPage() {
device_hostname: data.device_hostname ?? null, device_hostname: data.device_hostname ?? null,
}, },
})), })),
onTaskMeta: (data) => setTotalDevices(data.total),
}) })
useEffect(() => {
logEndRef.current?.scrollIntoView({ behavior: 'smooth' })
}, [log])
// Merge DB results with live SSE updates // Merge DB results with live SSE updates
const mergedResults = useMemo<(TaskResult & Partial<DeviceResultState>)[]>(() => { const mergedResults = useMemo<(TaskResult & Partial<DeviceResultState>)[]>(() => {
if (!task?.device_results) return [] if (!task?.device_results) return []
@ -210,6 +208,7 @@ export function TaskDetailPage() {
const s = r.status const s = r.status
if (s === 'success') return <Tag icon={<CheckCircleOutlined />} color="success">Успешно</Tag> if (s === 'success') return <Tag icon={<CheckCircleOutlined />} color="success">Успешно</Tag>
if (s === 'failed') return <Tag icon={<CloseCircleOutlined />} color="error">Ошибка</Tag> if (s === 'failed') return <Tag icon={<CloseCircleOutlined />} color="error">Ошибка</Tag>
if (s === 'skipped') return <Tag icon={<StopOutlined />} color="warning">Пропущено</Tag>
return <Tag icon={<LoadingOutlined />} color="processing">Ожидание</Tag> return <Tag icon={<LoadingOutlined />} color="processing">Ожидание</Tag>
}, },
}, },
@ -348,6 +347,30 @@ export function TaskDetailPage() {
</Col> </Col>
</Row> </Row>
{totalDevices != null && totalDevices > 0 && (
<Row style={{ marginBottom: 16 }}>
<Col span={24}>
{(() => {
const completed = mergedResults.filter((r) => r.status !== 'running').length
const failed = mergedResults.filter((r) => r.status === 'failed').length
const percent = Math.round((completed / totalDevices) * 100)
const isDone = !isActive
const progressStatus = isDone
? (failed > 0 ? 'exception' : 'success')
: 'active'
return (
<Progress
percent={percent}
status={progressStatus}
format={() => `${completed} / ${totalDevices}`}
strokeColor={failed > 0 ? undefined : '#52c41a'}
/>
)
})()}
</Col>
</Row>
)}
<Row gutter={16}> <Row gutter={16}>
<Col span={24}> <Col span={24}>
<Card <Card
@ -378,7 +401,7 @@ export function TaskDetailPage() {
</div> </div>
)) ))
)} )}
<div ref={logEndRef} />
</div> </div>
</Card> </Card>
</Col> </Col>