#!/usr/bin/env python3 """ One-time CMDB → Helix data migration script. Usage: cd server python scripts/migrate_cmdb.py --phase preflight python scripts/migrate_cmdb.py --phase all python scripts/migrate_cmdb.py --phase domains python scripts/migrate_cmdb.py --phase servers python scripts/migrate_cmdb.py --phase dns python scripts/migrate_cmdb.py --phase load_balancers python scripts/migrate_cmdb.py --phase cdn python scripts/migrate_cmdb.py --phase backfill_lb_owners python scripts/migrate_cmdb.py --phase validate Requirements: - Cloud SQL Proxy running: cloud-sql-proxy --gcloud-auth --address 127.0.0.1 --port 3307 gcp-digital-infra:asia-east2:pch-sql-db - .env loaded (DATABASE_URL_SYNC for target) - alembic head ≥ 173 (the migration that adds the modifier + business columns this script writes). Preflight checks the head and aborts if the target schema is older. Re-run safety (2026-05-28): This script's ``ON DUPLICATE KEY UPDATE`` clauses do NOT clear ``deleted_at`` — a re-run on a row that was legitimately soft-deleted (admin decommission, real sync sweep of a row removed cloud-side) MUST preserve the tombstone. The 2026-04-24 mass-sweep rescue is handled exclusively by alembic migration 212 (one-shot, reviewable). If a future operator needs to bulk-resurrect another cohort, write another alembic migration scoped to the affected rows; do NOT add ``deleted_at=NULL`` here — it would silently un-tombstone every legitimately-decommissioned row on the next re-run. Owner reassignment chain (migration 173): legacy.owner (str) → empty → owner_person_id = NULL (no fallback claim) email lookup hits → person row if active and not Max Liu → use that person if departed/leaver_pending/left → walk manager_person_id upward (depth ≤ 10) until first active non-Max-Liu lead → use them any step hits Max Liu (liuqi@pingcap.com) → fall back to 孟凡生 chain breaks (NULL manager / all departed) → fall back to 孟凡生 not found in user table → fall back to 孟凡生 Fallback owner: fansheng.meng@pingcap.com (孟凡生 Jason Meng). """ import argparse import json import logging import os import re import sys import uuid from datetime import datetime from pathlib import Path import pymysql import pymysql.cursors from dotenv import load_dotenv load_dotenv(Path(__file__).parent.parent / ".env") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("migrate_cmdb") # ── Source CMDB (via Cloud SQL Proxy) ────────────────────────────────────── SRC_HOST = "127.0.0.1" SRC_PORT = 3307 SRC_USER = "cmdb" SRC_PASS = os.environ["CMDB_DB_PASSWORD"] # set in .env before running SRC_DB = "cmdb" # ── Target Helix (TiDB Cloud) ────────────────────────────────────────────── _db_url = os.environ["DATABASE_URL_SYNC"] _m = re.match(r"mysql\+pymysql://([^:]+):([^@]+)@([^:]+):(\d+)/(\w+)", _db_url) DST_USER, DST_PASS, DST_HOST, DST_PORT, DST_DB = _m.groups() DST_PORT = int(DST_PORT) # ── Owner reassignment policy (migration 173) ────────────────────────────── # Resolved once at startup against the LIVE identity_persons table; the # script aborts if either email is missing so we never silently leave # owner_person_id at NULL for the entire fallback chain. FALLBACK_OWNER_EMAIL = "fansheng.meng@pingcap.com" # 孟凡生 Jason Meng # Max Liu (刘奇) is the CEO; he should NEVER be picked as a reassignment # target because resources accumulate on him from the manager chain and # he is too senior to actually own day-to-day infra. Block by ID at # runtime (his actual work_email is liuqi@pingcap.com, NOT max.liu@). BLOCKED_LEAD_EMAIL = "liuqi@pingcap.com" # Max Liu / 刘奇 # How many manager hops to walk before giving up and going to fallback. # 10 is far more than any realistic org-tree depth (the prod tree is # 5-6 levels including the CEO). Bound it explicitly to prevent infinite # loops if the manager_person_id graph develops a cycle. MAX_MANAGER_HOPS = 10 # ── Destination integration slugs (queried at runtime) ───────────────────── # Migration 173 fixes a longstanding bug in the previous script where # every legacy row landed under UCloud catch-all — including DNS # records (which belong under Tencent DNSPod) and Aliyun domains. # Mapping table indexed by legacy ``vendor_id``: LEGACY_VENDOR_SLUGS = { "1": "ksyun-legacy", # 金山云 — server / LB only "2": "ucloud-primary", # UCloud — server / LB / DNS / CDN / OSS # ``3`` was AWS in legacy CMDB but we have prod AWS-primary; legacy # rows likely none. Leave unmapped → catch-all (legacy-bare-metal). "4": "legacy-bare-metal", # Qiniu CDN — UCloud's CDN driver returns # UCDN-only domains, so Qiniu rows parked # under UCloud would be swept every sync. # Park under bare-metal (enabled=false, # never iterated) instead. "5": "tencent-primary", # Tencent DNSPod — DNS records belong here "6": "aliyun-primary", # Aliyun — domain registration "9": "legacy-bare-metal", # GoDaddy — no driver; bare-metal is # the explicit no-sweep parking lot. # (Was UCloud catch-all + accidentally # safe because UCloud doesn't support # DOMAINS; explicit is better.) "10": "legacy-bare-metal", # Namecheap — same as GoDaddy. } def src_conn(): return pymysql.connect( host=SRC_HOST, port=SRC_PORT, user=SRC_USER, password=SRC_PASS, database=SRC_DB, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=False, ) def dst_conn(): return pymysql.connect( host=DST_HOST, port=DST_PORT, user=DST_USER, password=DST_PASS, database=DST_DB, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=False, ssl={"fake_flag_to_enable_tls": True}, ) # ── Pre-flight ────────────────────────────────────────────────────────────── def _check_alembic_head(dst) -> bool: """Verify the target DB is on migration 173 (the migration that added ``last_modified_by_person_id``, ``owner_person_id`` on LB, and the legacy business columns this script writes).""" with dst.cursor() as c: c.execute("SELECT version_num FROM alembic_version") row = c.fetchone() head = (row or {}).get("version_num") try: head_int = int(head) if head else 0 except ValueError: head_int = 0 if head_int < 173: log.error( "Target alembic head is %s; need ≥ 173 (the migration that " "adds modifier + LB owner FK + legacy business columns). " "Run `alembic upgrade head` before this script.", head, ) return False log.info("Target alembic head: %s ✓", head) return True def _check_fallback_owner(dst) -> tuple[str, str] | None: """Resolve the fallback + blocked-lead person ids from work_email. Aborts the whole run if either is missing — without those we can't apply rule 5 of the user's brief and would silently leave records NULL-owned.""" with dst.cursor() as c: c.execute( "SELECT id, work_email, display_name, employment_status " "FROM identity_persons " "WHERE work_email IN (%s, %s) AND deleted_at IS NULL", (FALLBACK_OWNER_EMAIL, BLOCKED_LEAD_EMAIL), ) rows = {r["work_email"].lower(): r for r in c.fetchall()} fallback = rows.get(FALLBACK_OWNER_EMAIL.lower()) blocked = rows.get(BLOCKED_LEAD_EMAIL.lower()) if not fallback: log.error( "Fallback owner %s NOT found in identity_persons. Cannot " "apply owner reassignment policy — aborting. (Sync HRIS " "first or update FALLBACK_OWNER_EMAIL.)", FALLBACK_OWNER_EMAIL, ) return None if not blocked: log.error( "Blocked lead %s NOT found in identity_persons. Without " "this we can't enforce the 'don't pick the CEO' rule — " "aborting. (Update BLOCKED_LEAD_EMAIL if Max Liu's email " "has changed.)", BLOCKED_LEAD_EMAIL, ) return None log.info( " Fallback owner: %s (%s, %s)", fallback["id"], fallback["display_name"], fallback["employment_status"], ) log.info( " Blocked lead: %s (%s, %s)", blocked["id"], blocked["display_name"], blocked["employment_status"], ) return fallback["id"], blocked["id"] def preflight(src, dst) -> bool: log.info("=== PREFLIGHT CHECKS ===") ok = True if not _check_alembic_head(dst): ok = False if _check_fallback_owner(dst) is None: ok = False with src.cursor() as c: c.execute("SHOW TABLES") src_tables = {list(r.values())[0] for r in c.fetchall()} for t in ("domain", "server", "user"): if t not in src_tables: log.error(f"Source table missing: {t}") ok = False for t in ("domain", "server", "dns_record", "load_balancer", "cdn"): if t in src_tables: c.execute(f"SELECT COUNT(*) as n FROM `{t}`") log.info(f" src.{t}: {c.fetchone()['n']} rows") with dst.cursor() as c: for t in ("integrations", "cmdb_domains", "cmdb_servers", "cmdb_user_mapping", "cmdb_dns_records", "cmdb_load_balancers", "cmdb_cdn_domains"): c.execute( "SELECT COUNT(*) FROM information_schema.tables " "WHERE table_schema=DATABASE() AND table_name=%s", (t,) ) if c.fetchone()["COUNT(*)"] == 0: log.error(f"Target table missing: {t}") ok = False # Owner coverage (raw look-up only; manager-walk runs at apply time). with src.cursor() as c: c.execute("SELECT DISTINCT owner FROM domain WHERE owner != '' AND owner IS NOT NULL") domain_owners = {r["owner"] for r in c.fetchall()} c.execute("SELECT DISTINCT owner FROM server WHERE owner != '' AND owner IS NOT NULL") all_owners = domain_owners | {r["owner"] for r in c.fetchall()} try: c.execute("SELECT DISTINCT owner FROM dns_record WHERE owner != '' AND owner IS NOT NULL") all_owners |= {r["owner"] for r in c.fetchall()} except Exception: pass try: c.execute("SELECT DISTINCT owner FROM load_balancer WHERE owner != '' AND owner IS NOT NULL") all_owners |= {r["owner"] for r in c.fetchall()} except Exception: pass try: c.execute("SELECT DISTINCT owner FROM cdn WHERE owner != '' AND owner IS NOT NULL") all_owners |= {r["owner"] for r in c.fetchall()} except Exception: pass c.execute("SELECT user_name, email FROM user") cmdb_users = {r["user_name"]: (r["email"] or "").lower() for r in c.fetchall()} with dst.cursor() as c: c.execute("SELECT work_email FROM identity_persons WHERE deleted_at IS NULL") helix_emails = {r["work_email"].lower() for r in c.fetchall()} matched = sum( 1 for o in all_owners if cmdb_users.get(o, "") in helix_emails and cmdb_users.get(o, "") != "" ) coverage = matched / len(all_owners) * 100 if all_owners else 100 log.info( f"Owner direct-match coverage: {coverage:.1f}% ({matched}/{len(all_owners)}). " "Unmatched names will walk the manager chain or fall back to 孟凡生.", ) return ok # ── Phase 1: Resolve destination integrations ────────────────────────────── BARE_METAL_SLUG = "legacy-bare-metal" def resolve_destination_integrations(dst) -> dict[str, str]: """Return ``{vendor_id: integration_id}`` for every legacy vendor we know how to land, plus the special ``legacy-bare-metal`` slot (keyed by :data:`BARE_METAL_SLUG`) where rows without a cloud-issued instance_id are parked. Aborts if a required integration slug is missing. The ``legacy-bare-metal`` integration is seeded by migration 212 as ``kind=cloud`` + ``enabled=false`` + no driver — the cloud-sync cron skips disabled integrations, so rows parked there are never visited by ``_safe_sweep`` and stay alive forever (the safe home for manually-entered IDC / bare-metal servers that no driver can see). """ log.info("=== Resolving destination integrations ===") needed_slugs = set(LEGACY_VENDOR_SLUGS.values()) | {BARE_METAL_SLUG} with dst.cursor() as c: c.execute( "SELECT id, slug FROM integrations " "WHERE kind='cloud' AND slug IN ({})".format( ",".join(["%s"] * len(needed_slugs)) ), tuple(needed_slugs), ) slug_to_id = {r["slug"]: r["id"] for r in c.fetchall()} missing = needed_slugs - set(slug_to_id) if missing: raise RuntimeError( f"Missing integration rows for slugs {sorted(missing)}. " "Add them via /admin/integrations or alembic before retrying.", ) vendor_to_int = { vendor: slug_to_id[slug] for vendor, slug in LEGACY_VENDOR_SLUGS.items() } vendor_to_int[BARE_METAL_SLUG] = slug_to_id[BARE_METAL_SLUG] for vendor, int_id in sorted(vendor_to_int.items()): log.info(" vendor_id=%s → integration_id=%s", vendor, int_id) return vendor_to_int def _integration_for_vendor( vendor_id: str | int | None, vendor_to_int: dict[str, str], ) -> str: """Resolve the destination integration_id for a legacy vendor. Unknown vendors fall through to ``legacy-bare-metal`` (rather than the prior UCloud catch-all) — UCloud has live sync that would sweep rows whose instance_id format it doesn't recognize, so an unknown vendor parked there silently soft-deletes itself within a day. Bare-metal is ``enabled=false`` and never iterated. """ v = str(vendor_id) if vendor_id is not None else "" return vendor_to_int.get(v, vendor_to_int[BARE_METAL_SLUG]) def _bare_metal_integration_id(vendor_to_int: dict[str, str]) -> str: """Resolve the ``legacy-bare-metal`` integration ID — the dedicated parking lot for rows whose cloud-issued ``instance_id`` is empty (no live driver discovers them, so they must not live under any active integration's sweep scope). Seeded by migration 212. """ return vendor_to_int[BARE_METAL_SLUG] # ── Phase 2: User mapping (with manager fallback) ────────────────────────── class OwnerResolver: """Resolves legacy CMDB owner usernames to Helix person ids with the rule-5 fallback chain baked in. Reads the entire identity_persons table once at construction so per-row lookups are constant-time.""" def __init__( self, cmdb_users: dict[str, str], # legacy user_name → email person_by_id: dict[str, dict], # helix id → person row person_by_email: dict[str, str], # lowercased email → helix id fallback_person_id: str, blocked_lead_id: str, ) -> None: self.cmdb_users = cmdb_users self.person_by_id = person_by_id self.person_by_email = person_by_email self.fallback = fallback_person_id self.blocked_lead = blocked_lead_id self._cache: dict[str, str | None] = {} def resolve(self, legacy_owner: str | None) -> str | None: """Apply rule 5 of migration 173's owner-reassignment policy. Returns the person id to write into ``owner_person_id``. Empty legacy owner → None (no claim, no fallback).""" if not legacy_owner: return None key = legacy_owner.strip().lower() if not key: return None if key in self._cache: return self._cache[key] result = self._resolve_uncached(legacy_owner.strip()) self._cache[key] = result return result def _is_active(self, person: dict) -> bool: return (person.get("employment_status") or "").lower() == "active" def _is_blocked(self, person_id: str) -> bool: return person_id == self.blocked_lead def _resolve_uncached(self, legacy_owner: str) -> str: # Step 1: legacy user_name → email → identity_persons. email = (self.cmdb_users.get(legacy_owner) or "").strip().lower() if not email: return self.fallback # not in cmdb.user table helix_id = self.person_by_email.get(email) if not helix_id: return self.fallback # email never landed in HRIS person = self.person_by_id.get(helix_id) if person is None: return self.fallback # Step 2: active → use them, even if they ARE the CEO (Max Liu). # The user's brief blocks Max Liu only when we're *looking for a # substitute lead* (step 3); if a resource was directly assigned # to him, he stays the owner. Without this carve-out, every # legacy row where Max happens to be the explicit owner of a # personal sandbox / shared dev box would silently re-route to # 孟凡生 — an unwanted change of record. if self._is_active(person): return helix_id # Step 3: walk the manager chain. seen: set[str] = {helix_id} current = person for _ in range(MAX_MANAGER_HOPS): mgr_id = current.get("manager_person_id") if not mgr_id or mgr_id in seen: break seen.add(mgr_id) mgr = self.person_by_id.get(mgr_id) if mgr is None: break if self._is_blocked(mgr_id): # Encountered Max Liu in the chain → fallback to 孟凡生. # (Step 2 already returned if Max was the direct owner.) return self.fallback if self._is_active(mgr): return mgr_id current = mgr # Step 4: chain broke / everyone above is departed → fallback. return self.fallback def build_user_mapping(src, dst) -> OwnerResolver: log.info("=== PHASE: User mapping ===") with src.cursor() as c: c.execute("SELECT user_name, email, is_superuser FROM user") cmdb_users_rows = c.fetchall() with dst.cursor() as c: c.execute( "SELECT id, work_email, display_name, employment_status, " "manager_person_id FROM identity_persons WHERE deleted_at IS NULL" ) helix_people = c.fetchall() person_by_id = {p["id"]: p for p in helix_people} person_by_email = { (p["work_email"] or "").lower(): p["id"] for p in helix_people if p["work_email"] } fallback_id = person_by_email[FALLBACK_OWNER_EMAIL.lower()] blocked_id = person_by_email[BLOCKED_LEAD_EMAIL.lower()] cmdb_users: dict[str, str] = {} rows = [] for u in cmdb_users_rows: username = u["user_name"] or "" email = (u["email"] or "").strip().lower() cmdb_users[username] = email # Compute the eventual match status for the audit table — does the # username resolve to a Helix person via email (before fallback)? direct_helix_id = person_by_email.get(email) if email else None match_status = ( "matched" if direct_helix_id else ("no_email" if not email else "not_in_helix") ) rows.append({ "id": str(uuid.uuid4()), "cmdb_username": username, "cmdb_email": u["email"], "cmdb_is_superuser": bool(u["is_superuser"]), "helix_person_id": direct_helix_id, "match_status": match_status, }) with dst.cursor() as c: for r in rows: c.execute(""" INSERT INTO cmdb_user_mapping (id, cmdb_username, cmdb_email, cmdb_is_superuser, helix_person_id, match_status) VALUES (%(id)s, %(cmdb_username)s, %(cmdb_email)s, %(cmdb_is_superuser)s, %(helix_person_id)s, %(match_status)s) ON DUPLICATE KEY UPDATE helix_person_id=VALUES(helix_person_id), match_status=VALUES(match_status) """, r) dst.commit() resolver = OwnerResolver( cmdb_users=cmdb_users, person_by_id=person_by_id, person_by_email=person_by_email, fallback_person_id=fallback_id, blocked_lead_id=blocked_id, ) # Sanity: dry-run the resolver across every legacy owner so the operator # sees the fallback distribution before any data lands. distinct_owners: set[str] = set() with src.cursor() as c: for t in ("domain", "server", "dns_record", "load_balancer", "cdn"): try: c.execute( f"SELECT DISTINCT owner FROM `{t}` " "WHERE owner IS NOT NULL AND owner != ''" ) distinct_owners |= {r["owner"] for r in c.fetchall()} except Exception: pass direct = 0 chain = 0 fallback = 0 for o in distinct_owners: rid = resolver.resolve(o) if rid is None: continue if rid == fallback_id: email = (cmdb_users.get(o.strip()) or "").lower() if email == FALLBACK_OWNER_EMAIL.lower(): direct += 1 # owner literally IS the fallback person else: fallback += 1 else: email = (cmdb_users.get(o.strip()) or "").lower() direct_id = person_by_email.get(email) if direct_id == rid: direct += 1 else: chain += 1 log.info( "Owner resolution dry-run on %d distinct legacy owners: " "direct=%d, manager-chain=%d, fallback-to-孟凡生=%d", len(distinct_owners), direct, chain, fallback, ) return resolver # ── Phase 3a: Migrate domains ─────────────────────────────────────────────── def migrate_domains(src, dst, resolver: OwnerResolver, vendor_to_int: dict[str, str]): log.info("=== PHASE: Migrate domains ===") with src.cursor() as c: c.execute("SELECT * FROM domain") rows = c.fetchall() count = 0 with dst.cursor() as c: for r in rows: owner_str = (r.get("owner") or "").strip() expiry = r.get("expired_time") or r.get("expiry_date") if isinstance(expiry, str): try: expiry = datetime.strptime(expiry, "%Y-%m-%d %H:%M:%S").date() except ValueError: expiry = None # Use the proper registrar integration when possible (Aliyun=6, # GoDaddy/Namecheap fall through to catch-all UCloud). target = _integration_for_vendor(r.get("vendor_id"), vendor_to_int) resolved_owner = resolver.resolve(owner_str) # COALESCE-on-upgrade preserves any manual edits made in Helix # between migration runs. ``cmdb_domains`` does NOT carry an # ``owner_raw`` column (dropped in migration 042) so we write # only ``owner_person_id``. c.execute(""" INSERT INTO cmdb_domains (id, integration_id, domain_name, registrar, expiry_date, icp_filing_no, status, owner_person_id, possessor, remark, grade, raw_data, first_seen_at, last_synced_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) ON DUPLICATE KEY UPDATE registrar=VALUES(registrar), expiry_date=VALUES(expiry_date), icp_filing_no=VALUES(icp_filing_no), status=VALUES(status), owner_person_id=COALESCE(cmdb_domains.owner_person_id, VALUES(owner_person_id)), possessor=COALESCE(cmdb_domains.possessor, VALUES(possessor)), grade=COALESCE(cmdb_domains.grade, VALUES(grade)), remark=COALESCE(cmdb_domains.remark, VALUES(remark)), raw_data=VALUES(raw_data), last_synced_at=NOW() """, ( str(uuid.uuid4()), target, r.get("name", ""), r.get("possessor") or "legacy", expiry, r.get("icp_code"), "active" if r.get("status", "enable") == "enable" else "inactive", resolved_owner, r.get("possessor"), r.get("remark"), r.get("grade"), json.dumps({k: str(v) for k, v in r.items() if v is not None}, default=str), )) count += 1 dst.commit() log.info(f" Upserted {count} domains") # ── Phase 3b: Migrate servers ─────────────────────────────────────────────── def migrate_servers(src, dst, resolver: OwnerResolver, vendor_to_int: dict[str, str]): log.info("=== PHASE: Migrate servers ===") with src.cursor() as c: c.execute("SELECT * FROM server WHERE status != 'Deleted'") rows = c.fetchall() counts: dict[str, int] = {} with dst.cursor() as c: for r in rows: owner_str = (r.get("owner") or "").strip() public_ips = [ip for ip in [r.get("public_ip")] if ip] private_ips = [ip for ip in [r.get("private_ip")] if ip] # Use the raw cloud-issued instance_id directly so the live # provider sync's upsert collides with this row on the # ``(integration_id, instance_id)`` unique key and UPDATES it # instead of inserting a shadow row + soft-deleting this one # as a sweep orphan (prod incident 2026-04-24: 116 UCloud + # ~10 000 GCP/Aliyun rows wiped because the legacy script # synthesized ``legacy-{vendor}-{src_id}-{raw_iid}`` which # never matches what any provider driver returns). # # ``(integration_id, instance_id)`` is the unique key; rows # from different vendors land under different integrations # via ``vendor_to_int`` so cross-vendor duplicates of the # same instance_id literal (rare but possible) are isolated # by integration_id. # # Bare-metal / IDC rows have no cloud-issued instance_id — # for those we synthesize a deterministic ``legacy-bare-...`` # key and the alembic-seeded ``legacy-bare-metal`` # integration (kind=cloud, enabled=false, no driver) holds # them so the cloud-sync cron never iterates over them. raw_iid = (r.get("instance_id") or "").strip() if raw_iid: unique_iid = raw_iid[:512] target = _integration_for_vendor(r.get("vendor_id"), vendor_to_int) else: # Bare-metal / IDC: no cloud-issued instance_id. Park # under the legacy-bare-metal integration so no live # sync ever sees these rows + ``_safe_sweep`` can't # reach them. ``cmdb_servers.instance_id`` is # ``VARCHAR(512)``; cap defensively even though # ``legacy-bare-{vendor}-{src_id}`` is short in practice. unique_iid = ( f"legacy-bare-{r.get('vendor_id')}-{r.get('id')}"[:512] ) target = _bare_metal_integration_id(vendor_to_int) counts[target] = counts.get(target, 0) + 1 resolved_owner = resolver.resolve(owner_str) # IDC / bare-metal fields (sn, ipmi_*, cabinet_*, rack_*, # exchanger_*, model_*) go into raw_data JSON, not as schema # columns — see migration 173 docstring. # # SECURITY: ``ipmi_ip`` + ``ipmi_user`` here are BMC # management endpoints + admin usernames for bare-metal # servers. The legacy CMDB has no ``ipmi_password`` column # (verified absent — passwords are in a separate secrets # store not migrated here), so this dict carries only the # endpoint identifiers, not credentials. STILL: any future # response schema that surfaces ``raw_data._idc`` to a # non-IT-admin caller leaks the BMC fleet topology — DO # NOT add this path to any employee-readable list/detail # endpoint. The recycle-bin response (added 2026-05-28) # uses ``ServerResponse`` which excludes ``raw_data``; # keep that invariant. idc_extra = { k: r.get(k) for k in ( "sn", "ipmi_ip", "ipmi_user", "cabinet_name", "rack_name", "exchanger_name", "model_name", "model_id", ) if r.get(k) } raw_data = {k: str(v) for k, v in r.items() if v is not None} raw_data["_idc"] = idc_extra memory_mb = (r.get("memory") or 0) * 1024 if r.get("memory") else None c.execute(""" INSERT INTO cmdb_servers (id, integration_id, instance_id, name, status, region, zone, vcpu, memory_mb, public_ips, private_ips, owner_person_id, app, remark, instance_type, image_name, vpc_id, subnet_id, project_id, system_disk_gb, data_disk_gb, bandwidth_mbps, expired, raw_data, first_seen_at, last_synced_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) ON DUPLICATE KEY UPDATE name=VALUES(name), status=VALUES(status), region=VALUES(region), zone=VALUES(zone), vcpu=VALUES(vcpu), memory_mb=VALUES(memory_mb), public_ips=VALUES(public_ips), private_ips=VALUES(private_ips), owner_person_id=COALESCE(cmdb_servers.owner_person_id, VALUES(owner_person_id)), app=COALESCE(cmdb_servers.app, VALUES(app)), remark=COALESCE(cmdb_servers.remark, VALUES(remark)), instance_type=COALESCE(cmdb_servers.instance_type, VALUES(instance_type)), image_name=COALESCE(cmdb_servers.image_name, VALUES(image_name)), vpc_id=COALESCE(cmdb_servers.vpc_id, VALUES(vpc_id)), subnet_id=COALESCE(cmdb_servers.subnet_id, VALUES(subnet_id)), project_id=COALESCE(cmdb_servers.project_id, VALUES(project_id)), system_disk_gb=COALESCE(cmdb_servers.system_disk_gb, VALUES(system_disk_gb)), data_disk_gb=COALESCE(cmdb_servers.data_disk_gb, VALUES(data_disk_gb)), bandwidth_mbps=COALESCE(cmdb_servers.bandwidth_mbps, VALUES(bandwidth_mbps)), raw_data=VALUES(raw_data), last_synced_at=NOW() """, ( str(uuid.uuid4()), target, unique_iid, r.get("name", ""), {"running": "running", "stopped": "stopped"}.get( (r.get("status") or "").lower(), (r.get("status") or "unknown").lower() ), r.get("region_id") or r.get("region"), r.get("zone_id") or r.get("zone"), r.get("cpu"), memory_mb, json.dumps(public_ips), json.dumps(private_ips), resolved_owner, r.get("app"), r.get("remark"), r.get("instance_type"), r.get("image_name"), r.get("vpc_id"), r.get("subnet_id"), r.get("project_id"), r.get("system_disk"), r.get("data_disk"), r.get("band_with"), bool(r.get("expired")), json.dumps(raw_data, default=str), )) dst.commit() log.info(f" Upserted {sum(counts.values())} servers; per-integration: %s", counts) # ── Phase 3c: Migrate DNS records ────────────────────────────────────────── def migrate_dns(src, dst, resolver: OwnerResolver, vendor_to_int: dict[str, str]): log.info("=== PHASE: Migrate DNS records ===") with src.cursor() as c: c.execute(""" SELECT id AS src_id, record_id AS record_id, zone_name AS zone_name, vendor_id AS vendor_id, name AS name, type AS record_type, target AS value, ttl AS ttl, line AS line, status AS status, owner AS owner, remark AS remark, create_time AS create_time FROM dns_record """) rows = c.fetchall() counts: dict[str, int] = {} with dst.cursor() as c: for r in rows: owner_str = (r.get("owner") or "").strip() # Migration 173: route DNSPod records (vendor_id=5) under the # tencent-primary integration instead of UCloud catch-all. # GoDaddy/Namecheap still go to catch-all (no driver for them). raw_record_id = (str(r.get("record_id") or "").strip()) if raw_record_id: target = _integration_for_vendor(r.get("vendor_id"), vendor_to_int) record_id_value = raw_record_id[:128] else: # Legacy row has no cloud-issued record_id (very old # imports, manual entries). Park under legacy-bare-metal # so no live DNS driver discovers + sweeps. Synthesize # a deterministic key from src_id, capped at 128 chars # defensively (well under ``cmdb_dns_records.record_id`` # ``VARCHAR(400)`` in models.py:381 — the synthetic # shape is short in practice, the cap is belt-and- # braces). target = _bare_metal_integration_id(vendor_to_int) record_id_value = f"legacy-bare-{r['src_id']}"[:128] counts[target] = counts.get(target, 0) + 1 resolved_owner = resolver.resolve(owner_str) c.execute(""" INSERT INTO cmdb_dns_records (id, integration_id, zone_name, record_id, name, record_type, value, ttl, line, status, owner_person_id, remark, raw_data, first_seen_at, last_synced_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE name=VALUES(name), record_type=VALUES(record_type), value=VALUES(value), ttl=VALUES(ttl), line=VALUES(line), status=VALUES(status), owner_person_id=COALESCE(cmdb_dns_records.owner_person_id, VALUES(owner_person_id)), remark=COALESCE(cmdb_dns_records.remark, VALUES(remark)), raw_data=VALUES(raw_data), last_synced_at=NOW() """, ( str(uuid.uuid4()), target, r["zone_name"] or "", record_id_value, r["name"] or "@", (r["record_type"] or "A").upper(), r["value"] or "", r["ttl"] or 600, r["line"] or None, r["status"] or "enable", resolved_owner, r.get("remark"), json.dumps({k: str(v) for k, v in r.items() if v is not None}, default=str), r["create_time"] or datetime.now(), )) dst.commit() log.info(f" Upserted {sum(counts.values())} DNS records; per-integration: %s", counts) # ── Phase 3d: Migrate load balancers ─────────────────────────────────────── def migrate_load_balancers(src, dst, resolver: OwnerResolver, vendor_to_int: dict[str, str]): log.info("=== PHASE: Migrate load balancers ===") with src.cursor() as c: c.execute("SELECT * FROM load_balancer") rows = c.fetchall() counts: dict[str, int] = {} with dst.cursor() as c: for r in rows: owner_str = (r.get("owner") or "").strip() # Use raw cloud-issued lb_id so live driver upsert collides # cleanly on ``(integration_id, lb_id)`` instead of inserting # a shadow row + sweeping this one (same incident as servers). raw_lb_id = (str(r.get("lb_id") or "").strip()) if raw_lb_id: lb_id_value = raw_lb_id[:128] target = _integration_for_vendor(r.get("vendor_id"), vendor_to_int) else: # No cloud-issued ID — park under legacy-bare-metal so # no driver iterates over this row. Cap defensively at # 128 chars — well under ``cmdb_load_balancers.lb_id`` # ``VARCHAR(512)`` (see models.py:529). The synthetic # ``legacy-bare-{vendor}-{id}`` shape is always short # in practice; the cap is belt-and-braces against a # future vendor_id / src_id literal that surprises us. lb_id_value = ( f"legacy-bare-{r.get('vendor_id')}-{r.get('id')}"[:128] ) target = _bare_metal_integration_id(vendor_to_int) counts[target] = counts.get(target, 0) + 1 resolved_owner = resolver.resolve(owner_str) c.execute(""" INSERT INTO cmdb_load_balancers (id, integration_id, lb_id, name, lb_type, network_type, ip_address, region, status, owner_raw, owner_person_id, app, remark, tag, vpc_id, subnet_id, project_id, listener_count, raw_data, first_seen_at, last_synced_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE name=VALUES(name), lb_type=VALUES(lb_type), network_type=VALUES(network_type), ip_address=VALUES(ip_address), region=VALUES(region), status=VALUES(status), owner_raw=COALESCE(cmdb_load_balancers.owner_raw, VALUES(owner_raw)), owner_person_id=COALESCE(cmdb_load_balancers.owner_person_id, VALUES(owner_person_id)), app=COALESCE(cmdb_load_balancers.app, VALUES(app)), remark=COALESCE(cmdb_load_balancers.remark, VALUES(remark)), tag=COALESCE(cmdb_load_balancers.tag, VALUES(tag)), vpc_id=COALESCE(cmdb_load_balancers.vpc_id, VALUES(vpc_id)), subnet_id=COALESCE(cmdb_load_balancers.subnet_id, VALUES(subnet_id)), project_id=COALESCE(cmdb_load_balancers.project_id, VALUES(project_id)), listener_count=COALESCE(cmdb_load_balancers.listener_count, VALUES(listener_count)), raw_data=VALUES(raw_data), last_synced_at=NOW() """, ( str(uuid.uuid4()), target, lb_id_value, r.get("lb_name") or r.get("name") or "", "classic", # source has no lb_type; legacy Aliyun SLB = classic r.get("network_type") or "public", r.get("ip_address") or "", r.get("region_id") or r.get("region") or "", "active", owner_str or None, resolved_owner, r.get("app"), r.get("remark"), r.get("tag"), r.get("vpc_id"), r.get("subnet_id"), r.get("project_id"), r.get("listener_count"), json.dumps({k: str(v) for k, v in r.items() if v is not None}, default=str), r.get("create_time") or datetime.now(), )) dst.commit() log.info(f" Upserted {sum(counts.values())} load balancers; per-integration: %s", counts) # ── Phase 3e: Migrate CDN domains ─────────────────────────────────────────── def migrate_cdn(src, dst, resolver: OwnerResolver, vendor_to_int: dict[str, str]): """Legacy CDN rows are mostly Qiniu (vendor_id=4) which has no Helix integration — park them under UCloud catch-all. Migration 173 also threads ``owner_person_id`` + ``cert_id`` + ``remark`` so the audit trail survives import.""" log.info("=== PHASE: Migrate CDN domains ===") with src.cursor() as c: c.execute("SELECT * FROM cdn") rows = c.fetchall() count = 0 with dst.cursor() as c: for r in rows: owner_str = (r.get("owner") or "").strip() target = _integration_for_vendor(r.get("vendor_id"), vendor_to_int) resolved_owner = resolver.resolve(owner_str) c.execute(""" INSERT INTO cmdb_cdn_domains (id, integration_id, domain_name, cname, status, source_type, source_address, is_https, owner_raw, owner_person_id, cert_id, remark, raw_data, first_seen_at, last_synced_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE cname=VALUES(cname), status=VALUES(status), source_type=VALUES(source_type), source_address=VALUES(source_address), is_https=VALUES(is_https), owner_raw=COALESCE(cmdb_cdn_domains.owner_raw, VALUES(owner_raw)), owner_person_id=COALESCE(cmdb_cdn_domains.owner_person_id, VALUES(owner_person_id)), cert_id=COALESCE(cmdb_cdn_domains.cert_id, VALUES(cert_id)), remark=COALESCE(cmdb_cdn_domains.remark, VALUES(remark)), raw_data=VALUES(raw_data), last_synced_at=NOW() """, ( str(uuid.uuid4()), target, r.get("name") or "", r.get("cname") or "", r.get("status") or "online", r.get("source_type") or "domain", r.get("source_target") or "", bool(r.get("is_https") or r.get("id_https")), # Go JSON tag typo: id_https owner_str or None, resolved_owner, str(r["cert_id"]) if r.get("cert_id") else None, r.get("remark"), json.dumps({k: str(v) for k, v in r.items() if v is not None}, default=str), r.get("create_time") or datetime.now(), )) count += 1 dst.commit() log.info(f" Upserted {count} CDN domains") # ── Phase 4: Backfill LB owners from owner_raw on existing rows ──────────── def backfill_lb_owners(dst): """Migration 173 added ``owner_person_id`` to ``cmdb_load_balancers``. The alembic backfill matches ``owner_raw`` to ``work_email``; this phase additionally tries the OwnerResolver chain on rows whose raw value didn't directly match an email.""" log.info("=== PHASE: Backfill LB owner_person_id ===") with dst.cursor() as c: c.execute( "SELECT id, owner_raw FROM cmdb_load_balancers " "WHERE owner_raw IS NOT NULL AND owner_raw != '' " " AND owner_person_id IS NULL AND deleted_at IS NULL" ) rows = c.fetchall() if not rows: log.info(" No LB rows need owner backfill.") return # We can't reuse the legacy-CMDB OwnerResolver here because we don't # have a Cloud SQL Proxy session for this phase. Use a Helix-only # lookup chain: owner_raw → identity_persons by work_email or # display_name, no manager-chain walk (operator can run owner # transfers later). with dst.cursor() as c: c.execute( "SELECT id, work_email, display_name " "FROM identity_persons WHERE deleted_at IS NULL" ) people = c.fetchall() email_to_id = {(p["work_email"] or "").lower(): p["id"] for p in people if p["work_email"]} display_to_id: dict[str, str] = {} for p in people: if p["display_name"]: display_to_id.setdefault(p["display_name"], p["id"]) updated = 0 with dst.cursor() as c: for r in rows: raw = (r["owner_raw"] or "").strip().lower() pid = email_to_id.get(raw) or display_to_id.get(r["owner_raw"].strip()) if not pid: continue c.execute( "UPDATE cmdb_load_balancers SET owner_person_id=%s WHERE id=%s", (pid, r["id"]), ) updated += 1 dst.commit() log.info(f" Backfilled owner_person_id on {updated}/{len(rows)} LB rows") # ── Phase 5: Validate ────────────────────────────────────────────────────── def validate(src, dst): log.info("=== VALIDATION ===") with src.cursor() as c: c.execute("SELECT COUNT(*) as n FROM domain") src_domains = c.fetchone()["n"] c.execute("SELECT COUNT(*) as n FROM server WHERE status != 'Deleted'") src_servers = c.fetchone()["n"] c.execute("SELECT COUNT(*) as n FROM dns_record") src_dns = c.fetchone()["n"] try: c.execute("SELECT COUNT(*) as n FROM load_balancer") src_lbs = c.fetchone()["n"] except Exception: src_lbs = "N/A" try: c.execute("SELECT COUNT(*) as n FROM cdn") src_cdn = c.fetchone()["n"] except Exception: src_cdn = "N/A" with dst.cursor() as c: c.execute("SELECT COUNT(*) as n FROM cmdb_domains WHERE deleted_at IS NULL") dst_domains = c.fetchone()["n"] c.execute("SELECT COUNT(*) as n FROM cmdb_servers WHERE deleted_at IS NULL") dst_servers = c.fetchone()["n"] c.execute("SELECT COUNT(*) as n FROM cmdb_dns_records WHERE deleted_at IS NULL") dst_dns = c.fetchone()["n"] c.execute("SELECT COUNT(*) as n FROM cmdb_load_balancers WHERE deleted_at IS NULL") dst_lbs = c.fetchone()["n"] c.execute("SELECT COUNT(*) as n FROM cmdb_cdn_domains WHERE deleted_at IS NULL") dst_cdn = c.fetchone()["n"] # FK integrity (sample across the 5 tables). c.execute(""" SELECT COUNT(*) as n FROM cmdb_domains d LEFT JOIN identity_persons p ON d.owner_person_id = p.id WHERE d.owner_person_id IS NOT NULL AND p.id IS NULL """) fk_violations_d = c.fetchone()["n"] c.execute(""" SELECT COUNT(*) as n FROM cmdb_servers s LEFT JOIN identity_persons p ON s.owner_person_id = p.id WHERE s.owner_person_id IS NOT NULL AND p.id IS NULL """) fk_violations_s = c.fetchone()["n"] c.execute(""" SELECT COUNT(*) as n FROM cmdb_dns_records r LEFT JOIN identity_persons p ON r.owner_person_id = p.id WHERE r.owner_person_id IS NOT NULL AND p.id IS NULL """) fk_violations_r = c.fetchone()["n"] fk_violations = fk_violations_d + fk_violations_s + fk_violations_r # Owner resolution stats across servers / domains / dns / lb / cdn. c.execute(""" SELECT SUM(owner_person_id IS NOT NULL) as resolved, COUNT(*) as total FROM ( SELECT owner_person_id FROM cmdb_domains WHERE deleted_at IS NULL UNION ALL SELECT owner_person_id FROM cmdb_servers WHERE deleted_at IS NULL UNION ALL SELECT owner_person_id FROM cmdb_dns_records WHERE deleted_at IS NULL UNION ALL SELECT owner_person_id FROM cmdb_load_balancers WHERE deleted_at IS NULL UNION ALL SELECT owner_person_id FROM cmdb_cdn_domains WHERE deleted_at IS NULL ) t """) owner_stats = c.fetchone() log.info(f" Domains: src={src_domains} dst={dst_domains}") log.info(f" Servers: src={src_servers} dst={dst_servers}") log.info(f" DNS records: src={src_dns} dst={dst_dns}") log.info(f" Load balancers: src={src_lbs} dst={dst_lbs}") log.info(f" CDN domains: src={src_cdn} dst={dst_cdn}") log.info(f" FK violations: {fk_violations} (expected 0)") log.info( f" Owner resolution: " f"{owner_stats['resolved']}/{owner_stats['total']} rows have owner_person_id" ) if fk_violations > 0: log.error("FK violations found — investigate before cutover") return False return True # ── Main ──────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser() parser.add_argument( "--phase", choices=[ "preflight", "all", "domains", "servers", "dns", "load_balancers", "cdn", "backfill_lb_owners", "validate", ], default="preflight", ) args = parser.parse_args() # backfill_lb_owners only needs the target DB. if args.phase == "backfill_lb_owners": dst = dst_conn() try: backfill_lb_owners(dst) finally: dst.close() return src = src_conn() dst = dst_conn() try: if args.phase == "preflight": ok = preflight(src, dst) sys.exit(0 if ok else 1) if not preflight(src, dst): log.error("Preflight failed — fix issues before migrating") sys.exit(1) vendor_to_int = resolve_destination_integrations(dst) resolver = build_user_mapping(src, dst) if args.phase in ("all", "domains"): migrate_domains(src, dst, resolver, vendor_to_int) if args.phase in ("all", "servers"): migrate_servers(src, dst, resolver, vendor_to_int) if args.phase in ("all", "dns"): migrate_dns(src, dst, resolver, vendor_to_int) if args.phase in ("all", "load_balancers"): migrate_load_balancers(src, dst, resolver, vendor_to_int) if args.phase in ("all", "cdn"): migrate_cdn(src, dst, resolver, vendor_to_int) if args.phase in ("all", "validate"): ok = validate(src, dst) sys.exit(0 if ok else 1) finally: src.close() dst.close() if __name__ == "__main__": main()