dv/utp_service
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

фигня какая-то ты дурной

Browse source
This commit is contained in:
dv 2026-04-29 14:56:02 +03:00
parent a774b613c7
commit e4b993dbf3
2 changed files with 90 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

72
backend/app/services/discovery/orchestrator.py
View file

@ -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 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
)
return True, config_text
return True, config_text, ssh_port
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,
"[discovery] Reading rack config from %s:%d timed out; retrying (%d/2)",
rack_host, ssh_port, attempt + 1,
)
await _emit("disc_action", {
"message": f"Повторное чтение конфигурации {rack_host} ({rack_name})..."
"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: {last_exc}")
return False, None
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}")

24
backend/app/services/discovery/ssh_client.py
View file

@ -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."""
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:
return await ssh_run_multi(host, port, options, f"cat {config_path}", timeout=30)
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 timed out; retrying (%d/2)",
host, attempt + 1,
"[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 failed: %s", host, exc)
return None
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 failed: %s", host, exc)
return None
return None
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(