sse ssh discovery
This commit is contained in:
parent
56b0eaa495
commit
c8429814bf
6 changed files with 300 additions and 26 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -90,3 +90,7 @@ class SyncResult(BaseModel):
|
|||
updated: int
|
||||
skipped: int
|
||||
errors: list[str] = []
|
||||
|
||||
|
||||
class DiscoveryStarted(BaseModel):
|
||||
task_id: str
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -220,15 +220,16 @@ export const useSyncObject = (objId: number) => {
|
|||
})
|
||||
}
|
||||
|
||||
export const useDiscoverObject = (objId: number) => {
|
||||
const qc = useQueryClient()
|
||||
return useMutation({
|
||||
mutationFn: () =>
|
||||
api.post<SyncResult>(`/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<DiscoveryStarted>(`/api/v1/objects/${objId}/discover`).then((r) => r.data),
|
||||
})
|
||||
|
||||
export interface SheetsSyncPayload {
|
||||
spreadsheet_id: string
|
||||
sheet_name?: string
|
||||
|
|
|
|||
210
frontend/src/components/DiscoveryDrawer.tsx
Normal file
210
frontend/src/components/DiscoveryDrawer.tsx
Normal file
|
|
@ -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<RackStatus[]>([])
|
||||
const [current, setCurrent] = useState(0)
|
||||
const [total, setTotal] = useState(0)
|
||||
const [result, setResult] = useState<DiscoveryResult | null>(null)
|
||||
const ctrlRef = useRef<AbortController | null>(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 <LoadingOutlined style={{ color: '#1677ff' }} />
|
||||
if (state === 'done') return <CheckCircleOutlined style={{ color: '#52c41a' }} />
|
||||
if (state === 'error') return <CloseCircleOutlined style={{ color: '#ff4d4f' }} />
|
||||
return <MinusCircleOutlined style={{ color: '#d9d9d9' }} />
|
||||
}
|
||||
|
||||
return (
|
||||
<Drawer
|
||||
title={`SSH Discovery — ${objectName}`}
|
||||
open={open}
|
||||
onClose={onClose}
|
||||
width={440}
|
||||
footer={null}
|
||||
>
|
||||
{/* Connecting phase */}
|
||||
{phase === 'connecting' && (
|
||||
<Space style={{ marginBottom: 16 }}>
|
||||
<LoadingOutlined />
|
||||
<Typography.Text>Подключение к серверу...</Typography.Text>
|
||||
</Space>
|
||||
)}
|
||||
|
||||
{/* Progress bar */}
|
||||
{phase !== 'connecting' && (
|
||||
<div style={{ marginBottom: 16 }}>
|
||||
<Typography.Text type="secondary" style={{ fontSize: 12 }}>
|
||||
{phase === 'done' ? 'Завершено' : `Обход стоек: ${current} / ${total}`}
|
||||
</Typography.Text>
|
||||
<Progress
|
||||
percent={phase === 'done' ? 100 : percent}
|
||||
status={phase === 'done' ? 'success' : 'active'}
|
||||
style={{ marginBottom: 0 }}
|
||||
/>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* Rack list */}
|
||||
{racks.length > 0 && (
|
||||
<div style={{ marginBottom: 16 }}>
|
||||
{racks.map((r) => (
|
||||
<div
|
||||
key={r.rack_name}
|
||||
style={{
|
||||
display: 'flex',
|
||||
alignItems: 'center',
|
||||
padding: '5px 0',
|
||||
gap: 8,
|
||||
borderBottom: '1px solid #f0f0f0',
|
||||
}}
|
||||
>
|
||||
{rackIcon(r.state)}
|
||||
<Typography.Text style={{ flex: 1 }}>{r.rack_name}</Typography.Text>
|
||||
<Typography.Text type="secondary" style={{ fontSize: 12 }}>
|
||||
{r.rack_host}
|
||||
</Typography.Text>
|
||||
{r.state === 'done' && (
|
||||
<Tag color="blue" style={{ marginLeft: 4 }}>
|
||||
{r.found} уст.
|
||||
</Tag>
|
||||
)}
|
||||
{r.state === 'error' && <Tag color="red">SSH ошибка</Tag>}
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* Final result */}
|
||||
{phase === 'done' && result && (
|
||||
<div style={{ borderTop: '1px solid #f0f0f0', paddingTop: 12 }}>
|
||||
<Space wrap>
|
||||
<Tag color="green">+{result.created} добавлено</Tag>
|
||||
<Tag color="blue">{result.updated} обновлено</Tag>
|
||||
<Tag>{result.skipped} без изменений</Tag>
|
||||
</Space>
|
||||
{result.errors.length > 0 && (
|
||||
<div style={{ marginTop: 10 }}>
|
||||
{result.errors.map((e, i) => (
|
||||
<Typography.Text
|
||||
key={i}
|
||||
type="danger"
|
||||
style={{ display: 'block', fontSize: 12, marginBottom: 4 }}
|
||||
>
|
||||
⚠ {e}
|
||||
</Typography.Text>
|
||||
))}
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
)}
|
||||
</Drawer>
|
||||
)
|
||||
}
|
||||
|
|
@ -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<string | null>(null)
|
||||
const [discoveryOpen, setDiscoveryOpen] = useState(false)
|
||||
const [csvFile, setCsvFile] = useState<File | null>(null)
|
||||
const [csvPreview, setCsvPreview] = useState<DeviceImportPreview | null>(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() {
|
|||
</Form>
|
||||
</Modal>
|
||||
|
||||
<DiscoveryDrawer
|
||||
open={discoveryOpen}
|
||||
onClose={() => setDiscoveryOpen(false)}
|
||||
objectName={obj?.name ?? ''}
|
||||
taskId={discoveryTaskId}
|
||||
onComplete={handleDiscoveryComplete}
|
||||
/>
|
||||
|
||||
{/* Google Sheets Sync Modal */}
|
||||
<Modal
|
||||
title="Синхронизация из Google Sheets"
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue