diff --git a/backend/database/migrations.py b/backend/database/migrations.py index c235975..4f5ab07 100644 --- a/backend/database/migrations.py +++ b/backend/database/migrations.py @@ -1796,7 +1796,24 @@ async def ensure_agent_activity_logs_table(): # Additive + idempotent; no existing table is altered, agents never read this # table. Same caveat as v1.10.4: this bump re-seeds the four built-in roles to # their defaults, so export role customizations before upgrading. -SCHEMA_VERSION = 12 +# +# v1.12.0 (issue #73 — several VIPs per node, several addresses per VIP): bumped +# 12 -> 13 as a DELTA. A database already at 12 runs only the steps listed in +# _delta_steps() — two additive columns — and keeps its role customizations; the full +# sequence (and with it the built-in role re-seed) runs only for databases older than +# _DELTA_BASE_VERSION or without a marker, exactly as before. Every delta step is ALSO +# called from _run_all_migrations_inner(), so both paths end at the same schema. +# A future change that alters seeded/role data needs a full bump: raise +# _DELTA_BASE_VERSION to the new SCHEMA_VERSION instead of adding a delta step. +SCHEMA_VERSION = 13 +_DELTA_BASE_VERSION = 12 + + +def _delta_steps(): + """(version, step) pairs a database at >= _DELTA_BASE_VERSION runs instead of the full + sequence. Each step takes the migration connection and raises on failure, so the version + marker is not written and the next start retries.""" + return [(13, ensure_vip_multi_address_columns)] async def run_all_migrations(): @@ -1845,7 +1862,13 @@ async def run_all_migrations(): logger.info(f"Schema already at version {applied_version} (>= {SCHEMA_VERSION}); skipping migration run.") return - await _run_all_migrations_inner() + if applied_version is not None and applied_version >= _DELTA_BASE_VERSION: + for version, step in _delta_steps(): + if version > applied_version: + logger.info(f"Applying schema delta {applied_version} -> {version}") + await step(lock_conn) + else: + await _run_all_migrations_inner() try: await lock_conn.execute(""" @@ -1935,6 +1958,7 @@ async def _run_all_migrations_inner(): # Issue #27 — HA/VIP Keepalived management (v1.7.0): two brand-new tables. # MUST run after its FK targets (haproxy_cluster_pools/agents/users), all created above. await ensure_vip_tables() + await ensure_vip_multi_address_columns() # v1.9.0 — CSR creation: brand-new ssl_csrs table. FK-references # ssl_certificates/users, both created above. @@ -2408,6 +2432,36 @@ async def ensure_vip_tables(): if conn: await close_database_connection(conn) + +async def ensure_vip_multi_address_columns(conn=None): + """v1.12.0 (issue #73). Additive + idempotent: + + * vip_instances.additional_ips — the VIP's addresses beyond virtual_ip, as a JSON list + of {"address", "prefix_length"}. Empty for every existing VIP, which renders exactly + as before. + * vip_discoveries.analysis_version — which parser produced the stored analysis. Rows + written by an older parser read 0, and the agent is asked to re-report them once. + + Raises on failure (unlike ensure_vip_tables): code reads these columns unconditionally, + so the schema version must not be recorded without them. + """ + own = conn is None + if own: + conn = await get_database_connection() + try: + # IF EXISTS: ensure_vip_tables() tolerates its own failure, and a missing VIP table + # must keep degrading only the VIP feature, not stop the backend from starting. + await conn.execute( + "ALTER TABLE IF EXISTS vip_instances ADD COLUMN IF NOT EXISTS additional_ips " + "JSONB NOT NULL DEFAULT '[]'::jsonb;") + await conn.execute( + "ALTER TABLE IF EXISTS vip_discoveries ADD COLUMN IF NOT EXISTS analysis_version " + "SMALLINT NOT NULL DEFAULT 0;") + logger.info("✅ VIP multi-address columns ensured (issue #73)") + finally: + if own: + await close_database_connection(conn) + async def add_ssl_certificate_id_to_backend_servers(): """Add ssl_certificate_id column to backend_servers table for SSL certificate management""" conn = None diff --git a/backend/routers/agent.py b/backend/routers/agent.py index 3f4b1e1..6990886 100644 --- a/backend/routers/agent.py +++ b/backend/routers/agent.py @@ -931,16 +931,17 @@ async def delete_agent(agent_id: int, authorization: str = Header(None)): # with no way to be told to tear down (review MED-3). Make the operator remove it # from the VIP first — that stages a clean PENDING change they can apply. try: - vip_member = await conn.fetchrow( + # A node may belong to several VIPs (issue #73); name them all. + vip_names = [r["name"] for r in await conn.fetch( "SELECT v.name FROM vip_members vm JOIN vip_instances v ON v.id = vm.vip_id " - "WHERE vm.agent_id = $1 AND v.is_active = TRUE LIMIT 1", agent_id) + "WHERE vm.agent_id = $1 AND v.is_active = TRUE ORDER BY v.name", agent_id)] except Exception: # noqa: BLE001 — vip_* may not exist on older schemas; don't block delete - vip_member = None - if vip_member: + vip_names = [] + if vip_names: await close_database_connection(conn) raise HTTPException( status_code=409, - detail=(f"This node is a member of VIP '{vip_member['name']}'. Remove it from the " + detail=(f"This node is a member of VIP '{', '.join(vip_names)}'. Remove it from the " f"VIP on the HA / VIP page (and apply) before deleting the agent.")) await conn.execute("DELETE FROM agents WHERE id = $1", agent_id) @@ -2281,6 +2282,37 @@ async def get_agent_ssl_certificates(agent_name: str, since: Optional[str] = Non raise HTTPException(status_code=500, detail=str(e)) +async def _node_applied_vips(conn, agent_id: int): + """Issue #73 — the active VIPs this node runs: every active membership that has an applied + snapshot, ordered by VIP id (the order compose_node_keepalived_conf writes them in).""" + return await conn.fetch(""" + SELECT v.id AS vip_id, v.name AS vip_name, v.is_active, v.track_haproxy, + v.purge_on_teardown, + m.applied_config_content, m.applied_config_hash, m.takeover_expected_hash + FROM vip_members m JOIN vip_instances v ON v.id = m.vip_id + WHERE m.agent_id = $1 AND v.is_active = TRUE + AND m.applied_config_content IS NOT NULL AND m.applied_config_content <> '' + ORDER BY v.id + """, agent_id) + + +async def _flag_compose_error(conn, agent_id: int, vip_ids, exc: Exception) -> None: + """Show a merge failure on the node's members. The agent never acks a status it does not + know, so without this the VIPs would sit at SYNCING with nothing saying why.""" + if not vip_ids: + return + msg = f"cannot merge this node's VIPs into one keepalived.conf ({exc}); re-apply the VIPs"[:500] + try: + await conn.execute(""" + UPDATE vip_members SET last_deploy_state='error', last_deploy_message=$3, + last_deploy_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP + WHERE agent_id=$1 AND vip_id = ANY($2::int[]) + AND (last_deploy_state IS DISTINCT FROM 'error' OR last_deploy_message IS DISTINCT FROM $4) + """, agent_id, list(vip_ids), msg, msg) + except Exception as e: # noqa: BLE001 + logger.error(f"could not record the keepalived compose failure: {e}") + + @router.get("/{agent_name}/keepalived-config") async def get_agent_keepalived_config(agent_name: str, x_api_key: Optional[str] = Header(None)): """Issue #27 (v1.7.0) — deliver this agent's APPLIED keepalived snapshot, or a @@ -2293,8 +2325,8 @@ async def get_agent_keepalived_config(agent_name: str, x_api_key: Optional[str] the agent API key is a SHARED/global install token, so the config is resolved by the requested agent_name and a token/name mismatch is an advisory audit log, NOT a 403 (a hard 403 would break every VIP member whose name isn't the one row the shared token - resolves to — review HIGH-1). Any unexpected error degrades to not_configured (B-7) so - the agent stays inert; a node with no membership row always gets not_configured. + resolves to — review HIGH-1). Any unexpected error degrades to a status the agent ignores + (B-7) so the agent stays inert; a node with no membership row always gets not_configured. """ conn = None try: @@ -2303,6 +2335,7 @@ async def get_agent_keepalived_config(agent_name: str, x_api_key: Optional[str] # name mismatch — do not 403 (HIGH-1). Done before the agent lookup so an # unauthenticated caller can't probe which agent names exist. from auth_middleware import validate_agent_api_key + from services.keepalived_parser import ANALYSIS_VERSION if not x_api_key: raise HTTPException(status_code=401, detail="Agent API key required") agent_auth = await validate_agent_api_key(x_api_key) @@ -2333,7 +2366,12 @@ async def get_agent_keepalived_config(agent_name: str, x_api_key: Optional[str] -- good: the file never changes, so the agent never speaks again. Telling it -- what we actually hold lets it recover on its own, with no extra request and -- no one having to touch the node. - EXISTS (SELECT 1 FROM vip_discoveries vd WHERE vd.agent_id = a.id) + -- + -- v1.12.0: only a discovery analysed by the CURRENT parser counts. An older + -- analysis reads as unknown, so the agent re-reports once and the stored + -- blockers/candidates catch up without anyone touching the node. + EXISTS (SELECT 1 FROM vip_discoveries vd + WHERE vd.agent_id = a.id AND vd.analysis_version >= $2) AS discovery_known FROM agents a LEFT JOIN haproxy_clusters hc ON hc.pool_id = a.pool_id @@ -2354,7 +2392,7 @@ async def get_agent_keepalived_config(agent_name: str, x_api_key: Optional[str] OR hc.keepalived_config_path = '/etc/keepalived/keepalived.conf'), hc.id LIMIT 1 - """, agent_name) + """, agent_name, ANALYSIS_VERSION) if not agent: raise HTTPException(status_code=404, detail=f"Agent '{agent_name}' not found") config_path = agent['keepalived_config_path'] or '/etc/keepalived/keepalived.conf' @@ -2367,13 +2405,57 @@ async def get_agent_keepalived_config(agent_name: str, x_api_key: Optional[str] m.applied_config_content, m.applied_config_hash, m.takeover_expected_hash FROM vip_members m JOIN vip_instances v ON v.id = m.vip_id WHERE m.agent_id = $1 - -- Active VIP first (an agent has at most one). With NO active VIP, pick the most - -- RECENTLY updated inactive membership so a teardown reflects the latest delete - -- (incl. its purge flag) — not a stale older VIP the node was once part of. + -- Active VIP first. With NO active VIP, pick the most RECENTLY updated inactive + -- membership so a teardown reflects the latest delete (incl. its purge flag) — + -- not a stale older VIP the node was once part of. ORDER BY v.is_active DESC, v.updated_at DESC, v.id DESC LIMIT 1 """, agent['id']) + # Issue #73: a node may belong to several active VIPs. Their applied snapshots are + # merged into the one keepalived.conf the node deploys; a single VIP is served exactly + # as before (same content, same hash). + # Any failure in this block must answer with a status the agent ignores: on a node + # whose file is ours the agent treats not_configured as "tear keepalived down". Here + # the members are also flagged, so the operator sees why the node is not converging. + node_vips = [] + merged = None + if row and row['is_active']: + try: + node_vips = await _node_applied_vips(conn, agent['id']) + if len(node_vips) > 1: + from services.keepalived_config import ( + build_haproxy_check_script, compose_node_keepalived_conf) + composed = compose_node_keepalived_conf( + [(r['vip_id'], r['applied_config_content']) for r in node_vips]) + takeover = next((r['takeover_expected_hash'] for r in node_vips + if r['takeover_expected_hash']), None) + merged = { + "desired_state": "enabled", + "install_if_missing": True, + "vip_id": node_vips[0]['vip_id'], + "vip_ids": [r['vip_id'] for r in node_vips], + "vip_name": ", ".join(str(r['vip_name']) for r in node_vips), + "config_content": composed, + "config_hash": hashlib.md5(composed.encode("utf-8")).hexdigest(), + "check_script": build_haproxy_check_script() + if any(r['track_haproxy'] for r in node_vips) else "", + "allow_takeover": bool(takeover), + "takeover_expected_hash": takeover, + } + except Exception as ce: # noqa: BLE001 + logger.error(f"keepalived compose failed for '{agent_name}': {ce}") + await _flag_compose_error(conn, agent['id'], [r['vip_id'] for r in node_vips], ce) + return {"agent_name": agent_name, "status": "compose_error", "config_path": config_path, + "discovery_known": bool(agent["discovery_known"]), "keepalived": None} + if merged is not None: + return {"agent_name": agent_name, "status": "available", "config_path": config_path, + "discovery_known": bool(agent["discovery_known"]), "keepalived": merged} + if len(node_vips) == 1: + # The most recently updated active VIP may be a pending one with nothing applied + # yet; serving it would read as not_configured and tear down the running VIP. + row = node_vips[0] + if not row: return {"agent_name": agent_name, "status": "not_configured", "config_path": config_path, "discovery_known": bool(agent["discovery_known"]), "keepalived": None} if not row['is_active']: @@ -2413,8 +2495,10 @@ async def get_agent_keepalived_config(agent_name: str, x_api_key: Optional[str] raise except Exception as e: logger.error(f"keepalived-config delivery failed for '{agent_name}': {e}") - # Degrade to no-op rather than 500 (B-7) — keeps the fleet inert on any error. - return {"agent_name": agent_name, "status": "not_configured", + # Degrade to a no-op rather than 500 (B-7). The status must be one the agent does not + # act on: not_configured is a teardown on a node whose file is ours, so a transient + # database error would have taken a running VIP down. + return {"agent_name": agent_name, "status": "unavailable", "config_path": "/etc/keepalived/keepalived.conf", "keepalived": None} finally: if conn: @@ -2476,12 +2560,54 @@ async def agent_keepalived_status(agent_name: str, status_data: dict, x_api_key: WHERE agent_id=$1 """, agent['id'], state, message, config_hash, config_hash) else: - await conn.execute(f""" - UPDATE vip_members SET last_deploy_state=$3, last_deploy_message=$4, - last_deploy_hash=$5, last_deploy_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP, - {_retire_takeover.format(p="$6")} - WHERE agent_id=$1 AND vip_id=$2 - """, agent['id'], int(vip_id), state, message, config_hash, config_hash) + # Issue #73: a node running several VIPs acks the merged file once, with the first + # VIP's id and the merged file's hash. Spread the ack over every VIP in that file; + # each member is converged against its OWN applied hash. + node_vips = await _node_applied_vips(conn, agent['id']) + node_vip_ids = [r['vip_id'] for r in node_vips] + converged = False + if len(node_vips) > 1 and int(vip_id) in node_vip_ids: + from services.keepalived_config import compose_node_keepalived_conf + try: + composed = compose_node_keepalived_conf( + [(r['vip_id'], r['applied_config_content']) for r in node_vips]) + composed_hash = hashlib.md5(composed.encode("utf-8")).hexdigest() + except Exception: # noqa: BLE001 — unmergeable means not converged, never a 500 + composed_hash = None + converged = bool(composed_hash) and config_hash == composed_hash + if converged: + await conn.execute(""" + UPDATE vip_members SET last_deploy_state=$3, last_deploy_message=$4, + last_deploy_hash=applied_config_hash, last_deploy_at=CURRENT_TIMESTAMP, + updated_at=CURRENT_TIMESTAMP, takeover_expected_hash=NULL + WHERE agent_id=$1 AND vip_id = ANY($2::int[]) + """, agent['id'], node_vip_ids, state, message) + else: + await conn.execute(""" + UPDATE vip_members SET last_deploy_state=$3, last_deploy_message=$4, + last_deploy_hash=$5, last_deploy_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP + WHERE agent_id=$1 AND vip_id = ANY($2::int[]) + """, agent['id'], node_vip_ids, state, message, config_hash) + else: + await conn.execute(f""" + UPDATE vip_members SET last_deploy_state=$3, last_deploy_message=$4, + last_deploy_hash=$5, last_deploy_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP, + {_retire_takeover.format(p="$6")} + WHERE agent_id=$1 AND vip_id=$2 + """, agent['id'], int(vip_id), state, message, config_hash, config_hash) + converged = any(r['vip_id'] == int(vip_id) and r['applied_config_hash'] == config_hash + for r in node_vips) + if state == "enabled" and converged: + # The file the node now runs holds only its active VIPs, so a deleted VIP's + # membership on this node is torn down even though no teardown was sent for it. + await conn.execute(""" + UPDATE vip_members m SET last_deploy_state='disabled', + last_deploy_message='removed from this node''s keepalived.conf', + last_deploy_at=CURRENT_TIMESTAMP, updated_at=CURRENT_TIMESTAMP + FROM vip_instances v + WHERE v.id = m.vip_id AND m.agent_id=$1 AND v.is_active = FALSE + AND m.last_deploy_state IS DISTINCT FROM 'disabled' + """, agent['id']) return {"status": "ok"} except HTTPException: raise @@ -2539,7 +2665,8 @@ async def agent_keepalived_discovery(agent_name: str, payload: dict, x_api_key: is_managed = bool(payload.get("is_managed")) config_hash = hashlib.md5(content.encode("utf-8", "replace")).hexdigest() - from services.keepalived_parser import analyse_keepalived_conf, KeepalivedParseError + from services.keepalived_parser import ( + ANALYSIS_VERSION, analyse_keepalived_conf, KeepalivedParseError) from services.keepalived_config import encrypt_vrrp_secret parse_error = None @@ -2562,8 +2689,8 @@ async def agent_keepalived_discovery(agent_name: str, payload: dict, x_api_key: await conn.execute(""" INSERT INTO vip_discoveries (agent_id, config_path, config_hash, is_managed, raw_config_masked, - auth_pass_encrypted, analysis, parse_error, reported_at) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,CURRENT_TIMESTAMP) + auth_pass_encrypted, analysis, parse_error, analysis_version, reported_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,CURRENT_TIMESTAMP) ON CONFLICT (agent_id) DO UPDATE SET config_path = EXCLUDED.config_path, config_hash = EXCLUDED.config_hash, @@ -2572,9 +2699,11 @@ async def agent_keepalived_discovery(agent_name: str, payload: dict, x_api_key: auth_pass_encrypted = EXCLUDED.auth_pass_encrypted, analysis = EXCLUDED.analysis, parse_error = EXCLUDED.parse_error, + analysis_version = EXCLUDED.analysis_version, reported_at = CURRENT_TIMESTAMP """, agent['id'], config_path, config_hash, is_managed, masked, auth_enc, - json.dumps(analysis) if analysis is not None else None, parse_error) + json.dumps(analysis) if analysis is not None else None, parse_error, + ANALYSIS_VERSION) return {"status": "recorded", "config_hash": config_hash} except HTTPException: raise diff --git a/backend/routers/vip.py b/backend/routers/vip.py index a2c7cfb..3d4469c 100644 --- a/backend/routers/vip.py +++ b/backend/routers/vip.py @@ -282,21 +282,9 @@ async def restore_vip_from_rejected_version(conn, version_id: int): if not await conn.fetchrow("SELECT 1 FROM vip_instances WHERE id=$1", vip_id): return "the VIP no longer exists — re-create it from the HA/VIP page" sv, sm = ps["vip"], ps.get("members", []) - # One-VIP-per-agent must STILL hold after reactivation: if a member was meanwhile added - # to another active VIP (while this one was rejected/soft-deleted), undoing here would - # create a double-membership — and since the delivery endpoint serves one VIP per agent, - # reactivating this (never-applied → not_configured) VIP would tear down the other live - # VIP on that node. Block it with a clear message (review round-3 FINDING 2). - member_ids = [m["agent_id"] for m in sm] - if member_ids: - dup = await conn.fetchrow( - "SELECT a.name AS agent_name, v.name AS vip_name FROM vip_members vm " - "JOIN vip_instances v ON v.id = vm.vip_id JOIN agents a ON a.id = vm.agent_id " - "WHERE vm.agent_id = ANY($1) AND v.is_active = TRUE AND v.id <> $2 LIMIT 1", - member_ids, vip_id) - if dup: - return (f"node '{dup['agent_name']}' now belongs to active VIP '{dup['vip_name']}' — " - f"remove it there first, then undo") + # Issue #73: a member that meanwhile joined another active VIP keeps it. The delivery + # endpoint serves a node every active VIP that has an applied snapshot, so reactivating + # this one (never applied, or applied earlier) cannot displace the other. # Preserve the running (applied) snapshot for members that remain, so undo of an edit # never momentarily flips a live node to not_configured (T-1). prev = {r["agent_id"]: r for r in await conn.fetch( @@ -540,24 +528,9 @@ async def _validate_members_against_pool(conn, pool_id: int, members: List, vip_ if row["pool_id"] != pool_id: raise HTTPException(status_code=400, detail=f"agent {m.agent_id} is not in pool {pool_id}") - # One active VIP per agent (v1): a node deploys a single keepalived.conf, and the - # delivery endpoint serves one VIP per agent — a second active membership would never - # converge (stuck SYNCING) instead of erroring. Reject it up front (review MED-2). - agent_ids = [m.agent_id for m in members] - if agent_ids: - dq = ("SELECT a.name AS agent_name, v.name AS vip_name " - "FROM vip_members vm JOIN vip_instances v ON v.id = vm.vip_id " - "JOIN agents a ON a.id = vm.agent_id " - "WHERE vm.agent_id = ANY($1) AND v.is_active = TRUE") - dparams = [agent_ids] - if exclude_vip_id is not None: - dq += " AND v.id <> $2" - dparams.append(exclude_vip_id) - dup = await conn.fetchrow(dq + " LIMIT 1", *dparams) - if dup: - raise HTTPException(status_code=409, - detail=f"node '{dup['agent_name']}' is already a member of VIP " - f"'{dup['vip_name']}' — a node can belong to only one VIP") + # Issue #73: a node may belong to several VIPs. The delivery endpoint merges their applied + # snapshots into the node's one keepalived.conf; the pool-scoped VRID uniqueness + # (uq_vip_vrid_active) keeps their VRRP instances apart. # VIP must not be an existing agent's primary IP (INET cast, SQL-1) clash = await conn.fetchval("SELECT 1 FROM agents WHERE ip_address = $1::inet LIMIT 1", vip_virtual_ip) if clash: @@ -826,6 +799,23 @@ async def apply_vip(vip_id: int, request: Request, authorization: str = Header(N master_prio = masters[0]["priority"] if any(m["role"] == "BACKUP" and m["priority"] >= master_prio for m in members): raise HTTPException(status_code=400, detail="MASTER priority must exceed every BACKUP") + # Issue #73: an adoption made before v1.12.0 may still be waiting for this Apply. Taking + # over a file with several vrrp_instance blocks would remove the others from the node. + multi = await conn.fetchrow(""" + SELECT a.name AS agent_name, (d.analysis->>'instance_count')::int AS n + FROM vip_members m + JOIN vip_discoveries d ON d.agent_id = m.agent_id + JOIN agents a ON a.id = m.agent_id + WHERE m.vip_id = $1 AND m.takeover_expected_hash IS NOT NULL + AND d.config_hash = m.takeover_expected_hash + AND COALESCE((d.analysis->>'instance_count')::int, 0) > 1 + LIMIT 1 + """, vip_id) + if multi: + raise HTTPException(status_code=409, detail=( + f"{multi['agent_name']}'s keepalived.conf declares {multi['n']} vrrp_instance blocks; " + f"taking it over would remove the others from the node. Reject this adoption.")) + auth_plain = decrypt_vrrp_secret(v["auth_pass_encrypted"]) if v["auth_pass_encrypted"] else None # If a secret is set but can't be decrypted (SECRET_KEY/VIP_ENCRYPTION_KEY rotated or @@ -1049,6 +1039,16 @@ def _discovery_row_to_api(row) -> dict: analysis = json.loads(analysis) except (json.JSONDecodeError, TypeError): analysis = None + # Issue #73: an analysis stored by an older parser does not carry the multi-instance + # blocker yet; show it here exactly as adopt_vip enforces it. + count = int((analysis or {}).get("instance_count") or 0) if isinstance(analysis, dict) else 0 + if count > 1: + from services.keepalived_parser import MULTI_INSTANCE_MARKER, multi_instance_blocker + for cand in analysis.get("candidates") or []: + blockers = cand.setdefault("blockers", []) + if not any(MULTI_INSTANCE_MARKER in b for b in blockers): + blockers.append(multi_instance_blocker(count)) + cand["adoptable"] = False return { "agent_id": row["agent_id"], "agent_name": row["agent_name"], @@ -1133,6 +1133,7 @@ async def _collect_instance_participants(conn, *, pool_id: int, vrid: int, virtu "agent_id": r["agent_id"], "agent_name": r["agent_name"], "ip_address": r["ip_address"], "config_hash": r["config_hash"], "auth_pass_encrypted": r["auth_pass_encrypted"], + "instance_count": int((analysis or {}).get("instance_count") or 0), "candidate": cand, }) break @@ -1206,7 +1207,8 @@ async def adopt_vip(payload: dict, request: Request, authorization: str = Header except (TypeError, ValueError): raise HTTPException(status_code=400, detail="prefix_length must be an integer") - from services.keepalived_parser import remaining_blockers + from services.keepalived_parser import ( + MULTI_INSTANCE_MARKER, multi_instance_blocker, remaining_blockers) vrid = vip_fields.get("virtual_router_id") virtual_ip = vip_fields.get("virtual_ip") @@ -1236,6 +1238,9 @@ async def adopt_vip(payload: dict, request: Request, authorization: str = Header prefix_supplied=vip_fields.get("prefix_length") is not None, accept_data_loss=bool(payload.get("accept_data_loss")), ) + # Issue #73: also for discoveries analysed before the parser reported this itself. + if p["instance_count"] > 1 and not any(MULTI_INSTANCE_MARKER in b for b in rem): + rem.append(multi_instance_blocker(p["instance_count"])) if rem: raise HTTPException(status_code=422, detail={ "message": f"This keepalived config cannot be adopted as-is ({p['agent_name']})", diff --git a/backend/services/keepalived_config.py b/backend/services/keepalived_config.py index 14eb9f5..80bbbed 100644 --- a/backend/services/keepalived_config.py +++ b/backend/services/keepalived_config.py @@ -190,3 +190,101 @@ def render_keepalived_conf(*, vip: dict, members: List[dict], this_agent: dict, lines.append(" }") lines.append("}") return "\n".join(lines) + "\n" + + +# --------------------------------------------------------------------------- +# Issue #73 — several VIPs on one node +# --------------------------------------------------------------------------- +# A node deploys ONE keepalived.conf, and each member row stores the full single-VIP file +# render_keepalived_conf produced at apply time. A node that belongs to several VIPs gets +# those applied files merged into one: global_defs once, then every VIP's vrrp_script and +# vrrp_instance. The health-check script is renamed per VIP (chk_haproxy_) because each +# VIP's failover weight is derived from its own priorities, and keepalived needs unique +# script names. A node with a single VIP is served its applied file unchanged. +class KeepalivedComposeError(ValueError): + """An applied snapshot is not in the shape render_keepalived_conf produces.""" + + +def _top_level_blocks(conf: str) -> List[tuple]: + """Split our own render into (name, lines) top-level blocks; comment lines are skipped.""" + blocks: List[tuple] = [] + current: Optional[List[str]] = None + depth = 0 + for line in conf.split("\n"): + stripped = line.strip() + if current is None: + if not stripped or stripped.startswith("#"): + continue + if not stripped.endswith("{"): + raise KeepalivedComposeError(f"unexpected top-level line: {stripped[:60]}") + current = [line] + depth = 1 + continue + current.append(line) + if stripped.startswith("auth_pass "): + continue # the secret may contain braces; it is a single token, never a block + depth += stripped.count("{") - stripped.count("}") + if depth == 0: + blocks.append((" ".join(current[0].strip().rstrip("{").split()), current)) + current = None + elif depth < 0: + raise KeepalivedComposeError("unbalanced braces") + if current is not None: + raise KeepalivedComposeError("unterminated block") + return blocks + + +def compose_node_keepalived_conf(applied: List[tuple]) -> str: + """Merge the applied single-VIP files of one node into the file the node deploys. + + `applied` is [(vip_id, applied_config_content), ...]. One entry is returned unchanged, + so nodes with a single VIP keep the exact bytes (and hash) they already run. Raises + KeepalivedComposeError when a snapshot does not have the expected shape; the caller + must then leave the node alone rather than deploy a partial file. + """ + if not applied: + raise KeepalivedComposeError("nothing to compose") + if len(applied) == 1: + return applied[0][1] + + header: List[str] = [f"{OWNERSHIP_MARKER} — DO NOT EDIT"] + global_defs: Optional[List[str]] = None + scripts: List[str] = [] + instances: List[str] = [] + for vip_id, conf in sorted(applied, key=lambda item: int(item[0])): + vip_id = int(vip_id) + for line in (conf or "").split("\n"): + if line.startswith("# VIP "): + header.append(line) + blocks = _top_level_blocks(conf or "") + names = [name for name, _ in blocks] + if names.count("global_defs") != 1 or names.count(f"vrrp_instance {_vrrp_instance_name(vip_id)}") != 1: + raise KeepalivedComposeError(f"VIP {vip_id}: applied config has an unexpected layout") + if any(n not in ("global_defs", "vrrp_script chk_haproxy", f"vrrp_instance {_vrrp_instance_name(vip_id)}") + for n in names): + raise KeepalivedComposeError(f"VIP {vip_id}: applied config has an unexpected block") + script_name = f"chk_haproxy_{vip_id}" + for name, body in blocks: + if name == "global_defs": + if global_defs is None: + global_defs = body + elif name == "vrrp_script chk_haproxy": + scripts.append("\n".join([f"vrrp_script {script_name} {{"] + body[1:])) + else: + renamed, in_track = [], False + for line in body: + s = line.strip() + if s == "track_script {": + in_track = True + elif in_track and s == "}": + in_track = False + elif in_track and s == "chk_haproxy": + line = line.replace("chk_haproxy", script_name) + renamed.append(line) + instances.append("\n".join(renamed)) + + out: List[str] = header + global_defs + [""] + for block in scripts: + out.extend([block, ""]) + out.append("\n\n".join(instances)) + return "\n".join(out) + "\n" diff --git a/backend/services/keepalived_parser.py b/backend/services/keepalived_parser.py index aa28bf1..608d3ed 100644 --- a/backend/services/keepalived_parser.py +++ b/backend/services/keepalived_parser.py @@ -31,6 +31,10 @@ import ipaddress import re from typing import Any, Dict, List, Optional, Tuple +# Stored with every discovery (vip_discoveries.analysis_version). Bump it whenever the +# analysis or its blockers change: rows from an older parser make the agent re-report once. +ANALYSIS_VERSION = 1 + # Directives `render_keepalived_conf` emits, and therefore the only ones a takeover can # reproduce. Anything else found in a vrrp_instance is reported in `unsupported`. _SUPPORTED_INSTANCE_KEYS = { @@ -379,6 +383,17 @@ _DEFAULT_STATE = "BACKUP" OUR_CHECK_SCRIPT_NAME = "chk_haproxy" +MULTI_INSTANCE_MARKER = "vrrp_instance blocks" + + +def multi_instance_blocker(count: int) -> str: + """Adoption replaces the node's whole keepalived.conf with ONE instance. Not resolvable: + the text must contain neither _LOSS_MARKER nor _PREFIX_MARKER.""" + return (f"the file declares {count} {MULTI_INSTANCE_MARKER}; adoption takes over the whole " + f"file, so the other instances would be removed from the node — adopting from a " + f"keepalived.conf with several instances is not supported") + + def build_adoption_candidate(parsed: Dict[str, Any], instance: Dict[str, Any]) -> Dict[str, Any]: """Map one parsed `vrrp_instance` onto vip_instances / vip_members fields. @@ -399,6 +414,13 @@ def build_adoption_candidate(parsed: Dict[str, Any], instance: Dict[str, Any]) - f"line {d['line']}: `{d['directive']}` — OpenManager's renderer cannot reproduce " f"this, so adopting would delete it") + # Adoption replaces the node's whole keepalived.conf with the render of ONE instance, so + # every other instance in the file would disappear from the node. Not resolvable: none of + # the wording below may contain _LOSS_MARKER or _PREFIX_MARKER. + instance_total = len(parsed.get("instances") or []) + if instance_total > 1: + blockers.append(multi_instance_blocker(instance_total)) + vips = instance.get("virtual_ips") or [] if len(vips) == 0: blockers.append("the instance declares no virtual_ipaddress — nothing to adopt") @@ -526,6 +548,7 @@ def analyse_keepalived_conf(text: str) -> Dict[str, Any]: """Parse + map in one call: the shape the discovery endpoint stores and the UI renders.""" parsed = parse_keepalived_conf(text) return { + "analysis_version": ANALYSIS_VERSION, "instance_count": len(parsed["instances"]), "sync_groups": parsed["sync_groups"], "global_defs": parsed["global_defs"], diff --git a/backend/tests/test_keepalived_parser.py b/backend/tests/test_keepalived_parser.py index 487a8b1..e2ff669 100644 --- a/backend/tests/test_keepalived_parser.py +++ b/backend/tests/test_keepalived_parser.py @@ -18,7 +18,7 @@ os.environ.setdefault("SECRET_KEY", "test-secret-key-for-keepalived-parser-tests from services import keepalived_config as kc # noqa: E402 from services.keepalived_parser import ( # noqa: E402 KeepalivedParseError, analyse_keepalived_conf, build_adoption_candidate, - parse_keepalived_conf, + parse_keepalived_conf, remaining_blockers, ) @@ -237,7 +237,12 @@ vrrp_instance VI_2 { names = [c["instance_name"] for c in analysed["candidates"]] assert names == ["VI_1", "VI_2"] assert [c["vip"]["virtual_ip"] for c in analysed["candidates"]] == ["10.0.0.100", "10.0.0.200"] - assert all(c["adoptable"] for c in analysed["candidates"]) + # Issue #73: adoption takes over the whole file, so adopting either instance would remove + # the other from the node. Neither may be adoptable, and no flag may waive it. + for c in analysed["candidates"]: + assert c["adoptable"] is False + assert remaining_blockers(c["blockers"], prefix_supplied=True, accept_data_loss=True) + assert any("2 vrrp_instance blocks" in b for b in c["blockers"]) def test_unbalanced_braces_raise(): diff --git a/backend/tests/test_vip_adoption_instance.py b/backend/tests/test_vip_adoption_instance.py index e3086fd..f293d71 100644 --- a/backend/tests/test_vip_adoption_instance.py +++ b/backend/tests/test_vip_adoption_instance.py @@ -460,7 +460,8 @@ def test_discovery_cache_defers_to_the_server(): assert agent_router.count('AS discovery_known') == 1, ( "the keepalived-config query must report whether a discovery row exists" ) - assert agent_router.count('"discovery_known": bool(agent["discovery_known"])') == 5, ( + # 7 = the five original paths + the v1.12.0 merged-delivery and compose_error paths. + assert agent_router.count('"discovery_known": bool(agent["discovery_known"])') == 7, ( "every response path that knows the agent must carry the flag — the adoptable nodes are " "precisely the not_configured ones" ) diff --git a/backend/tests/test_vip_multi_node.py b/backend/tests/test_vip_multi_node.py new file mode 100644 index 0000000..0220833 --- /dev/null +++ b/backend/tests/test_vip_multi_node.py @@ -0,0 +1,172 @@ +"""Issue #73 (v1.12.0) — several VIPs on one node. + +Pure tests for the merge of a node's applied single-VIP files, plus source-level guards for the delivery / ack / adoption / migration +invariants that need a live database to exercise end to end. +""" +from __future__ import annotations + +import hashlib +import os +import pathlib +import sys + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from services import keepalived_config as kc # noqa: E402 +from services.keepalived_parser import ( # noqa: E402 + ANALYSIS_VERSION, analyse_keepalived_conf, parse_keepalived_conf, +) + +BACKEND = pathlib.Path(__file__).resolve().parents[1] +AGENT_ROUTER = (BACKEND / "routers" / "agent.py").read_text() +VIP_ROUTER = (BACKEND / "routers" / "vip.py").read_text() +MIGRATIONS = (BACKEND / "database" / "migrations.py").read_text() + +NODE = {"role": "MASTER", "priority": 150, "network_interface": "eth0", + "agent_id": 1, "ip_address": "10.0.0.11"} +PEER = {"role": "BACKUP", "priority": 100, "network_interface": "eth0", + "agent_id": 2, "ip_address": "10.0.0.12"} + + +def _vip(vid, ip, vrid, **over): + base = {"id": vid, "name": f"vip-{vid}", "virtual_ip": ip, "prefix_length": 24, + "virtual_router_id": vrid, "advert_int": 1, "use_unicast": True, + "track_haproxy": True} + base.update(over) + return base + + +def _applied(vip, auth="s3cr3t", this=NODE, members=(NODE, PEER)): + peers = [m["ip_address"] for m in members if m["agent_id"] != this["agent_id"]] + return kc.render_keepalived_conf(vip=vip, members=list(members), this_agent=this, + peer_ips=peers, auth_pass_plain=auth) + + +A = _vip(3, "10.0.0.100", 51) +B = _vip(7, "10.0.0.200", 52, track_haproxy=False) + + +class TestCompose: + def test_single_vip_is_served_byte_identical(self): + conf = _applied(A) + assert kc.compose_node_keepalived_conf([(3, conf)]) == conf + + def test_two_vips_merge_into_one_valid_file(self): + out = kc.compose_node_keepalived_conf([(3, _applied(A)), (7, _applied(B))]) + assert out.count("global_defs {") == 1 + assert out.startswith(kc.OWNERSHIP_MARKER) + assert "vrrp_script chk_haproxy_3 {" in out + assert "vrrp_script chk_haproxy {" not in out + assert "vrrp_instance VI_3 {" in out and "vrrp_instance VI_7 {" in out + assert out.index("VI_3") < out.index("VI_7") + assert " chk_haproxy_3\n" in out + parsed = parse_keepalived_conf(out) + assert [i["instance_name"] for i in parsed["instances"]] == ["VI_3", "VI_7"] + assert [i["track_scripts"] for i in parsed["instances"]] == [["chk_haproxy_3"], []] + + def test_order_follows_vip_id_not_input_order(self): + a, b = (3, _applied(A)), (7, _applied(B)) + assert kc.compose_node_keepalived_conf([b, a]) == kc.compose_node_keepalived_conf([a, b]) + + def test_each_instance_keeps_its_own_failover_weight(self): + a2 = _vip(9, "10.0.0.150", 53) + low_peer = {**PEER, "priority": 50} + out = kc.compose_node_keepalived_conf([ + (3, _applied(A)), (9, _applied(a2, members=(NODE, low_peer)))]) + assert "weight -51" in out and "weight -101" in out + + def test_braces_in_the_secret_do_not_break_the_merge(self): + out = kc.compose_node_keepalived_conf([(3, _applied(A, auth="a}b{")), + (7, _applied(B, auth="}x"))]) + assert "auth_pass a}b{" in out and "auth_pass }x" in out + assert out.count("vrrp_instance") == 2 + + @pytest.mark.parametrize("bad", [ + "", + "global_defs {\n}\n", + "vrrp_instance VI_3 {\n state MASTER\n", + "something else\n", + ]) + def test_unexpected_snapshot_fails_closed(self, bad): + with pytest.raises(kc.KeepalivedComposeError): + kc.compose_node_keepalived_conf([(3, _applied(A)), (7, bad)]) + + def test_foreign_block_fails_closed(self): + tampered = _applied(B) + "virtual_server 10.0.0.1 80 {\n}\n" + with pytest.raises(kc.KeepalivedComposeError): + kc.compose_node_keepalived_conf([(3, _applied(A)), (7, tampered)]) + + def test_snapshot_of_another_vip_id_fails_closed(self): + with pytest.raises(kc.KeepalivedComposeError): + kc.compose_node_keepalived_conf([(3, _applied(A)), (8, _applied(B))]) + + def test_nothing_to_compose_raises(self): + with pytest.raises(kc.KeepalivedComposeError): + kc.compose_node_keepalived_conf([]) + + +class TestDeliveryAndAckGuards: + def test_failures_answer_with_statuses_the_agent_ignores(self): + assert '"status": "compose_error"' in AGENT_ROUTER + generic = AGENT_ROUTER.split("keepalived-config delivery failed", 1)[1][:600] + assert '"status": "unavailable"' in generic + assert "not_configured" not in generic.split("return", 1)[1].split("}", 1)[0] + + def test_merged_delivery_hash_is_md5_of_the_composed_text(self): + assert 'hashlib.md5(composed.encode("utf-8")).hexdigest()' in AGENT_ROUTER + out = kc.compose_node_keepalived_conf([(3, _applied(A)), (7, _applied(B))]) + assert len(hashlib.md5(out.encode("utf-8")).hexdigest()) == 32 + + def test_only_applied_snapshots_are_merged(self): + q = AGENT_ROUTER.split("async def _node_applied_vips", 1)[1].split("async def ", 1)[0] + assert "v.is_active = TRUE" in q + assert "m.applied_config_content IS NOT NULL" in q + + def test_ack_failure_to_merge_is_not_a_500(self): + ack = AGENT_ROUTER.split("acks the merged file once", 1)[1][:1200] + assert "except Exception" in ack and "composed_hash = None" in ack + + def test_discovery_known_asks_old_analyses_to_be_reported_again(self): + assert "vd.analysis_version >= $2" in AGENT_ROUTER + assert ANALYSIS_VERSION >= 1 + assert analyse_keepalived_conf(_applied(A))["analysis_version"] == ANALYSIS_VERSION + + +class TestAdoptionGuards: + def test_apply_refuses_an_inflight_adoption_of_a_multi_instance_file(self): + apply = VIP_ROUTER.split("async def apply_vip", 1)[1].split("\nasync def ", 1)[0] + assert "d.config_hash = m.takeover_expected_hash" in apply + assert "(d.analysis->>'instance_count')::int, 0) > 1" in apply + assert "status_code=409" in apply + + def test_old_analyses_get_the_blocker_in_the_list_and_on_adopt(self): + assert VIP_ROUTER.count("multi_instance_blocker(") >= 2 + row_api = VIP_ROUTER.split("def _discovery_row_to_api", 1)[1].split("\ndef ", 1)[0] + assert 'cand["adoptable"] = False' in row_api + + +class TestDeltaMigration: + def test_versions(self): + assert "SCHEMA_VERSION = 13" in MIGRATIONS + assert "_DELTA_BASE_VERSION = 12" in MIGRATIONS + assert "return [(13, ensure_vip_multi_address_columns)]" in MIGRATIONS + + def test_full_sequence_also_creates_the_columns(self): + inner = MIGRATIONS.split("async def _run_all_migrations_inner", 1)[1].split("\nasync def ", 1)[0] + assert inner.index("await ensure_vip_tables()") < inner.index( + "await ensure_vip_multi_address_columns()") + + def test_delta_path_does_not_reseed_roles(self): + run = MIGRATIONS.split("async def run_all_migrations", 1)[1].split("\nasync def ", 1)[0] + delta = run.split("applied_version >= _DELTA_BASE_VERSION", 1)[1].split("else:", 1)[0] + assert "_delta_steps()" in delta + assert "_run_all_migrations_inner" not in delta + assert "update_system_roles" not in delta + + def test_columns_are_additive_with_safe_defaults(self): + body = MIGRATIONS.split("async def ensure_vip_multi_address_columns", 1)[1].split( + "\nasync def ", 1)[0].replace('"\n "', "") + assert "ADD COLUMN IF NOT EXISTS additional_ips JSONB NOT NULL DEFAULT '[]'::jsonb" in body + assert "ADD COLUMN IF NOT EXISTS analysis_version SMALLINT NOT NULL DEFAULT 0" in body diff --git a/frontend/src/components/VIPManagement.js b/frontend/src/components/VIPManagement.js index 688b8c0..ec85c2f 100644 --- a/frontend/src/components/VIPManagement.js +++ b/frontend/src/components/VIPManagement.js @@ -78,7 +78,7 @@ const mergeBlockers = (lists) => { // This component uses raw fetch(), but extractApiError expects an axios-shaped error // (err.response.data). Read the fetch Response body and reuse the envelope-aware extractor -// so backend messages — e.g. the 409 "node already in VIP X" — actually reach the user. +// so backend messages — e.g. the 409 "address already used by VIP X" — actually reach the user. const fetchApiError = async (res, fallback) => { try { const data = await res.json(); return extractApiError({ response: { data } }, fallback); } catch { return fallback; }