From e4b993dbf3fd61245d0869f571fde616dee98720 Mon Sep 17 00:00:00 2001 From: dv Date: Wed, 29 Apr 2026 14:56:02 +0300 Subject: [PATCH] =?UTF-8?q?=D1=84=D0=B8=D0=B3=D0=BD=D1=8F=20=D0=BA=D0=B0?= =?UTF-8?q?=D0=BA=D0=B0=D1=8F-=D1=82=D0=BE=20=D1=82=D1=8B=20=D0=B4=D1=83?= =?UTF-8?q?=D1=80=D0=BD=D0=BE=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/services/discovery/orchestrator.py | 104 ++++++++++++------ backend/app/services/discovery/ssh_client.py | 40 +++---- 2 files changed, 90 insertions(+), 54 deletions(-) diff --git a/backend/app/services/discovery/orchestrator.py b/backend/app/services/discovery/orchestrator.py index eaa94f4..3fdd4f5 100644 --- a/backend/app/services/discovery/orchestrator.py +++ b/backend/app/services/discovery/orchestrator.py @@ -61,6 +61,7 @@ async def discover_via_ssh( return result ssh_port = obj.ssh_port or 22 + embedded_ssh_ports = _embedded_ssh_ports(obj) # First option's password is used by single-auth Proxmox scan (legacy). _first = ssh_options[0] ssh_password = _first.password if _first.type == "password" else "" @@ -141,18 +142,18 @@ async def discover_via_ssh( db, obj, rack_host, config_path ) - rack_ssh_ok, config_text = await _ssh_fetch_rack_config( - rack_host, ssh_port, ssh_options, effective_config_path, rack_name, result, _emit + rack_ssh_ok, config_text, rack_ssh_port = await _ssh_fetch_rack_config( + rack_host, embedded_ssh_ports, ssh_options, effective_config_path, rack_name, result, _emit ) _rack_ssh_ok = rack_ssh_ok await _upsert_rack_device( db, obj, external_id, rack_host, rack_name, - rack_id_mapping, rack_ssh_ok, now, result + rack_id_mapping, rack_ssh_ok, rack_ssh_port, now, result ) if rack_ssh_ok: - neigh_ips = await get_ip_neighbors_multi(rack_host, ssh_port, ssh_options) + neigh_ips = await get_ip_neighbors_multi(rack_host, rack_ssh_port or embedded_ssh_ports[0], ssh_options) neighbor_candidates.update(neigh_ips) discovered_device_ips.add(rack_host) @@ -208,7 +209,7 @@ async def discover_via_ssh( await _emit("disc_action", {"message": "Слейв-стойки: ARP-кандидатов не найдено"}) await _discover_slave_racks( - db, obj, slave_candidates, rack_hosts, ssh_port, ssh_options, + db, obj, slave_candidates, rack_hosts, embedded_ssh_ports, ssh_options, config_path, all_rack_ips_set, now, result, _emit ) @@ -229,6 +230,15 @@ async def discover_via_ssh( # ── Phase helpers ───────────────────────────────────────────────────────────── +def _embedded_ssh_ports(obj: Object) -> list[int]: + """Return SSH ports to try for embedded rack PCs.""" + ports = [22] + object_port = obj.ssh_port or 22 + if object_port not in ports: + ports.append(object_port) + return ports + + async def _resolve_db_connection( obj: Object, ssh_port: int, @@ -475,44 +485,45 @@ async def _resolve_rack_config_path( async def _ssh_fetch_rack_config( rack_host: str, - ssh_port: int, + ssh_ports: list[int], ssh_options: list, config_path: str, rack_name: str, result: DiscoveryResult, _emit, -) -> tuple[bool, str | None]: +) -> tuple[bool, str | None, int | None]: """Attempt to SSH into rack_host and read its config file. - Returns (ssh_ok, config_text). + Returns (ssh_ok, config_text, ssh_port_used). """ - await _emit("disc_action", {"message": f"SSH → {rack_host} ({rack_name})"}) + await _emit("disc_action", {"message": f"SSH -> {rack_host} ({rack_name})"}) last_exc: Exception | None = None - for attempt in range(1, 3): - try: - config_text = await ssh_run_multi( - rack_host, ssh_port, ssh_options, f"cat {config_path}", timeout=30 - ) - return True, config_text - except TimeoutError as exc: - last_exc = exc - if attempt < 2: - logger.warning( - "[discovery] Reading rack config from %s timed out; retrying (%d/2)", - rack_host, attempt + 1, + for ssh_port in ssh_ports: + for attempt in range(1, 3): + try: + config_text = await ssh_run_multi( + rack_host, ssh_port, ssh_options, f"cat {config_path}", timeout=30 ) - await _emit("disc_action", { - "message": f"Повторное чтение конфигурации {rack_host} ({rack_name})..." - }) - continue - break - except Exception as exc: - last_exc = exc - break - - result.errors.append(f"SSH to {rack_host} failed: {last_exc}") - return False, None + return True, config_text, ssh_port + except TimeoutError as exc: + last_exc = exc + if attempt < 2: + logger.warning( + "[discovery] Reading rack config from %s:%d timed out; retrying (%d/2)", + rack_host, ssh_port, attempt + 1, + ) + await _emit("disc_action", { + "message": f"Retrying config read {rack_host}:{ssh_port} ({rack_name})..." + }) + continue + break + except Exception as exc: + last_exc = exc + logger.info("[discovery] SSH to rack %s:%d failed: %s", rack_host, ssh_port, exc) + break + result.errors.append(f"SSH to {rack_host} failed on ports {ssh_ports}: {last_exc}") + return False, None, None async def _upsert_rack_device( db: AsyncSession, @@ -522,6 +533,7 @@ async def _upsert_rack_device( rack_name: str, rack_id_mapping: dict, rack_ssh_ok: bool, + rack_ssh_port: int | None, now: datetime, result: DiscoveryResult, ) -> None: @@ -531,6 +543,11 @@ async def _upsert_rack_device( "[discovery] Processing rack %s (%s), ssh_ok=%s", rack_host, rack_hostname, rack_ssh_ok, ) + rack_port_override = ( + rack_ssh_port + if rack_ssh_ok and rack_ssh_port and rack_ssh_port != (obj.ssh_port or 22) + else None + ) try: existing = (await db.execute( select(Device).where(Device.object_id == obj.id, Device.ip == rack_host) @@ -543,6 +560,7 @@ async def _upsert_rack_device( category="embedded", role="other", source="auto_ssh", rack_id=rack_id_mapping.get(external_id), status=status, last_seen=now if rack_ssh_ok else None, + ssh_port_override=rack_port_override, )) logger.info("[discovery] %s (%s) - ADDED: status=%s", rack_host, rack_hostname, status) result.created += 1 @@ -559,6 +577,9 @@ async def _upsert_rack_device( changes.append("status: →online") existing.last_seen = now changes.append("last_seen: updated") + if existing.ssh_port_override != rack_port_override: + existing.ssh_port_override = rack_port_override + changes.append(f"ssh_port_override: ->{rack_port_override or 'default'}") if existing.rack_id is None and rack_id_mapping.get(external_id) is not None: existing.rack_id = rack_id_mapping[external_id] changes.append(f"rack_id: None→{existing.rack_id}") @@ -718,7 +739,7 @@ async def _discover_slave_racks( obj: Object, slave_candidates: set[str], rack_hosts: list[tuple], - ssh_port: int, + ssh_ports: list[int], ssh_options: list, config_path: str, all_rack_ips_set: set[str], @@ -731,7 +752,7 @@ async def _discover_slave_racks( await _emit("disc_action", {"message": f"Проверяем слейв-кандидата {candidate_ip}..."}) try: - slave_config = await fetch_slave_config(candidate_ip, ssh_port, ssh_options, config_path) + slave_config, slave_ssh_port = await fetch_slave_config(candidate_ip, ssh_ports, ssh_options, config_path) is_slave = False slave_identifier = "" master_ip_for_slave = "" @@ -760,7 +781,10 @@ async def _discover_slave_racks( slave_rack_id = await _ensure_slave_rack(db, obj, slave_identifier, result) - await _upsert_slave_device(db, obj, candidate_ip, slave_identifier, slave_rack_id, master_ip_for_slave, now, result) + await _upsert_slave_device( + db, obj, candidate_ip, slave_identifier, slave_rack_id, + master_ip_for_slave, slave_ssh_port, now, result + ) all_rack_ips_with_slave = list(all_rack_ips_set) + [candidate_ip] slave_devices = parse_config(slave_config, candidate_ip, all_rack_ips_with_slave, obj.server_ip) @@ -820,9 +844,15 @@ async def _upsert_slave_device( slave_identifier: str, slave_rack_id: int | None, master_ip: str, + slave_ssh_port: int | None, now: datetime, result: DiscoveryResult, ) -> None: + slave_port_override = ( + slave_ssh_port + if slave_ssh_port and slave_ssh_port != (obj.ssh_port or 22) + else None + ) try: existing = (await db.execute( select(Device).where(Device.object_id == obj.id, Device.ip == candidate_ip) @@ -832,6 +862,7 @@ async def _upsert_slave_device( object_id=obj.id, ip=candidate_ip, hostname=slave_identifier, category="embedded", role="other", source="auto_ssh", rack_id=slave_rack_id, status="online", last_seen=now, + ssh_port_override=slave_port_override, )) logger.info("[discovery] %s (%s) - ADDED: slave rack PC", candidate_ip, slave_identifier) result.created += 1 @@ -847,6 +878,9 @@ async def _upsert_slave_device( changes.append("status: →online") existing.last_seen = now changes.append("last_seen: updated") + if existing.ssh_port_override != slave_port_override: + existing.ssh_port_override = slave_port_override + changes.append(f"ssh_port_override: ->{slave_port_override or 'default'}") if existing.rack_id is None and slave_rack_id: existing.rack_id = slave_rack_id changes.append(f"rack_id: →{slave_rack_id}") diff --git a/backend/app/services/discovery/ssh_client.py b/backend/app/services/discovery/ssh_client.py index 1511551..b0ec2e3 100644 --- a/backend/app/services/discovery/ssh_client.py +++ b/backend/app/services/discovery/ssh_client.py @@ -138,25 +138,27 @@ async def discover_mikrotik_multi(host: str, port: int, options: list) -> str | async def fetch_slave_config( - host: str, port: int, options: list, config_path: str, -) -> str | None: - """Read a remote config file from a slave candidate. Returns None on failure.""" - for attempt in range(1, 3): - try: - return await ssh_run_multi(host, port, options, f"cat {config_path}", timeout=30) - except TimeoutError as exc: - if attempt < 2: - logger.warning( - "[discovery] Reading slave config from %s timed out; retrying (%d/2)", - host, attempt + 1, - ) - continue - logger.warning("[discovery] Reading slave config from %s failed: %s", host, exc) - return None - except Exception as exc: - logger.warning("[discovery] Reading slave config from %s failed: %s", host, exc) - return None - return None + host: str, ports: list[int], options: list, config_path: str, +) -> tuple[str | None, int | None]: + """Read a remote config file from a slave candidate. Returns (config, port).""" + for port in ports: + for attempt in range(1, 3): + try: + config_text = await ssh_run_multi(host, port, options, f"cat {config_path}", timeout=30) + return config_text, port + except TimeoutError as exc: + if attempt < 2: + logger.warning( + "[discovery] Reading slave config from %s:%d timed out; retrying (%d/2)", + host, port, attempt + 1, + ) + continue + logger.warning("[discovery] Reading slave config from %s:%d failed: %s", host, port, exc) + break + except Exception as exc: + logger.warning("[discovery] Reading slave config from %s:%d failed: %s", host, port, exc) + break + return None, None async def discover_proxmox_via_port_scan(