#!/usr/bin/env python3 """Operator-run, one-stage-at-a-time role tests for the two-node lab. No installation, automatic recovery, fencing, or maintenance-mode changes. See docs/lab-role-tests.md before running a mutating stage. """ import argparse import hashlib import json import os from pathlib import Path import pwd import shlex import socket import subprocess import tempfile import time import xml.etree.ElementTree as ET PROVIDER = Path('/usr/lib/ocf/resource.d/loxthelion') AGENT = PROVIDER / 'managed-container' NODES = {'DAVID-ELITEDESK': '192.168.1.6', 'DAVID-DV6': '192.168.1.2'} PINNED = {'DAVID-DV6': {'managed-container': 'a4b775a0ddcabdb0558da9b34c7fe2a883c07d0ef308c1c7224526abf53f63a0', 'managed-container.d/auth.sh': '93191771d3cb8aefadb2a16d2f89d64a8b6c584071a473439be818237bdf7f17', 'managed-container.d/common.sh': '97e93d9164e854e5f3688cf9d3cc656b001eb30ef3c2073c907450ca43588f3d', 'managed-container.d/container.sh': 'ac6bc363a2301f1491d162ef609103dbfbceb3b099a3556614c6165952309374', 'managed-container.d/demote.sh': '0af3bb362a303017372c631354ba94ef9e723bf14e64d2eff91b6662e54a5be0', 'managed-container.d/notify.sh': '6a98fcdda82cea9365db95b4ee779be5c567b3d846f48b90fbc8c254af697670', 'managed-container.d/postgres.sh': 'd2779ce6e7d1f426ffca2e41bbb8de77503005dc38d658d69f184020365796eb', 'managed-container.d/promote.sh': '3085ec9b4410905ab9d4370786ed776eac3470a0415aa1b4f608b083945aeb58', 'managed-container.d/promotion_state.py': '5166fc3370f49ab14cd975ded4b03a74b9bd7dbff5762b41a15552c42f934ea7', 'managed-container.d/rejoin.sh': '21beea72bcb7f24a51c17ea48666ac87df9436702ad77b166b45f4647e28e8ad', 'managed-container.d/replication.sh': '6b2021887e3b4cd6a12490e7a8dff98320b2133b472ab25a22a0eead46137547'}, 'DAVID-ELITEDESK': {'managed-container': '01eb9b161c8e3db289f09bf63953be779921e15bdf8b1a4d8ce7e60032b826e4', 'managed-container.d/applications/postgresql/auth.sh': '10932919c4339b553c7c9dbf30b45a917c51c0a08f362e12678003d58b12fa82', 'managed-container.d/applications/postgresql/demote.sh': 'da4e58cadf1df3953cb97a0512e3931ee1b9d3ee61c39a1ad1a6047ef537addc', 'managed-container.d/applications/postgresql/notify.sh': '6a98fcdda82cea9365db95b4ee779be5c567b3d846f48b90fbc8c254af697670', 'managed-container.d/applications/postgresql/plugin.sh': 'd3a989de97ceb61a56828f006d5ccb87b7c34fc155a944585bedf928785021fc', 'managed-container.d/applications/postgresql/postgres.sh': 'fa222e6767249999767eb82fac2d85e7e35404da240517a8b630571e77282053', 'managed-container.d/applications/postgresql/promote.sh': '3085ec9b4410905ab9d4370786ed776eac3470a0415aa1b4f608b083945aeb58', 'managed-container.d/applications/postgresql/promotion_state.py': '5166fc3370f49ab14cd975ded4b03a74b9bd7dbff5762b41a15552c42f934ea7', 'managed-container.d/applications/postgresql/rejoin.sh': '28b0e44554a88ccb87cd16a68315c045dd37d354aceb2fa25da7c7399a659b23', 'managed-container.d/applications/postgresql/replication.sh': '6b2021887e3b4cd6a12490e7a8dff98320b2133b472ab25a22a0eead46137547', 'managed-container.d/common.sh': 'a482ef3cc1048e6a4f427e76dd2dce173ffd00760dbed1afa601a3fd05b0ad99', 'managed-container.d/container.sh': '3a27ac9c63c61310da8b8711fa2c4400b37ea8eae2ee2a79e28bc63180b06c7c', 'managed-container.d/engine.sh': '53978e2367394695d61c6fb38f2ee46814474c5a4e0448d2692f1a3ce583af93', 'managed-container.d/lifecycles/systemd.sh': '6c92c095ce10ac95d843fdc1c49e31c19491ca22fdac960affd8649606d76d82', 'managed-container.d/runtimes/podman.sh': 'dc5820067a78c4a1c531426902793aae97a03fd64912b017b3b102031dcc1e04'}} QUERY = """SELECT json_build_object( 'recovery',pg_is_in_recovery(), 'paused',pg_is_wal_replay_paused(), 'system_id',(SELECT system_identifier::text FROM pg_control_system()), 'started',pg_postmaster_start_time()::text, 'receiver',(SELECT status FROM pg_stat_wal_receiver), 'sender',(SELECT sender_host FROM pg_stat_wal_receiver), 'slot',(SELECT slot_name FROM pg_stat_wal_receiver), 'replay',pg_last_wal_replay_lsn()::text);""" # pg_is_wal_replay_paused() is only callable during recovery. QUERY = QUERY.replace('pg_is_wal_replay_paused()', 'CASE WHEN pg_is_in_recovery() THEN pg_is_wal_replay_paused() ELSE false END') def require(ok, message): if not ok: raise RuntimeError(message) def run(args, **kwargs): result = subprocess.run(args, text=True, capture_output=True, timeout=90, env=dict(PATH='/usr/sbin:/usr/bin:/sbin:/bin', LC_ALL='C'), **kwargs) require(result.returncode == 0, f'{args[0]} failed: {result.stderr.strip()}') return result.stdout.strip() def owner_args(*args): account = pwd.getpwnam('david') return ['runuser', '-u', 'david', '--', 'env', 'HOME=' + account.pw_dir, 'XDG_RUNTIME_DIR=/run/user/' + str(account.pw_uid), *args] def cluster(): cib = ET.fromstring(run(['cibadmin', '--query'])) def prop(name): return run(['crm_attribute', '--type', 'crm_config', '--name', name, '--query', '--quiet']) require(prop('cluster-name') == 'postgres-ra-test' and prop('maintenance-mode') == 'true', 'Expected postgres-ra-test with maintenance=true; nothing changed') require({n.get('uname') for n in cib.findall('./configuration/nodes/node')} == set(NODES), 'Unexpected cluster membership') resource = cib.find("./configuration/resources/clone[@id='pg-ha-clone']/primitive[@id='pg-ha']") require(resource is not None and (resource.get('class'), resource.get('provider'), resource.get('type')) == ('ocf', 'loxthelion', 'managed-container'), 'Unexpected resource') attrs = {} for group in resource.findall('instance_attributes'): require(not group.get('id-ref') and group.find('rule') is None, 'Unsupported parameter rules') for pair in group.findall('nvpair'): key, value = pair.get('name'), pair.get('value') require(key and value is not None and key not in attrs, 'Ambiguous parameters') attrs[key] = value defaults = dict(service_user='david', service_name='postgres.service', container_name='postgres', pg_os_user='postgres', db_user='david', db_name='david', application='postgresql', container_runtime='podman', lifecycle_manager='systemd') require(all(attrs.get(k, v) == v for k, v in defaults.items()), 'Unexpected service/application settings') return dict(defaults, **attrs) def snapshot(node): # Executes only read operations as david on either host. Never inspects container environment. code = """import hashlib,json,pathlib,socket,subprocess def cmd(args): p=subprocess.run(['timeout','--kill-after=1s','5s']+args,text=True,capture_output=True) if p.returncode: raise RuntimeError('Read-only observation failed: '+args[0]) return p.stdout.strip() files=FILES base=pathlib.Path('/usr/lib/ocf/resource.d/loxthelion') assert all(not (base/n).is_symlink() for n in files) print(json.dumps(dict(node=socket.gethostname(), hashes={n:hashlib.sha256((base/n).read_bytes()).hexdigest() for n in files}, service=cmd(['systemctl','--user','show','postgres.service','--property=ActiveState','--value']), container=cmd(['podman','inspect','--format','{{.Id}}','postgres']), db=json.loads(cmd(['podman','exec','--user','postgres','--env','PGOPTIONS=-c statement_timeout=2000', 'postgres','psql','-X','-v','ON_ERROR_STOP=1','-U','david','-d','david','-Atqc',QUERY]))))) """.replace('FILES', repr(list(PINNED[node]))).replace('QUERY', repr(QUERY)) if node == socket.gethostname(): args = owner_args('python3', '-c', code) else: args = owner_args('ssh', '-F', '/dev/null', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=5', '-o', 'StrictHostKeyChecking=yes', 'david@' + NODES[node], 'python3 -c ' + shlex.quote(code)) value = json.loads(run(args, cwd=pwd.getpwnam('david').pw_dir)) require(value['node'] == node and value['hashes'] == PINNED[node], 'Node or installed code differs: ' + node) require(value['service'] == 'active', 'Maintenance service not active: ' + node) return value def standby(db, source): return (db['recovery'] is True and db['paused'] is False and db['receiver'] == 'streaming' and db['sender'] == NODES[source]) def guard(stage, node, local, peer): require(local['system_id'] == peer['system_id'], 'Database system identifiers differ') other = next(n for n in NODES if n != node) if stage == 'demote': require(local['recovery'] is False and standby(peer, node), 'Need local primary and streaming peer standby') elif stage == 'promote': require(local['recovery'] is True and local['paused'] is False and standby(peer, node), 'Both nodes must be in recovery and the demoted peer must stream from this node') elif stage == 'rejoin': require(node == 'DAVID-ELITEDESK' and standby(local, other) and peer['recovery'] is False, 'Rejoin test requires ELITEDESK streaming from primary DV6') def main(stage): node = socket.gethostname() require(os.geteuid() == 0 and node in NODES, 'Run with sudo on an expected lab node') os.umask(0o077) peer = next(n for n in NODES if n != node) attrs = cluster() before, remote = snapshot(node), snapshot(peer) guard(stage, node, before['db'], remote['db']) if stage == 'inspect': print(json.dumps(dict(local=before, peer=remote, maintenance=True), indent=2)) return evidence = Path(tempfile.mkdtemp(prefix='managed-container-roles-lab-', dir='/var/backups')) print('Results:', evidence, flush=True) result = dict(stage=stage, node=node, before=before, peer_before=remote, actions=[], status='running') env = dict(PATH='/usr/sbin:/usr/bin:/sbin:/bin', OCF_ROOT='/usr/lib/ocf', OCF_RESOURCE_INSTANCE='pg-ha', OCF_RESKEY_CRM_meta_on_node=node, **{'OCF_RESKEY_' + k: v for k, v in attrs.items()}) def save(): (evidence / 'results.json').write_text(json.dumps(result, indent=2) + '\n') def action(name, expected=0, extra=None): require(cluster() == attrs, 'Cluster configuration or maintenance changed') response = subprocess.run(['timeout', '--kill-after=5s', '120s', str(AGENT), name], env=dict(env, **(extra or {})), cwd='/', text=True, capture_output=True) result['actions'].append(dict(action=name, rc=response.returncode, expected=expected)) (evidence / f'{len(result["actions"]):02d}-{name}.log').write_text(response.stdout + response.stderr) save() require(response.returncode == expected, f'{name} returned {response.returncode}; expected {expected}') print(f'PASS {name}: rc={response.returncode}', flush=True) save() try: action('validate-all') action('monitor', 8 if before['db']['recovery'] is False else 0) # Reobserve immediately before mutation. No action retry or inferred fencing certificate. current, remote = snapshot(node), snapshot(peer) require(current['container'] == before['container'] and all(current['db'][k] == before['db'][k] for k in ('system_id', 'started', 'recovery')), 'Local incarnation/role changed during preflight') guard(stage, node, current['db'], remote['db']) if stage == 'demote': action('notify', extra=dict(OCF_RESKEY_CRM_meta_notify_type='pre', OCF_RESKEY_CRM_meta_notify_operation='demote', OCF_RESKEY_CRM_meta_notify_demote_uname=node, OCF_RESKEY_CRM_meta_notify_promote_uname=peer)) action('demote') elif stage == 'promote': action('promote') elif stage == 'rejoin': # This stage tests rewind/recovery of a healthy standby, not divergent former-primary repair. run(owner_args('timeout', '--kill-after=5s', '60s', 'podman', 'exec', '--user', 'postgres', 'postgres', 'sh', '-c', 'exec "$(pg_config --bindir)/pg_ctl" -D "$PGDATA" stop -m fast -w')) action('monitor', 7) require(run(['crm_attribute', '--promotion=pg-ha', '--node', node, '--lifetime=reboot', '--query', '--quiet']) == '-INFINITY', 'Stopped PostgreSQL retained an eligible score') require(snapshot(peer)['db']['recovery'] is False, 'DV6 is no longer primary') action('rejoin', extra=dict(OCF_RESKEY_rejoin_source_host=NODES[peer], OCF_RESKEY_replication_slot='ha_david_elitedesk', OCF_RESKEY_application_name=node, OCF_RESKEY_rejoin_allow_basebackup='false')) deadline = time.monotonic() + 60 while True: after, remote = snapshot(node), snapshot(peer) require(after['db']['system_id'] == before['db']['system_id'], 'Local database identity changed') require(after['container'] == before['container'], 'Maintenance container was replaced') require(remote['db']['system_id'] == before['db']['system_id'], 'Peer database identity changed') if stage == 'promote': require(after['db']['recovery'] is False and standby(remote['db'], node), 'Unexpected roles after promotion') expected_score = '1000' else: require(standby(after['db'], peer), 'Local node did not return to streaming recovery') require(after['db']['started'] != before['db']['started'], 'Expected a new postmaster') require(remote['db']['recovery'] is (stage == 'demote'), 'Unexpected peer role') expected_score = '100' action('monitor', 8 if stage == 'promote' else 0) value = run(['crm_attribute', '--promotion=pg-ha', '--node', node, '--lifetime=reboot', '--query', '--quiet']) if value == expected_score: break require(time.monotonic() < deadline, 'Expected promotion score did not converge') time.sleep(2) result.update(after=after, peer_after=remote, score=value, status='passed') require(cluster() == attrs, 'Cluster configuration changed') save() print(f'PASS {stage}: verified roles, streaming, same database/container, score={value}.', flush=True) print('Maintenance remains true. No fencing or automatic recovery ran.', flush=True) except BaseException as error: result.update(status='failed', error=str(error)) save() print('HALTED. Do not continue to the next stage. Inspect:', evidence, flush=True) raise if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('stage', choices=['inspect', 'demote', 'promote', 'rejoin']) try: main(parser.parse_args().stage) except Exception as error: raise SystemExit('STOP: ' + str(error))