From 349434fc4ded0258f6cc467556e1e0020e3de5c6 Mon Sep 17 00:00:00 2001 From: taylanbakircioglu Date: Sat, 17 Oct 2026 13:04:38 +0300 Subject: [PATCH] feat(vip): let a node belong to several keepalived VIPs (issue #73) A node could be a member of one VIP only: the form refused a second one, and the delivery endpoint served each agent a single VIP's keepalived.conf. - Delivery merges the node's applied VIPs into the one file it deploys: global_defs once, then per VIP a vrrp_script chk_haproxy_ (each keeps its own failover weight) and its vrrp_instance, ordered by VIP id. A node with one VIP gets the same bytes and hash as before, so existing nodes do not rewrite or reload. - One ack converges every VIP in the merged file; a deleted VIP's membership is marked disabled once the node runs a file without it. - A snapshot that cannot be merged leaves the node alone (status compose_error, members show the reason). A database error during delivery answers "unavailable" instead of "not_configured", which on a managed node meant teardown. - Adoption refuses a keepalived.conf with several vrrp_instance blocks (taking it over would remove the others), also for analyses stored by older versions and at Apply for an adoption made before the upgrade. Discoveries record the parser version; files analysed by an older one are re-reported once by v1.11.1+ agents. - Schema 13 is a delta from 12 (two additive columns); built-in roles are not re-seeded on a database already at 12. No agent change. --- backend/database/migrations.py | 58 ++++++- backend/routers/agent.py | 177 +++++++++++++++++--- backend/routers/vip.py | 73 ++++---- backend/services/keepalived_config.py | 98 +++++++++++ backend/services/keepalived_parser.py | 23 +++ backend/tests/test_keepalived_parser.py | 9 +- backend/tests/test_vip_adoption_instance.py | 3 +- backend/tests/test_vip_multi_node.py | 172 +++++++++++++++++++ frontend/src/components/VIPManagement.js | 2 +- 9 files changed, 551 insertions(+), 64 deletions(-) create mode 100644 backend/tests/test_vip_multi_node.py 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; }