"""Linkage analysis: Infrahub records that lost their OpenStack server, and OpenStack servers that no Infrahub record claims. A VM in ERROR is not always a failed build. Sometimes the build succeeded and only the *link* between Infrahub and OpenStack was never written - so Infrahub reports ERROR (or CREATING) with no usable openstack_id while a perfectly good server of the same name is running. Those look identical on an alert dashboard and are opposite problems: one needs a rebuild, the other needs a record fixed and is quietly billing nobody. This scans both sides in bulk and pairs them up by name. It also finds the reverse - OpenStack servers with no Infrahub record at all - which is what `Suspected Orphan VM` was meant to catch before its input metric went empty. """ from __future__ import annotations import difflib import threading import time from typing import Any, Optional from . import cxbridge REGIONS = ("ca1", "ca2", "us1", "no1") # Infrahub states where a missing OpenStack link is suspicious rather than normal. # HIBERNATED is excluded: shelved instances legitimately have no running server. UNLINKED_SUSPECT_STATES = {"ERROR", "CREATING", "BUILD", "ACTIVE", "REBOOTING", "RESTORING"} # Projects to ignore, matching the CX-Tools tempest suppression. def _is_ignorable(name: str) -> bool: low = str(name or "").lower() return "tempest" in low _REGION_ALIAS = {"canada-1": "ca1", "canada-2": "ca2", "us-1": "us1", "norway-1": "no1", "ca1": "ca1", "ca2": "ca2", "us1": "us1", "no1": "no1"} def _region_alias(region: str) -> str: return _REGION_ALIAS.get(str(region or "").strip().lower(), "") def _norm(name: str) -> str: return str(name or "").strip().lower() class Scan: """One full cross-region scan. Slow (a server list per region), so cached.""" def __init__(self): self.started = 0.0 self.finished = 0.0 self.state = "idle" # idle | running | done | error self.error = "" self.progress = "" self.result: dict[str, Any] = {} self._lock = threading.Lock() def to_json(self) -> dict[str, Any]: return { "state": self.state, "error": self.error, "progress": self.progress, "started": self.started, "finished": self.finished, "age_seconds": int(time.time() - self.finished) if self.finished else None, "result": self.result, } def run(self, snapshot: Any, regions: tuple[str, ...] = REGIONS) -> None: with self._lock: if self.state == "running": return self.state = "running" self.started = time.time() self.error = "" self.progress = "starting" try: self.result = self._scan(snapshot, regions) self.state = "done" except Exception as exc: # surfaced in the UI rather than crashing the server self.state = "error" self.error = f"{type(exc).__name__}: {exc}" finally: self.finished = time.time() self.progress = "" # --- the analysis ------------------------------------------------------ def _scan(self, snapshot: Any, regions: tuple[str, ...]) -> dict[str, Any]: # 1. Everything OpenStack has, per region. os_by_id: dict[str, dict[str, str]] = {} os_by_name: dict[str, list[dict[str, str]]] = {} region_counts: dict[str, int] = {} failures: dict[str, str] = {} for region in regions: self.progress = f"listing OpenStack servers in {region}" ok, rows, raw = _server_list(region) if not ok: failures[region] = raw[:200] continue region_counts[region] = len(rows) for row in rows: sid = str(row.get("ID") or "") if not sid: continue rec = { "id": sid, "name": str(row.get("Name") or ""), "status": str(row.get("Status") or ""), "task": str(row.get("Task State") or ""), "host": str(row.get("Host") or ""), "project_id": str(row.get("Project ID") or ""), "flavor": str(row.get("Flavor") or ""), "region": region, } os_by_id[sid] = rec os_by_name.setdefault(_norm(rec["name"]), []).append(rec) # 2. Everything Infrahub has, from the bulk metric snapshot. self.progress = "comparing against Infrahub" infrahub = list(snapshot.by_openstack_id.values()) + list(snapshot.by_instance_name.values()) seen: set[str] = set() ih_records: list[dict[str, str]] = [] for row in infrahub: key = f"{row.get('openstack_id','')}|{row.get('instance_name','')}" if key in seen: continue seen.add(key) ih_records.append(row) ih_osids = {str(r.get("openstack_id") or "") for r in ih_records if r.get("openstack_id")} ih_osids.discard("") ih_osids.discard("None") # 3a. Infrahub records whose OpenStack server is missing or never linked. # # Only records in a region that was actually listed can be judged: if a # region failed, every VM in it would look "missing from OpenStack". scanned = set(region_counts) skipped_unscanned = 0 broken_links: list[dict[str, Any]] = [] for row in ih_records: status = str(row.get("status") or "").upper() if status not in UNLINKED_SUSPECT_STATES: continue if _region_alias(str(row.get("region") or "")) not in scanned: skipped_unscanned += 1 continue osid = str(row.get("openstack_id") or "") has_link = bool(osid) and osid != "None" if has_link and osid in os_by_id: continue # properly linked, nothing to see name = str(row.get("instance_name") or "") if _is_ignorable(name): continue candidates = os_by_name.get(_norm(name), []) # An exact-name server that nothing else claims is a very strong # candidate for the link that was never written. unclaimed = [c for c in candidates if c["id"] not in ih_osids] match = unclaimed[0] if unclaimed else (candidates[0] if candidates else None) broken_links.append({ "instance_name": name, "infrahub_status": status, "infrahub_openstack_id": osid or "(none)", "organization": str(row.get("organization") or ""), "region": str(row.get("region") or ""), "flavor": str(row.get("flavor_name") or ""), "gpus": str(row.get("_gpus") or ""), "reason": ( "Infrahub holds an OpenStack ID that OpenStack does not have" if has_link else "Infrahub never recorded an OpenStack ID" ), "candidate": match, "candidate_claimed_by_other": bool(match and match["id"] in ih_osids), "confidence": ( "high" if match and not match["id"] in ih_osids and match["status"] not in ("", "ERROR") else "medium" if match else "none" ), }) # 3b. OpenStack servers no Infrahub record claims. orphans: list[dict[str, Any]] = [] ih_names = {_norm(str(r.get("instance_name") or "")) for r in ih_records} for sid, rec in os_by_id.items(): if sid in ih_osids or _is_ignorable(rec["name"]): continue orphans.append({**rec, "name_known_to_infrahub": _norm(rec["name"]) in ih_names}) linkable = [b for b in broken_links if b["candidate"] and not b["candidate_claimed_by_other"]] return { "scanned_regions": region_counts, "region_failures": failures, "openstack_servers": len(os_by_id), "infrahub_records": len(ih_records), "broken_links": sorted(broken_links, key=lambda b: (b["confidence"] != "high", b["instance_name"])), "likely_linkage_failures": len(linkable), "skipped_unscanned_regions": skipped_unscanned, "orphans": sorted(orphans, key=lambda o: (not o["name_known_to_infrahub"], o["name"]))[:400], "orphan_total": len(orphans), } def _server_list(region: str) -> tuple[bool, list[dict[str, Any]], str]: ok, data, raw = cxbridge.os_json( region, ["server", "list", "--all-projects", "--long", "-f", "json"], timeout=180 ) if ok and isinstance(data, list): return True, [x for x in data if isinstance(x, dict)], "" return False, [], str(raw) def enrich(region: str, openstack_id: str) -> dict[str, Any]: """Fetch created time and fault for one candidate, on demand.""" c = cxbridge.cx() ok, srv, raw = c.server_show(cxbridge.config(), region, openstack_id) if not ok: return {"ok": False, "error": str(raw)[:200]} fault = srv.get("fault") return { "ok": True, "created": str(srv.get("created") or srv.get("Created") or ""), "launched": str(srv.get("OS-SRV-USG:launched_at") or ""), "status": str(srv.get("status") or ""), "host": str(srv.get("OS-EXT-SRV-ATTR:host") or ""), "project_id": str(srv.get("project_id") or ""), "fault": (fault.get("message") if isinstance(fault, dict) else str(fault or "")) or "None", }