Coverage for app/services/deployment

1"""Live-status joiner for the deployment Infrastructure tab.

3Marries the cached Terraform state (parsed by ``tf_state_parser``)

4with live OpenStack data via the per-user ``Connection`` helper from

5``openstack_client``. Two stages, one per endpoint:

7* Stage 1 (list view, ``build_resource_views``): one

8 ``Connection.compute.find_server`` per compute instance, in a small

9 ThreadPool with per-call timeout. Cheap enough for every list GET.

11* Stage 2 (drawer view, ``build_resource_detail``): for a single

12 compute instance, pull ports / security-group rule counts / volume

13 attachments. Called only when the user opens the drawer.

15A single live-fetch failure must not fail the whole endpoint: each VM

16degrades independently to ``drift=missing`` (fetch returned nothing) or

17``drift=stale`` (fetch raised), and TF-cached attributes survive so the

18UI can still render something.

19"""

21from __future__ import annotations

23import logging

24from concurrent.futures import ThreadPoolExecutor, as_completed

25from dataclasses import dataclass, field

26from typing import Any, Literal

28from sqlalchemy.orm import Session

30from app.models import User

31from app.services.openstack_client import user_connection

32from app.services.tf_state_parser import TfResource, parse_tf_state

34logger = logging.getLogger(__name__)

37# Per-VM live-fetch budget. The ThreadPool fans out so total endpoint

38# latency is ``max(per_call)`` not ``sum(per_call)``.

39_PER_SERVER_TIMEOUT_S = 5.0

40_MAX_PARALLEL_FETCHES = 8

43# Live-status enums kept narrow so the frontend can switch on them.

44# ``status``/``vm_state``/``power_state``/``task_state`` pass through

45# as-is from OpenStack (stable vocabulary: ACTIVE/BUILD/ERROR/...).

46ResourceDrift = Literal["in_sync", "stale", "missing"]

49@dataclass

50class LifecycleStates:

51 """Server lifecycle quad — together they cover every diagnostic case."""

52 status: str | None = None

53 task_state: str | None = None

54 vm_state: str | None = None

55 power_state: str | None = None

56 # Only set when ``status == "ERROR"`` — Nova fills ``fault.message``

57 # with the underlying error the user needs to act on.

58 fault_message: str | None = None

61@dataclass

62class HardwareSpec:

63 """Compact hardware/image footprint for the card."""

64 flavor_name: str | None = None

65 ram_mb: int | None = None

66 vcpus: int | None = None

67 disk_gb: int | None = None

68 image_id: str | None = None

69 # Stage-1 leaves image_name unresolved (no extra round-trip);

70 # stage-2 fills it via ``conn.image.find_image``.

71 image_name: str | None = None

72 availability_zone: str | None = None

73 # ISO-8601 timestamp from ``OS-SRV-USG:launched_at``; the frontend

74 # computes uptime against ``now``.

75 launched_at: str | None = None

78@dataclass

79class NetworkAddress:

80 """One entry per NIC × IP. Networks with multiple addresses (fixed

81 + floating) show up as several rows under the same network name."""

82 network: str

83 fixed_ip: str | None = None

84 floating_ip: str | None = None

85 mac: str | None = None

88@dataclass

89class NetworkPort:

90 """Stage-2: full neutron port info for one NIC of the server."""

91 port_id: str

92 network_id: str | None

93 status: str | None

94 mac: str | None

95 fixed_ip: str | None

96 security_group_ids: list[str] = field(default_factory=list)

99@dataclass

100class SecurityGroupSummary:

101 """Stage-2: a count summary plus SG metadata (not the full rule dump)."""

102 id: str

103 name: str

104 description: str | None

105 ingress_rules: int

106 egress_rules: int

107

108

109@dataclass

110class VolumeAttachment:

111 """Stage-2: one row per attached volume."""

112 volume_id: str

113 device: str | None # mountpoint inside the guest, e.g. "/dev/vdb"

114 size_gb: int | None

115 bootable: bool | None

116 status: str | None

117 name: str | None

118

119

120@dataclass

121class DeploymentResourceView:

122 """One row in the resource list (stage 1) or the response of the

123 detail endpoint (stage 2 — same shape, more fields populated)."""

124 address: str

125 type: str

126 category: str

127 team: str | None

128 provider_id: str

129 display_name: str

130 drift: ResourceDrift = "in_sync"

131 # Stage 1 (instance-only):

132 lifecycle: LifecycleStates | None = None

133 hardware: HardwareSpec | None = None

134 addresses: list[NetworkAddress] = field(default_factory=list)

135 # Stage 2 (instance-only, only set when ``include_detail=True``):

136 ports: list[NetworkPort] | None = None

137 security_groups: list[SecurityGroupSummary] | None = None

138 volumes: list[VolumeAttachment] | None = None

139 metadata: dict[str, str] | None = None

140

141

142# ----------------------------------------------------------------

143# Stage 1: list view

144# ----------------------------------------------------------------

145def build_resource_views(

146 *,

147 db: Session,

148 user: User,

149 tf_state_json: str | None,

150 refresh: bool,

151) -> list[DeploymentResourceView]:

152 """Parse the cached state and (when refresh=True) overlay live data.

153

154 Non-compute resources travel through with TF-cached data only;

155 the UI shows them in read-only network/security sections, where

156 a live refresh isn't meaningful.

157 """

158 cached = parse_tf_state(tf_state_json)

159 if not cached:

160 return []

161

162 views: list[DeploymentResourceView] = [

163 _view_from_cached(r) for r in cached

164 ]

165

166 if not refresh:

167 return views

168

169 instance_views = [v for v in views if v.category == "instance"]

170 if not instance_views:

171 return views

172

173 # Single connection, reused across the fan-out. Closing it on

174 # exit happens in the user_connection contextmanager.

175 with user_connection(db, user) as conn:

176 _enrich_instances_stage1(conn, instance_views)

177 return views

178

179

180def _view_from_cached(r: TfResource) -> DeploymentResourceView:

181 """Build the base view from cached state alone — used as the

182 starting point for both stage 1 and stage 2 enrichment."""

183 return DeploymentResourceView(

184 address=r.address,

185 type=r.type,

186 category=r.category,

187 team=r.team,

188 provider_id=r.provider_id,

189 display_name=r.display_name,

190 drift="in_sync",

191 )

192

193

194def _enrich_instances_stage1(

195 conn: Any, instance_views: list[DeploymentResourceView]

196) -> None:

197 """Fetch each server and write lifecycle/hardware/addresses.

198

199 Mutates ``instance_views`` in place. Failures are absorbed per-VM:

200 the view degrades to ``drift = stale`` / ``drift = missing`` and

201 other VMs continue.

202 """

203 def _one(view: DeploymentResourceView) -> None:

204 try:

205 server = conn.compute.find_server(view.provider_id, ignore_missing=True)

206 except Exception as exc: # noqa: BLE001 — SDK + transport

207 logger.warning(

208 "stage1: find_server(%s) failed: %s", view.provider_id, exc

209 )

210 view.drift = "stale"

211 return

212 if server is None:

213 view.drift = "missing"

214 return

215 view.lifecycle = _lifecycle_from(server)

216 view.hardware = _hardware_from(server)

217 view.addresses = _addresses_from(server)

218

219 # ThreadPool fans out the per-server calls with a per-call timeout

220 # via ``Future.result(timeout=...)``. A timed-out future is

221 # abandoned (the underlying HTTP request isn't cancelled).

222 if not instance_views:

223 return

224 workers = min(_MAX_PARALLEL_FETCHES, len(instance_views))

225 with ThreadPoolExecutor(max_workers=workers) as pool:

226 futures = {pool.submit(_one, v): v for v in instance_views}

227 for fut in as_completed(futures, timeout=_PER_SERVER_TIMEOUT_S * 4):

228 try:

229 fut.result(timeout=_PER_SERVER_TIMEOUT_S)

230 except Exception as exc: # noqa: BLE001

231 view = futures[fut]

232 logger.warning(

233 "stage1 timeout/error for %s: %s", view.provider_id, exc

234 )

235 if view.drift == "in_sync":

236 view.drift = "stale"

237

238

239# ----------------------------------------------------------------

240# Stage 2: detail view

241# ----------------------------------------------------------------

242def build_resource_detail(

243 *,

244 db: Session,

245 user: User,

246 tf_state_json: str | None,

247 address: str,

248) -> DeploymentResourceView | None:

249 """Return one resource enriched with stage-2 data (ports, SGs,

250 volumes, resolved image name, full metadata).

251

252 Returns None if the address isn't a known instance in the cached

253 state — the endpoint translates that into a 404.

254 """

255 cached = parse_tf_state(tf_state_json)

256 target = next((r for r in cached if r.address == address), None)

257 if target is None or target.category != "instance":

258 return None

259

260 view = _view_from_cached(target)

261 with user_connection(db, user) as conn:

262 try:

263 server = conn.compute.find_server(

264 target.provider_id, ignore_missing=True

265 )

266 except Exception as exc: # noqa: BLE001

267 logger.warning(

268 "stage2: find_server(%s) failed: %s", target.provider_id, exc

269 )

270 view.drift = "stale"

271 return view

272 if server is None:

273 view.drift = "missing"

274 return view

275

276 view.lifecycle = _lifecycle_from(server)

277 view.hardware = _hardware_from(server)

278 view.addresses = _addresses_from(server)

279 view.metadata = dict(getattr(server, "metadata", None) or {})

280

281 # Resolve the image name once. Compute-API only returns the

282 # ID; the human-friendly label lives in the Image service.

283 image_id = view.hardware.image_id if view.hardware else None

284 if image_id:

285 view.hardware.image_name = _resolve_image_name(conn, image_id)

286

287 # Ports + SG summaries + volumes. Each block degrades to []

288 # on failure — a partial drawer beats a 500.

289 view.ports = _fetch_ports(conn, target.provider_id)

290 view.security_groups = _fetch_sg_summaries(conn, view.ports)

291 view.volumes = _fetch_volumes(conn, target.provider_id)

292

293 return view

294

295

296def _fetch_ports(conn: Any, server_id: str) -> list[NetworkPort]:

297 out: list[NetworkPort] = []

298 try:

299 ports = list(conn.network.ports(device_id=server_id))

300 except Exception as exc: # noqa: BLE001

301 logger.warning("stage2: ports(%s) failed: %s", server_id, exc)

302 return out

303 for p in ports:

304 fixed_ips = getattr(p, "fixed_ips", None) or []

305 fixed_ip = None

306 if fixed_ips:

307 # Each entry is ``{"ip_address": ..., "subnet_id": ...}``.

308 fixed_ip = (fixed_ips[0] or {}).get("ip_address")

309 out.append(

310 NetworkPort(

311 port_id=str(getattr(p, "id", None) or ""),

312 network_id=getattr(p, "network_id", None),

313 status=getattr(p, "status", None),

314 mac=getattr(p, "mac_address", None),

315 fixed_ip=fixed_ip,

316 security_group_ids=list(getattr(p, "security_group_ids", None) or []),

317 )

318 )

319 return out

320

321

322def _fetch_sg_summaries(

323 conn: Any, ports: list[NetworkPort]

324) -> list[SecurityGroupSummary]:

325 """Resolve the distinct SG IDs referenced by the ports into a

326 per-SG summary with ingress/egress counts (not the full rule list)."""

327 sg_ids: set[str] = set()

328 for p in ports:

329 for sid in p.security_group_ids:

330 sg_ids.add(sid)

331 if not sg_ids:

332 return []

333 out: list[SecurityGroupSummary] = []

334 for sid in sg_ids:

335 try:

336 sg = conn.network.find_security_group(sid, ignore_missing=True)

337 except Exception as exc: # noqa: BLE001

338 logger.warning("stage2: find_security_group(%s) failed: %s", sid, exc)

339 continue

340 if sg is None:

341 continue

342 rules = list(getattr(sg, "security_group_rules", None) or [])

343

344 def _direction(r: Any) -> Any:

345 return r.get("direction") if isinstance(r, dict) else getattr(r, "direction", None)

346

347 ingress = sum(1 for r in rules if _direction(r) == "ingress")

348 egress = sum(1 for r in rules if _direction(r) == "egress")

349 out.append(

350 SecurityGroupSummary(

351 id=str(getattr(sg, "id", sid)),

352 name=str(getattr(sg, "name", "") or sid),

353 description=getattr(sg, "description", None),

354 ingress_rules=ingress,

355 egress_rules=egress,

356 )

357 )

358 return out

359

360

361def _fetch_volumes(conn: Any, server_id: str) -> list[VolumeAttachment]:

362 """Build the per-volume attachment list. Two API hops: compute for

363 attachments (volume_id + device), block_storage for

364 size/bootable/status. Either side may fail independently — a missing

365 block_storage just leaves the size fields None."""

366 out: list[VolumeAttachment] = []

367 try:

368 attachments = list(conn.compute.volume_attachments(server=server_id))

369 except Exception as exc: # noqa: BLE001

370 logger.warning("stage2: volume_attachments(%s) failed: %s", server_id, exc)

371 return out

372

373 for att in attachments:

374 vol_id = getattr(att, "volume_id", None) or getattr(att, "id", None)

375 if not vol_id:

376 continue

377 device = getattr(att, "device", None)

378 size = bootable = status = name = None

379 try:

380 vol = conn.block_storage.find_volume(vol_id, ignore_missing=True)

381 except Exception as exc: # noqa: BLE001

382 logger.warning("stage2: find_volume(%s) failed: %s", vol_id, exc)

383 vol = None

384 if vol is not None:

385 size = getattr(vol, "size", None)

386 bootable_raw = getattr(vol, "is_bootable", None)

387 if bootable_raw is None:

388 bootable_raw = getattr(vol, "bootable", None)

389 if isinstance(bootable_raw, str):

390 bootable = bootable_raw.lower() == "true"

391 elif isinstance(bootable_raw, bool):

392 bootable = bootable_raw

393 status = getattr(vol, "status", None)

394 name = getattr(vol, "name", None)

395 out.append(

396 VolumeAttachment(

397 volume_id=str(vol_id),

398 device=device,

399 size_gb=size,

400 bootable=bootable,

401 status=status,

402 name=name,

403 )

404 )

405 return out

406

407

408def _resolve_image_name(conn: Any, image_id: str) -> str | None:

409 try:

410 img = conn.image.find_image(image_id, ignore_missing=True)

411 except Exception as exc: # noqa: BLE001

412 logger.warning("stage2: find_image(%s) failed: %s", image_id, exc)

413 return None

414 return getattr(img, "name", None) if img is not None else None

415

416

417# ----------------------------------------------------------------

418# Server → dataclass adapters

419# ----------------------------------------------------------------

420def _lifecycle_from(server: Any) -> LifecycleStates:

421 """Pull the four lifecycle states + fault message from a server.

422

423 Tries the canonical snake_case attribute first, falling back to the

424 raw ``OS-EXT-*`` keys, so it's robust to either SDK flavor.

425 """

426 status = getattr(server, "status", None)

427 fault = None

428 if status == "ERROR":

429 fault_obj = getattr(server, "fault", None)

430 if isinstance(fault_obj, dict):

431 fault = fault_obj.get("message")

432 else:

433 fault = getattr(fault_obj, "message", None) if fault_obj else None

434

435 return LifecycleStates(

436 status=status,

437 task_state=getattr(server, "task_state", None),

438 vm_state=getattr(server, "vm_state", None),

439 power_state=_translate_power_state(getattr(server, "power_state", None)),

440 fault_message=fault,

441 )

442

443

444# Nova's ``power_state`` is an integer enum. We map to human-readable

445# labels so the frontend doesn't need to know the table.

446# https://docs.openstack.org/nova/latest/admin/manage-vm-states.html

447_POWER_STATE_LABELS = {

448 0: "NOSTATE",

449 1: "RUNNING",

450 3: "PAUSED",

451 4: "SHUTDOWN",

452 6: "CRASHED",

453 7: "SUSPENDED",

454}

455

456

457def _translate_power_state(raw: Any) -> str | None:

458 if raw is None:

459 return None

460 if isinstance(raw, str):

461 return raw

462 if isinstance(raw, int):

463 return _POWER_STATE_LABELS.get(raw, f"UNKNOWN({raw})")

464 return None

465

466

467def _hardware_from(server: Any) -> HardwareSpec:

468 flavor = getattr(server, "flavor", None) or {}

469 if not isinstance(flavor, dict):

470 # SDK may return a Munch / typed object — coerce best-effort

471 flavor = {

472 "original_name": getattr(flavor, "original_name", None),

473 "ram": getattr(flavor, "ram", None),

474 "vcpus": getattr(flavor, "vcpus", None),

475 "disk": getattr(flavor, "disk", None),

476 }

477 image = getattr(server, "image", None) or {}

478 image_id = image.get("id") if isinstance(image, dict) else getattr(image, "id", None)

479

480 return HardwareSpec(

481 flavor_name=flavor.get("original_name") or flavor.get("name"),

482 ram_mb=flavor.get("ram"),

483 vcpus=flavor.get("vcpus"),

484 disk_gb=flavor.get("disk"),

485 image_id=image_id,

486 image_name=None, # filled by stage-2 only

487 availability_zone=getattr(server, "availability_zone", None),

488 launched_at=getattr(server, "launched_at", None),

489 )

490

491

492def _addresses_from(server: Any) -> list[NetworkAddress]:

493 """Flatten the ``addresses`` dict into one row per (network, IP).

494

495 OpenStack's ``addresses`` is ``{network_name: [{"addr": ..., "type":

496 "fixed"|"floating", ...}, ...]}``. We pair fixed and floating IPs

497 that share the same network name into a single row when both

498 exist, otherwise emit one row per address.

499 """

500 raw = getattr(server, "addresses", None) or {}

501 if not isinstance(raw, dict):

502 return []

503 out: list[NetworkAddress] = []

504 for network_name, entries in raw.items():

505 if not isinstance(entries, list):

506 continue

507 fixed_ip = floating_ip = mac = None

508 for entry in entries:

509 if not isinstance(entry, dict):

510 continue

511 addr = entry.get("addr")

512 kind = entry.get("OS-EXT-IPS:type") or entry.get("type")

513 if kind == "floating":

514 floating_ip = addr

515 else:

516 # Some clouds omit ``OS-EXT-IPS:type``; assume fixed.

517 if fixed_ip is None:

518 fixed_ip = addr

519 mac = entry.get("OS-EXT-IPS-MAC:mac_addr") or entry.get("mac_addr")

520 out.append(

521 NetworkAddress(

522 network=str(network_name),

523 fixed_ip=fixed_ip,

524 floating_ip=floating_ip,

525 mac=mac,

526 )

527 )

528 return out

Coverage for app/services/deployment_status.py: 85.66%

265 statements