From c8429814bfb8b86b84083d5ddda6d98d2796a6e9 Mon Sep 17 00:00:00 2001 From: dv Date: Wed, 8 Apr 2026 15:58:47 +0300 Subject: [PATCH] sse ssh discovery --- backend/app/routers/objects.py | 48 ++++- backend/app/schemas/object.py | 4 + backend/app/services/device_discovery.py | 22 +- frontend/src/api/objects.ts | 15 +- frontend/src/components/DiscoveryDrawer.tsx | 210 +++++++++++++++++++ frontend/src/pages/InventoryObjectDetail.tsx | 27 ++- 6 files changed, 300 insertions(+), 26 deletions(-) create mode 100644 frontend/src/components/DiscoveryDrawer.tsx diff --git a/backend/app/routers/objects.py b/backend/app/routers/objects.py index 2c85cc6..5a9d1ff 100644 --- a/backend/app/routers/objects.py +++ b/backend/app/routers/objects.py @@ -1,3 +1,5 @@ +import asyncio + from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select, distinct from sqlalchemy.ext.asyncio import AsyncSession @@ -9,11 +11,13 @@ from app.models.object import Object from app.models.tag import Tag from app.models.user import User from app.schemas.admin import SheetsSyncRequest -from app.schemas.object import ObjectCreate, ObjectMetaResponse, ObjectRead, ObjectUpdate, SyncResult +from app.schemas.object import ObjectCreate, ObjectMetaResponse, ObjectRead, ObjectUpdate, SyncResult, DiscoveryStarted from app.services import crypto +from app.services.database import AsyncSessionLocal from app.services.device_discovery import discover_via_ssh from app.services.object_db import sync_from_object_db from app.services.sheets import sync_from_sheets +from app.services.sse import sse_manager, SSEEvent router = APIRouter(prefix="/api/v1/objects", tags=["objects"]) @@ -171,22 +175,44 @@ async def sync_devices( ) -@router.post("/{obj_id}/discover", response_model=SyncResult) +async def _run_discovery_background(obj_id: int, task_id: str) -> None: + """Run SSH discovery in background, emitting SSE progress events.""" + + async def emit(event_type: str, data: dict) -> None: + await sse_manager.publish(SSEEvent(type=event_type, task_id=task_id, data=data)) + + try: + async with AsyncSessionLocal() as session: + async with session.begin(): + result_row = await session.execute( + select(Object).options(selectinload(Object.tags)).where(Object.id == obj_id) + ) + obj = result_row.scalar_one_or_none() + if obj is None: + await emit("disc_done", {"created": 0, "updated": 0, "skipped": 0, "errors": ["Object not found"]}) + return + discovery_result = await discover_via_ssh(db=session, obj=obj, progress_cb=emit) + await emit("disc_done", { + "created": discovery_result.created, + "updated": discovery_result.updated, + "skipped": discovery_result.skipped, + "errors": discovery_result.errors, + }) + except Exception as exc: + await emit("disc_done", {"created": 0, "updated": 0, "skipped": 0, "errors": [str(exc)]}) + + +@router.post("/{obj_id}/discover", response_model=DiscoveryStarted) async def discover_devices( obj_id: int, db: AsyncSession = Depends(get_db), _: User = Depends(get_current_user), ): """SSH into each rack PC, parse caps config, discover network devices.""" - obj = await _load_object(db, obj_id) - result = await discover_via_ssh(db, obj) - await db.flush() - return SyncResult( - created=result.created, - updated=result.updated, - skipped=result.skipped, - errors=result.errors, - ) + await _load_object(db, obj_id) + task_id = f"disc-{obj_id}" + asyncio.create_task(_run_discovery_background(obj_id, task_id)) + return DiscoveryStarted(task_id=task_id) @router.post("/{obj_id}/sync/sheets", response_model=SyncResult) diff --git a/backend/app/schemas/object.py b/backend/app/schemas/object.py index fd1042c..4af65b7 100644 --- a/backend/app/schemas/object.py +++ b/backend/app/schemas/object.py @@ -90,3 +90,7 @@ class SyncResult(BaseModel): updated: int skipped: int errors: list[str] = [] + + +class DiscoveryStarted(BaseModel): + task_id: str diff --git a/backend/app/services/device_discovery.py b/backend/app/services/device_discovery.py index 8a027e6..e2a3a37 100644 --- a/backend/app/services/device_discovery.py +++ b/backend/app/services/device_discovery.py @@ -13,6 +13,7 @@ Limitations (devices not discoverable automatically): import logging import re +from collections.abc import Callable, Awaitable from datetime import datetime, timezone from configparser import RawConfigParser from dataclasses import dataclass, field @@ -477,13 +478,21 @@ class DiscoveryResult: actions: list[DeviceAction] = field(default_factory=list) +ProgressCallback = Callable[[str, dict], Awaitable[None]] + + async def discover_via_ssh( db: AsyncSession, obj: Object, config_path: str = "/etc/caps/config.ini", + progress_cb: ProgressCallback | None = None, ) -> DiscoveryResult: result = DiscoveryResult() + async def _emit(event_type: str, data: dict) -> None: + if progress_cb is not None: + await progress_cb(event_type, data) + if not obj.ssh_user or not obj.ssh_pass_enc: result.errors.append("SSH credentials not configured on this object (ssh_user / ssh_password)") return result @@ -585,6 +594,8 @@ async def discover_via_ssh( result.errors.append("No racks with host IPs found in object DB") return result + await _emit("disc_server", {"rack_count": len(rack_hosts)}) + # Step 2: SSH into each rack, parse config, upsert devices # neighbor_candidates: IPs seen in ARP tables of rack PCs; used for slave discovery # discovered_device_ips: all IPs added as devices during this run @@ -684,7 +695,14 @@ async def discover_via_ssh( if infra_devices: await db.flush() - for external_id, rack_host, rack_name in rack_hosts: + for rack_idx, (external_id, rack_host, rack_name) in enumerate(rack_hosts): + await _emit("disc_rack_start", { + "rack_name": rack_name, + "rack_host": rack_host, + "index": rack_idx, + "total": len(rack_hosts), + }) + # Resolve config_path override from the embedded device's connection_params existing_rack_pc = (await db.execute( select(Device).where( @@ -792,6 +810,7 @@ async def discover_via_ssh( await db.flush() except Exception as exc: result.errors.append(f"Flush error for offline rack {rack_host}: {exc}") + await _emit("disc_rack_done", {"rack_name": rack_name, "rack_host": rack_host, "found": 0, "ssh_ok": False}) continue # no config to parse # Collect ARP neighbours for slave rack discovery (best-effort) @@ -929,6 +948,7 @@ async def discover_via_ssh( await db.flush() except Exception as exc: result.errors.append(f"Flush error after rack {rack_host}: {exc}") + await _emit("disc_rack_done", {"rack_name": rack_name, "rack_host": rack_host, "found": len(devices_found), "ssh_ok": True}) # Step 2.5: Discover slave racks via ARP neighbor tables # Filter out already-known IPs: rack hosts, discovered devices, infra IPs, common gateways diff --git a/frontend/src/api/objects.ts b/frontend/src/api/objects.ts index 0f8dcd6..bc96f67 100644 --- a/frontend/src/api/objects.ts +++ b/frontend/src/api/objects.ts @@ -220,15 +220,16 @@ export const useSyncObject = (objId: number) => { }) } -export const useDiscoverObject = (objId: number) => { - const qc = useQueryClient() - return useMutation({ - mutationFn: () => - api.post(`/api/v1/objects/${objId}/discover`).then((r) => r.data), - onSuccess: () => qc.invalidateQueries({ queryKey: ['devices', objId] }), - }) +export interface DiscoveryStarted { + task_id: string } +export const useDiscoverObject = (objId: number) => + useMutation({ + mutationFn: () => + api.post(`/api/v1/objects/${objId}/discover`).then((r) => r.data), + }) + export interface SheetsSyncPayload { spreadsheet_id: string sheet_name?: string diff --git a/frontend/src/components/DiscoveryDrawer.tsx b/frontend/src/components/DiscoveryDrawer.tsx new file mode 100644 index 0000000..5e2d722 --- /dev/null +++ b/frontend/src/components/DiscoveryDrawer.tsx @@ -0,0 +1,210 @@ +import { + CheckCircleOutlined, + CloseCircleOutlined, + LoadingOutlined, + MinusCircleOutlined, +} from '@ant-design/icons' +import { Drawer, Progress, Space, Tag, Typography } from 'antd' +import { fetchEventSource } from '@microsoft/fetch-event-source' +import { useEffect, useRef, useState } from 'react' +import { useAuthStore } from '../store/auth' + +interface RackStatus { + rack_name: string + rack_host: string + state: 'pending' | 'running' | 'done' | 'error' + found: number + ssh_ok: boolean +} + +interface DiscoveryResult { + created: number + updated: number + skipped: number + errors: string[] +} + +interface Props { + open: boolean + onClose: () => void + objectName: string + taskId: string | null + onComplete?: () => void +} + +export function DiscoveryDrawer({ open, onClose, objectName, taskId, onComplete }: Props) { + const accessToken = useAuthStore((s) => s.accessToken) + const [phase, setPhase] = useState<'connecting' | 'running' | 'done'>('connecting') + const [racks, setRacks] = useState([]) + const [current, setCurrent] = useState(0) + const [total, setTotal] = useState(0) + const [result, setResult] = useState(null) + const ctrlRef = useRef(null) + + useEffect(() => { + if (!open || !taskId || !accessToken) return + + setPhase('connecting') + setRacks([]) + setCurrent(0) + setTotal(0) + setResult(null) + + const ctrl = new AbortController() + ctrlRef.current = ctrl + + fetchEventSource(`/api/v1/events?task_id=${taskId}`, { + headers: { Authorization: `Bearer ${accessToken}` }, + signal: ctrl.signal, + onmessage(ev) { + try { + const data = JSON.parse(ev.data) + + if (ev.event === 'disc_server') { + setTotal(data.rack_count) + setPhase('running') + + } else if (ev.event === 'disc_rack_start') { + setTotal(data.total) + setCurrent(data.index) + setRacks((prev) => { + if (prev.find((r) => r.rack_name === data.rack_name)) { + return prev.map((r) => + r.rack_name === data.rack_name ? { ...r, state: 'running' } : r + ) + } + return [ + ...prev, + { rack_name: data.rack_name, rack_host: data.rack_host, state: 'running', found: 0, ssh_ok: false }, + ] + }) + + } else if (ev.event === 'disc_rack_done') { + setCurrent((c) => c + 1) + setRacks((prev) => + prev.map((r) => + r.rack_name === data.rack_name + ? { ...r, state: data.ssh_ok ? 'done' : 'error', found: data.found ?? 0, ssh_ok: data.ssh_ok } + : r + ) + ) + + } else if (ev.event === 'disc_done') { + setResult({ + created: data.created ?? 0, + updated: data.updated ?? 0, + skipped: data.skipped ?? 0, + errors: data.errors ?? [], + }) + setPhase('done') + setCurrent(data.rack_count ?? total) + ctrl.abort() + onComplete?.() + } + } catch { + // ignore parse errors + } + }, + onerror() { + ctrl.abort() + }, + }) + + return () => ctrl.abort() + }, [open, taskId, accessToken]) + + const percent = total > 0 ? Math.round((current / total) * 100) : 0 + + const rackIcon = (state: RackStatus['state']) => { + if (state === 'running') return + if (state === 'done') return + if (state === 'error') return + return + } + + return ( + + {/* Connecting phase */} + {phase === 'connecting' && ( + + + Подключение к серверу... + + )} + + {/* Progress bar */} + {phase !== 'connecting' && ( +
+ + {phase === 'done' ? 'Завершено' : `Обход стоек: ${current} / ${total}`} + + +
+ )} + + {/* Rack list */} + {racks.length > 0 && ( +
+ {racks.map((r) => ( +
+ {rackIcon(r.state)} + {r.rack_name} + + {r.rack_host} + + {r.state === 'done' && ( + + {r.found} уст. + + )} + {r.state === 'error' && SSH ошибка} +
+ ))} +
+ )} + + {/* Final result */} + {phase === 'done' && result && ( +
+ + +{result.created} добавлено + {result.updated} обновлено + {result.skipped} без изменений + + {result.errors.length > 0 && ( +
+ {result.errors.map((e, i) => ( + + ⚠ {e} + + ))} +
+ )} +
+ )} + + ) +} diff --git a/frontend/src/pages/InventoryObjectDetail.tsx b/frontend/src/pages/InventoryObjectDetail.tsx index b8c674d..acacdd8 100644 --- a/frontend/src/pages/InventoryObjectDetail.tsx +++ b/frontend/src/pages/InventoryObjectDetail.tsx @@ -35,6 +35,7 @@ import { } from 'antd' import type { UploadFile } from 'antd' import { useMemo, useState } from 'react' +import { useQueryClient } from '@tanstack/react-query' import { useNavigate, useParams } from 'react-router-dom' import { useCreateDevice, @@ -55,6 +56,7 @@ import { import { useRacks, useCreateRack, useDeleteRack, type RackItem } from '../api/racks' import { useZones, useCreateZone, useDeleteZone, type ZoneItem } from '../api/zones' import { CategoryTag, DeviceStatusBadge } from '../components/StatusBadge' +import { DiscoveryDrawer } from '../components/DiscoveryDrawer' import { stripEmpty } from '../utils/form' import dayjs from 'dayjs' @@ -93,6 +95,7 @@ export function InventoryObjectDetailPage() { const { id } = useParams<{ id: string }>() const objId = Number(id) const navigate = useNavigate() + const queryClient = useQueryClient() const { data: obj, isLoading: objLoading } = useObject(objId) const { data: devices, isLoading: devLoading } = useDevices(objId) @@ -118,6 +121,8 @@ export function InventoryObjectDetailPage() { const [csvModal, setCsvModal] = useState(false) const [rackModal, setRackModal] = useState(false) const [sheetsModal, setSheetsModal] = useState(false) + const [discoveryTaskId, setDiscoveryTaskId] = useState(null) + const [discoveryOpen, setDiscoveryOpen] = useState(false) const [csvFile, setCsvFile] = useState(null) const [csvPreview, setCsvPreview] = useState(null) const [searchText, setSearchText] = useState('') @@ -143,18 +148,18 @@ export function InventoryObjectDetailPage() { const handleDiscover = async () => { try { - const result = await discoverMutation.mutateAsync() - const msg = `SSH Discovery: +${result.created} найдено, ${result.updated} обновлено` - if (result.errors.length > 0) { - message.warning(`${msg}. Ошибки: ${result.errors.join('; ')}`) - } else { - message.success(msg) - } + const { task_id } = await discoverMutation.mutateAsync() + setDiscoveryTaskId(task_id) + setDiscoveryOpen(true) } catch { message.error('Ошибка SSH discovery') } } + const handleDiscoveryComplete = () => { + queryClient.invalidateQueries({ queryKey: ['devices', objId] }) + } + const handleSyncSheets = async () => { try { const values = await sheetsForm.validateFields() @@ -813,6 +818,14 @@ export function InventoryObjectDetailPage() { + setDiscoveryOpen(false)} + objectName={obj?.name ?? ''} + taskId={discoveryTaskId} + onComplete={handleDiscoveryComplete} + /> + {/* Google Sheets Sync Modal */}