Lyncoln
ec2077f979
Extract PostgreSQL, Podman runtime and systemd lifecycle adapters while preserving existing lifecycle and scoring behavior. Update RPM layout and add guarded operator-run lab harnesses. Record successful manual stop/start, promotion, demotion and rewind-only rejoin with pinned installed hashes. The unfenced lab does not validate automatic failover, Pacemaker-managed transitions or divergent-data rewind. Validation: 103 offline tests, original-function differential checks, checksum manifest and whitespace checks passed.
251 lines
15 KiB
Python
251 lines
15 KiB
Python
#!/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))
|