Freeze HA scoring and document reported production failover

This commit is contained in:
Lyncoln
2026-09-12 18:07:28 -04:00
parent c8636ed6d4
commit 488a1672d7
29 changed files with 2285 additions and 46 deletions
+199
View File
@@ -0,0 +1,199 @@
#!/usr/bin/env python3
"""State and monitor integration tests: temporary files and fake boundaries only."""
import copy
import importlib.util
import json
import os
from pathlib import Path
import shlex
import tempfile
import unittest
from test_promotion_scores import Scores, LOAD, FIXTURE
ROOT = Path(__file__).resolve().parents[1]
HELPER = ROOT / 'refactored/managed-container.d/promotion_state.py'
spec = importlib.util.spec_from_file_location('promotion_state', HELPER)
policy = importlib.util.module_from_spec(spec)
spec.loader.exec_module(policy)
CONNECTED = dict(identity=['7684451577990111248', '/data/pg18', '1789200000.123456'],
recovery=True, paused=False, receiver='streaming', replay='0/30',
flush='0/20', strict=True)
DISCONNECTED = dict(CONNECTED, receiver=None, flush=None, strict=None)
class State(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='mc-evidence-')
self.addCleanup(self.tmp.cleanup)
self.evidence = policy.Evidence(Path(self.tmp.name) / 'promotion',
['pg-ha:0', 'owner', 'postgres.service', 'postgres'],
'test-boot', 1000.0, 1789201000.0, os.getuid())
def check(self, obs=DISCONNECTED, age=300):
return self.evidence.evaluate(copy.deepcopy(obs), age)
def seed(self):
self.assertEqual(self.check(CONNECTED), '100')
return json.loads(self.evidence.path.read_text())
def test_streaming_disconnected_eligible_and_persisted_lsn(self):
record = self.seed()
self.assertEqual(record['replay'], '0/30')
self.assertEqual(record['flush'], '0/20')
self.evidence.now += 50
self.assertEqual(self.check(), '50')
self.assertEqual(json.loads(self.evidence.path.read_text()), record)
self.assertEqual(self.evidence.path.stat().st_mode & 0o777, 0o600)
self.assertEqual(self.evidence.directory.stat().st_mode & 0o777, 0o700)
def test_never_verified(self):
self.assertEqual(self.check(), '-INFINITY')
def test_stale_and_exact_boundary(self):
self.seed(); self.evidence.now += 300
self.assertEqual(self.check(), '50')
self.evidence.now += .001
self.assertEqual(self.check(), '-INFINITY')
self.assertFalse(self.evidence.path.exists())
def test_disabled_and_invalid_age(self):
self.seed()
self.assertEqual(self.check(age=0), '-INFINITY')
for age in (-1, True, 100000, '300'):
with self.assertRaises(ValueError): self.check(age=age)
def test_malformed_state(self):
for value in ('{', '[]', '{}', 'null', '{"version":true}', '{"version":1,"version":1}', 'x' * 4097):
self.seed(); self.evidence.path.write_text(value)
self.assertEqual(self.check(), '-INFINITY', value[:40])
self.assertFalse(self.evidence.path.exists())
def test_invalid_record_fields(self):
for key, value in [('replay', '0/ZZ'), ('flush', '1/0'), ('boot', 'other'),
('observed_boottime', float('nan')), ('observed_boottime', 1001),
('observed_unix', True), ('context', ['other']), ('version', True)]:
record = self.seed(); record[key] = value
self.evidence.path.write_text(json.dumps(record))
self.assertEqual(self.check(), '-INFINITY', key)
def test_system_data_path_and_postmaster_identity_changes(self):
for index, value in enumerate(['999', '/new/pgdata', '1789200100.123456']):
self.seed(); obs = copy.deepcopy(DISCONNECTED); obs['identity'][index] = value
self.assertEqual(self.check(obs), '-INFINITY')
self.assertFalse(self.evidence.path.exists())
def test_reinitialization_requires_reverification_then_replaces_state(self):
self.seed()
disconnected = copy.deepcopy(DISCONNECTED)
disconnected['identity'][2] = '1789200100.123456' # Same system ID/basebackup.
self.assertEqual(self.check(disconnected), '-INFINITY')
connected = dict(CONNECTED, identity=disconnected['identity'])
self.assertEqual(self.check(connected), '100')
self.assertEqual(self.check(disconnected), '50')
self.assertEqual(json.loads(self.evidence.path.read_text())['identity'], connected['identity'])
def test_paused_primary_and_replay_regression_invalidate(self):
for changes in ({'paused': True}, {'recovery': False, 'paused': True},
{'replay': '0/10'}, {'replay': None}):
self.seed()
self.assertEqual(self.check(dict(DISCONNECTED, **changes)), '-INFINITY')
self.assertFalse(self.evidence.path.exists())
def test_connected_failure_cannot_use_disconnected_fallback(self):
for changes in ({'flush': '0/40'}, {'flush': None}, {'paused': True}):
self.seed()
self.assertEqual(self.check(dict(CONNECTED, **changes, strict=False)), '-INFINITY')
self.assertFalse(self.evidence.path.exists())
def test_reconnect_refreshes_timestamp_and_lsns(self):
self.seed(); self.evidence.now += 100
self.assertEqual(self.check(), '50')
self.assertEqual(self.check(dict(CONNECTED, replay='0/50', flush='0/40')), '100')
record = json.loads(self.evidence.path.read_text())
self.assertEqual(record['observed_boottime'], 1100)
self.assertEqual(record['replay'], '0/50')
self.assertEqual(self.check(dict(DISCONNECTED, replay='0/50')), '50')
def test_wall_clock_steps_do_not_extend_or_expire_observation(self):
self.seed(); self.evidence.wall -= 100000; self.evidence.now += 100
self.assertEqual(self.check(), '50')
self.evidence.wall += 200000; self.evidence.now += 201
self.assertEqual(self.check(), '-INFINITY')
def test_symlink_state_rejected_without_reading_target(self):
self.seed(); self.evidence.path.unlink()
target = Path(self.tmp.name) / 'target'; target.write_text('unchanged')
self.evidence.path.symlink_to(target)
self.assertEqual(self.check(), '-INFINITY')
self.assertEqual(target.read_text(), 'unchanged')
def test_unsafe_directory_rejected(self):
self.seed(); self.evidence.directory.chmod(0o777)
with self.assertRaises(ValueError): self.check()
# Reuse fake crm_attribute and the real monitor wrapper. The real state evaluator
# runs as this test user against a private directory; never writes /run or CIB.
class Monitor(unittest.TestCase):
setUp = Scores.setUp
run_shell = Scores.run_shell
state = Scores.state
score = Scores.score
def test_invalid_observation_query_and_helper_failure_replace_score(self):
fixture = monitor_fixture(self)
self.env['OBSERVATION'] = json.dumps(CONNECTED)
for failure in ('mc_pg_standby_observation() { echo malformed; }',
'mc_pg_standby_observation() { return 1; }',
'python3() { return 124; }'):
self.run_shell('mc_promotion_monitor', fixture=fixture)
self.score('100')
self.run_shell(failure + '; mc_promotion_monitor', fixture=fixture)
self.score('-INFINITY')
def monitor_fixture(test):
driver = test.path / 'driver.py'
driver.write_text('''import importlib.util,json,os,sys,time
from pathlib import Path
spec=importlib.util.spec_from_file_location('policy',sys.argv[1])
p=importlib.util.module_from_spec(spec);spec.loader.exec_module(p)
e=p.Evidence(Path(sys.argv[2]),['pg-ha:0','owner','service','container'],'boot',float(os.environ.get('TEST_NOW','1000')),1789201000,os.getuid())
print(e.evaluate(json.load(sys.stdin),300))
''')
# Restore the production score helper after the original tests' mock.
return FIXTURE + '\n. ' + shlex.quote(str(HELPER.with_name('postgres.sh'))) + '''
service_running() { [ "$SERVICE" = 1 ]; }
container_running() { [ "$CONTAINER" = 1 ]; }
pg_running() { [ "$RUNNING" = 1 ]; }
pg_ready() { return 0; }
pg_role() { echo "$ROLE"; }
mc_pg_standby_observation() { printf '%s' "$OBSERVATION"; }
python3() { command python3 ''' + ' '.join(map(shlex.quote, [str(driver), str(HELPER), str(test.path / 'evidence')])) + '''; }
timeout() { shift 2; "$@"; }
'''
def test_monitor_disconnect_reprobe_stopped_reconnect(self):
fixture = monitor_fixture(self)
self.env['OBSERVATION'] = json.dumps(CONNECTED)
self.run_shell('mc_promotion_monitor', fixture=fixture); self.score('100')
self.env['OBSERVATION'] = json.dumps(DISCONNECTED)
self.run_shell('mc_promotion_monitor', fixture=fixture); self.score('50')
data = self.state(); data['scores'] = {}; self.db.write_text(json.dumps(data))
self.run_shell('mc_promotion_monitor', fixture=fixture); self.score('50')
self.run_shell('RUNNING=0; mc_promotion_monitor', 7, fixture=fixture); self.score('-INFINITY')
self.env['OBSERVATION'] = json.dumps(CONNECTED)
self.run_shell('mc_promotion_monitor', fixture=fixture); self.score('100')
self.env['OBSERVATION'] = json.dumps(DISCONNECTED); self.env['TEST_NOW'] = '1301'
self.run_shell('mc_promotion_monitor', fixture=fixture); self.score('-INFINITY')
Monitor.test_monitor_disconnect_reprobe_stopped_reconnect = test_monitor_disconnect_reprobe_stopped_reconnect
if __name__ == '__main__':
# Explicit suite excludes imported Scores from duplicate discovery.
unittest.TextTestRunner(verbosity=2).run(unittest.TestSuite(
unittest.defaultTestLoader.loadTestsFromTestCase(c) for c in (State, Monitor)
)).wasSuccessful() or __import__('sys').exit(1)
+40
View File
@@ -0,0 +1,40 @@
#!/usr/bin/env python3
import importlib.util
from pathlib import Path
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('cleanup', ROOT / 'scripts/plan-promotion-cleanup.py')
cleanup = importlib.util.module_from_spec(spec); spec.loader.exec_module(cleanup)
class Cleanup(unittest.TestCase):
def plan(self, extra='', role='Promoted'):
with tempfile.TemporaryDirectory() as directory:
p = Path(directory) / 'cib.xml'
p.write_text(f'''<cib><configuration><resources><clone id="postgresql-ha-clone">
<meta_attributes id="meta"><nvpair name="target-role" value="{role}"/></meta_attributes>
</clone></resources><constraints><rsc_location id="actual-eve-preference" rsc="postgresql-ha-clone">
<rule id="actual-rule" role="Promoted" score="200"><expression attribute="#uname" operation="eq" value="david-eve"/></rule>
{extra}</rsc_location></constraints></configuration></cib>''')
return cleanup.plan(p, 'postgresql-ha-clone', 'david-eve')
def test_exact_commands_from_real_ids(self):
self.assertEqual(self.plan(), ['sudo pcs constraint remove actual-eve-preference'])
def test_preserve_unrelated_rule(self):
self.assertEqual(self.plan('<rule id="unrelated" role="Unpromoted" score="10"/>')[0],
'sudo pcs constraint remove actual-rule')
def test_ambiguous_match_refused(self):
with self.assertRaises(ValueError):
self.plan('<rule id="duplicate" role="Promoted" score="200"><expression attribute="#uname" operation="eq" value="david-eve"/></rule>')
def test_unexpected_target_role_refused(self):
with self.assertRaises(ValueError):
self.plan(role='Started')
if __name__ == '__main__':
unittest.main(verbosity=2)
+114
View File
@@ -0,0 +1,114 @@
#!/usr/bin/env python3
"""Real Pacemaker scheduler tests using synthetic CIB files only."""
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
import xml.etree.ElementTree as ET
@unittest.skipUnless(shutil.which('crm_simulate') and shutil.which('cibadmin'), 'Pacemaker CLI not installed')
class Scheduler(unittest.TestCase):
def simulate(self, scores, target=None, fail_primary=False):
if fail_primary and target is None:
target = 'Promoted' # Required production policy.
with tempfile.TemporaryDirectory(prefix='mc-scheduler-') as directory:
root = ET.fromstring(subprocess.check_output(['cibadmin', '--empty'], text=True))
root.set('have-quorum', '1'); root.set('dc-uuid', '1')
config = root.find('configuration')
props = ET.SubElement(config.find('crm_config'), 'cluster_property_set', id='test-props')
# ONLY this synthetic file: no fencing device is required for an offline simulation.
ET.SubElement(props, 'nvpair', id='test-stonith', name='stonith-enabled', value='true' if fail_primary else 'false')
for i, score in enumerate(scores, 1):
node = 'node-' + str(i)
ET.SubElement(config.find('nodes'), 'node', id=str(i), uname=node)
state = ET.SubElement(root.find('status'), 'node_state', id=str(i), uname=node,
in_ccm='true', crmd='online', join='member', expected='member')
transient = ET.SubElement(state, 'transient_attributes', id=str(i))
attrs = ET.SubElement(transient, 'instance_attributes', id='attrs-' + str(i))
# Names generated by crm_attribute --promotion=pg-ha, not old pgsql state.
ET.SubElement(attrs, 'nvpair', id='score-' + str(i), name='master-pg-ha', value=score)
ET.SubElement(attrs, 'nvpair', id='legacy-' + str(i), name='master-postgresql', value='-INFINITY')
if fail_primary:
ET.SubElement(config.find('resources'), 'primitive',
{'id': 'test-fence', 'class': 'stonith', 'type': 'fence_dummy'})
clone = ET.SubElement(config.find('resources'), 'clone', id='pg-ha-clone')
meta = ET.SubElement(clone, 'meta_attributes', id='clone-meta')
values = {'promotable': 'true', 'clone-max': str(len(scores)), 'clone-node-max': '1',
'promoted-max': '1', 'promoted-node-max': '1', 'notify': 'true'}
if target is not None:
values['target-role'] = target
for name, value in values.items():
ET.SubElement(meta, 'nvpair', id='clone-' + name, name=name, value=value)
ET.SubElement(clone, 'primitive', {'id': 'pg-ha', 'class': 'ocf', 'provider': 'loxthelion', 'type': 'managed-container'})
cib = Path(directory) / 'cib.xml'; graph = Path(directory) / 'graph.xml'
ET.ElementTree(root).write(cib)
if fail_primary:
# Establish running primary/standbys in a synthetic simulation;
# then inject an unexpected failure. No agents/devices execute.
subprocess.run(['crm_simulate', '--xml-file', str(cib), '--in-place'],
check=True, capture_output=True, text=True)
subprocess.run(['crm_simulate', '--xml-file', str(cib), '--run', '--save-graph', str(graph)]
+ (['--node-fail', 'node-1'] if fail_primary else []),
check=True, capture_output=True, text=True)
parsed = ET.parse(graph)
if fail_primary:
# Verify fencing is a transitive prerequisite of every promote.
actions = {op.get('id'): op for op in parsed.findall('./synapse/action_set/*')}
dependencies = {}
for synapse in parsed.findall('./synapse'):
for op in synapse.findall('./action_set/*'):
dependencies[op.get('id')] = {x.get('id') for x in synapse.findall('./inputs/trigger/*')}
fences = {key for key, op in actions.items() if op.get('operation') == 'stonith'}
self.assertTrue(fences, ET.tostring(parsed.getroot()).decode())
def ancestors(key, seen=None):
seen = set() if seen is None else seen
for parent in dependencies.get(key, set()) - seen:
seen.add(parent); ancestors(parent, seen)
return seen
for key, op in actions.items():
if op.get('operation') == 'promote':
self.assertTrue(ancestors(key) & fences, 'promotion must depend on fencing')
return [op.get('on_node') for op in ET.parse(graph).findall('./synapse/action_set/rsc_op')
if op.get('operation') == 'promote']
def test_default_started_allows_one_promotion(self):
self.assertEqual(len(self.simulate(['100', '100'])), 1)
def test_explicit_started_allows_one_promotion(self):
self.assertEqual(len(self.simulate(['100', '100'], 'Started')), 1)
def test_explicit_promoted_allows_one_promotion(self):
self.assertEqual(len(self.simulate(['100', '100'], 'Promoted')), 1)
def test_unpromoted_target_prevents_promotion(self):
self.assertEqual(self.simulate(['100', '100'], 'Unpromoted'), [])
def test_higher_score_wins_without_host_preference(self):
self.assertEqual(self.simulate(['100', '1000']), ['node-2'])
def test_ineligible_node_not_selected(self):
self.assertEqual(self.simulate(['-INFINITY', '100']), ['node-2'])
def test_all_ineligible_no_promotion(self):
self.assertEqual(self.simulate(['-INFINITY', '-INFINITY']), [])
def test_failed_primary_one_disconnected_candidate(self):
self.assertEqual(self.simulate(['1000', '50'], fail_primary=True), ['node-2'])
def test_failed_primary_two_disconnected_candidates(self):
first = self.simulate(['1000', '50', '50'], fail_primary=True)
self.assertEqual(len(first), 1)
self.assertIn(first[0], ['node-2', 'node-3'])
self.assertEqual(first, self.simulate(['1000', '50', '50'], fail_primary=True))
def test_failed_primary_all_disconnected_stale(self):
self.assertEqual(self.simulate(['1000', '-INFINITY', '-INFINITY'], fail_primary=True), [])
def test_connected_evidence_preferred(self):
self.assertEqual(self.simulate(['50', '100']), ['node-2'])
if __name__ == '__main__':
unittest.main(verbosity=2)
+292
View File
@@ -0,0 +1,292 @@
#!/usr/bin/env python3
"""Offline scoring/action tests. All cluster and PostgreSQL boundaries are fake."""
import json
import os
from pathlib import Path
import shlex
import shutil
import xml.etree.ElementTree as ET
import subprocess
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[1]
AGENT = ROOT / 'refactored/managed-container'
MODULES = ('common', 'container', 'postgres', 'auth', 'replication', 'notify', 'promote', 'demote', 'rejoin')
LOAD = 'MC_PROMOTION_RESOURCE_INSTANCE="${OCF_RESOURCE_INSTANCE:-}"\n: "${OCF_ROOT:=/usr/lib/ocf}"\n. "$OCF_ROOT/lib/heartbeat/ocf-shellfuncs"\n: "${OCF_RUNNING_PROMOTED:=${OCF_RUNNING_MASTER:-8}}"\n'
LOAD += '\n'.join('. ' + shlex.quote(str(AGENT.parent / 'managed-container.d' / (m + '.sh'))) for m in MODULES)
CRM = '''#!/usr/bin/env python3
import json, os, sys
from pathlib import Path
args=sys.argv[1:]
p=Path(os.environ['SCORE_DB'])
d=json.loads(p.read_text())
assert '--lifetime=reboot' in args, args
resource=next(x.split('=',1)[1] for x in args if x.startswith('--promotion='))
assert not any(x.startswith('--name') for x in args), args
node=os.environ.get('OCF_RESKEY_CRM_meta_on_node','node-a')
key=resource+'|'+node
d['calls'].append(args)
if '--query' in args:
p.write_text(json.dumps(d))
if os.environ.get('QUERY_FAIL')=='1': sys.exit(1)
if key not in d['scores']: sys.exit(6)
print(d['scores'][key]); sys.exit(0)
value=next(x.split('=',1)[1] for x in args if x.startswith('--update='))
if os.environ.get('UPDATE_FAIL')=='1':
p.write_text(json.dumps(d)); sys.exit(1)
d['scores'][key]=value
d['writes'].append([key,value])
p.write_text(json.dumps(d))
'''
# Original action implementations execute against these fake external boundaries.
FIXTURE = r'''
ROLE=t
RUNNING=1
SERVICE=1
CONTAINER=1
ELIGIBLE=1
SIGNAL=1
ocf_log() { :; }
ocf_exit_reason() { printf '%s\n' "$*" >&2; }
service_running() { [ "$SERVICE" = 1 ]; }
container_running() { [ "$CONTAINER" = 1 ]; }
pg_running() { [ "$RUNNING" = 1 ]; }
pg_ready() { return 0; }
pg_role() { echo "$ROLE"; }
pg_wal_receiver_status() { echo streaming; }
mc_pg_promotion_eligible() { [ "$ELIGIBLE" = 1 ]; }
mc_pg_standby_score() { mc_pg_promotion_eligible && echo 100; }
validate_all() { return 0; }
ensure_local_ha_slots() { return 0; }
ensure_rewind_privileges() { return 0; }
demote_target() { echo node-b; }
node_to_addr() { echo 192.0.2.2; }
candidate_is_standby() { return 0; }
candidate_caught_up() { return 0; }
ensure_remote_slot() { return 0; }
pgdata_path() { echo /test/pgdata; }
pg_remote_query() { echo f; }
pg_replication_exec() { return 0; }
pg_command() {
case "$*" in
*'test -f'*) [ "$SIGNAL" = 1 ] ;;
*'promote -w'*) ROLE=f ;;
*'stop -m fast -w'*) RUNNING=0 ;;
*'start -w'*) RUNNING=1 ;;
*) echo "Unexpected pg_command: $*" >&2; return 99 ;;
esac
}
as_service_user() {
case "$*" in
'systemctl --user stop '*) SERVICE=0; CONTAINER=0 ;;
'systemctl --user start '*) SERVICE=1; CONTAINER=1 ;;
*RA_PRIMARY_HOST=*) ROLE=t ;;
*) echo "Unexpected external access: $*" >&2; return 99 ;;
esac
}
sleep() { :; }
'''
class Scores(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix='mc-score-test-')
self.addCleanup(self.tmp.cleanup)
self.path = Path(self.tmp.name)
self.db = self.path / 'scores.json'
self.db.write_text(json.dumps({'scores': {}, 'calls': [], 'writes': []}))
(self.path / 'crm_attribute').write_text(CRM)
(self.path / 'crm_attribute').chmod(0o755)
self.env = dict(os.environ, PATH=str(self.path) + ':' + os.environ['PATH'],
SCORE_DB=str(self.db), OCF_RESOURCE_INSTANCE='pg-ha:0',
OCF_RESKEY_CRM_meta_on_node='node-a')
def run_shell(self, code, rc=0, fixture=FIXTURE, env=None):
result = subprocess.run(['bash', '-c', LOAD + fixture + '\n' + code],
env=env or self.env, text=True, capture_output=True, timeout=15)
self.assertEqual(result.returncode, rc, result.stderr)
return result
def state(self):
return json.loads(self.db.read_text())
def score(self, value, node='node-a'):
self.assertEqual(self.state()['scores']['pg-ha:0|' + node], value)
def test_primary_and_standby(self):
self.run_shell('ROLE=f; mc_promotion_monitor', 8)
self.score('1000')
self.env['OCF_RESKEY_CRM_meta_on_node'] = 'node-b'
self.run_shell('mc_promotion_monitor')
self.score('100', 'node-b')
def test_two_standbys_equal(self):
for node in ('node-a', 'node-b'):
self.env['OCF_RESKEY_CRM_meta_on_node'] = node
self.run_shell('mc_promotion_monitor')
self.score('100', node)
def test_ineligible_replaces_stale_positive(self):
cases = [('RUNNING=0', 7), ('CONTAINER=0', 7), ('SERVICE=0', 7),
('ELIGIBLE=0', 0), ('pg_ready() { return 1; }', 1),
('pg_role() { return 1; }', 1), ('ROLE=invalid', 1)]
for setup, rc in cases:
with self.subTest(setup=setup):
self.run_shell('mc_promotion_score_set 1000')
self.run_shell(setup + '; mc_promotion_monitor', rc)
self.score('-INFINITY')
def test_repeat_monitors_write_once(self):
for _ in range(3):
self.run_shell('mc_promotion_monitor')
self.assertEqual(len(self.state()['writes']), 1)
self.assertEqual(len(self.state()['calls']), 4) # Three queries, one update.
def test_restart_reprobe_recreates_attribute(self):
self.run_shell('mc_promotion_monitor')
data = self.state(); data['scores'] = {}; self.db.write_text(json.dumps(data))
self.run_shell('mc_promotion_monitor')
self.score('100')
self.assertEqual(len(self.state()['writes']), 2)
def test_stop_start_original_actions(self):
self.run_shell('mc_promotion_monitor; mc_promotion_action stop; '
'[ "$RUNNING$SERVICE$CONTAINER" = 000 ]; '
'mc_promotion_action start')
self.score('100')
self.assertEqual([x[1] for x in self.state()['writes']], ['100', '-INFINITY', '100'])
def test_primary_start_refusal(self):
self.run_shell('ROLE=f; mc_promotion_action start', 1)
self.score('-INFINITY')
def test_missing_standby_signal(self):
self.run_shell('RUNNING=0; SIGNAL=0; mc_promotion_action start', 1)
self.score('-INFINITY')
def test_promote_original_action(self):
self.run_shell('mc_promotion_monitor; mc_promotion_action promote')
self.score('1000')
self.assertEqual([x[1] for x in self.state()['writes']], ['100', '-INFINITY', '1000'])
def test_demote_original_action(self):
self.run_shell('ROLE=f; mc_promotion_monitor; mc_promotion_action demote; '
'[ "$ROLE$SERVICE$CONTAINER" = t11 ]')
self.score('100')
self.assertEqual([x[1] for x in self.state()['writes']], ['1000', '-INFINITY', '100'])
def test_failed_demote_stays_negative_until_next_observation(self):
self.run_shell('ROLE=f; demote_target() { return 1; }; mc_promotion_action demote', 1)
self.score('-INFINITY')
self.run_shell('ROLE=f; mc_promotion_monitor', 8)
self.score('1000')
def test_failed_promote_invariants_withdraw_score(self):
self.run_shell('ensure_local_ha_slots() { return 1; }; mc_promotion_action promote', 1)
self.score('-INFINITY')
def test_validation_failure_withdraws_score(self):
self.run_shell('mc_promotion_monitor; validate_all() { return 6; }; mc_promotion_action start', 6)
self.score('-INFINITY')
def test_stop_does_not_require_validation(self):
self.run_shell('validate_all() { return 6; }; mc_promotion_action stop')
self.score('-INFINITY')
def test_rejoin_original_action(self):
self.run_shell('RUNNING=0; REJOIN_SOURCE_HOST=192.0.2.2; REPLICATION_SLOT=ha_node; '
'mc_promotion_action rejoin')
self.score('100')
def test_update_failure_preserves_primary_role_rc(self):
self.env['UPDATE_FAIL'] = '1'
result = self.run_shell('ocf_log() { echo "$*" >&2; }; ROLE=f; mc_promotion_monitor', 8)
self.assertIn('prior score may remain', result.stderr)
self.assertFalse(self.state()['scores'])
def test_update_failure_does_not_block_stop(self):
self.env['UPDATE_FAIL'] = '1'
self.run_shell('mc_promotion_action stop; [ "$RUNNING" = 0 ]')
def test_query_failure_attempts_replace(self):
self.env['QUERY_FAIL'] = '1'
self.run_shell('ELIGIBLE=0; mc_promotion_monitor')
self.score('-INFINITY')
def test_scoring_dependency_validation(self):
self.run_shell('command() { return 1; }; mc_promotion_validate', 5)
def test_missing_instance_never_uses_default_attribute(self):
env = dict(self.env)
env.pop('OCF_RESOURCE_INSTANCE')
self.run_shell('mc_promotion_monitor', env=env)
self.assertFalse(self.state()['calls'])
def test_target_role_does_not_affect_scoring(self):
for value in ('', 'Started', 'Promoted'):
self.env['OCF_RESKEY_CRM_meta_target_role'] = value
self.run_shell('mc_promotion_monitor')
self.score('100')
def test_unrelated_legacy_attributes_untouched(self):
d = self.state()
d['scores']['master-postgresql'] = '-INFINITY'
d['scores']['postgresql-data-status'] = 'DISCONNECT'
d['scores']['postgresql-status'] = 'STOP'
self.db.write_text(json.dumps(d))
self.run_shell('mc_promotion_monitor')
for key, value in d['scores'].items():
self.assertEqual(self.state()['scores'][key], value)
@unittest.skipUnless(shutil.which('crm_attribute') and shutil.which('cibadmin'), 'Pacemaker CLI not installed')
def test_real_crm_attribute_against_offline_cib(self):
# CIB_file forces Pacemaker's file backend: no live daemon is contacted.
cib = self.path / 'offline.xml'
tree = ET.fromstring(subprocess.check_output(['cibadmin', '--empty'], text=True))
ET.SubElement(tree.find('./configuration/nodes'), 'node', id='1', uname='node-a')
ET.SubElement(tree.find('./status'), 'node_state', id='1', uname='node-a')
cib.write_text(ET.tostring(tree, encoding='unicode'))
env = dict(self.env, PATH=os.environ['PATH'], CIB_file=str(cib))
self.run_shell('mc_promotion_score_set 100', env=env)
before = cib.read_bytes()
self.run_shell('mc_promotion_score_set 100', env=env)
self.assertEqual(cib.read_bytes(), before, 'Identical observations must not rewrite CIB')
self.run_shell('mc_promotion_score_set -INFINITY', env=env)
pairs = ET.parse(cib).findall('./status/node_state/transient_attributes/instance_attributes/nvpair')
self.assertEqual([(p.get('name'), p.get('value')) for p in pairs], [('master-pg-ha:0', '-INFINITY')])
# Reprobe after loss of transient state must recreate the correct score.
tree = ET.parse(cib)
state = tree.find('./status/node_state')
state.remove(state.find('./transient_attributes'))
tree.write(cib)
self.run_shell('mc_promotion_score_set 1000', env=env)
self.assertEqual(ET.parse(cib).find('.//transient_attributes//nvpair').get('value'), '1000')
def test_actual_eligibility_helper_rejects_non_true_or_failed_sql(self):
# Exercise the production shell helper's output and error handling.
for answer, sql_rc, expected in [('t', 0, 0), ('f', 0, 1), ('', 0, 1),
('t\nf', 0, 1), ('t', 1, 1)]:
fixture = '\nas_service_user() { printf "%s" ' + shlex.quote(answer) + '; return ' + str(sql_rc) + '; }\n'
self.run_shell('mc_pg_promotion_eligible', expected, fixture=fixture)
def test_dispatch_routes_only_operational_actions_through_scoring(self):
# Use the real dispatcher with fake handlers; no cluster calls occur.
dispatch = AGENT.read_text().split('case "$1" in', 1)[1]
for action in ('start', 'stop', 'promote', 'demote', 'rejoin'):
code = 'mc_promotion_action() { echo "scored:$1"; return 0; }; set -- ' + action + '\ncase "$1" in' + dispatch
result = self.run_shell(code)
self.assertEqual(result.stdout.strip(), 'scored:' + action)
for action in ('monitor', 'status'):
code = 'mc_promotion_monitor() { echo scored-monitor; return 8; }; set -- ' + action + '\ncase "$1" in' + dispatch
self.assertEqual(self.run_shell(code, 8).stdout.strip(), 'scored-monitor')
for action, handler in [('meta-data', 'mc_promotion_metadata'), ('validate-all', 'mc_promotion_validate'), ('notify', 'managed_notify')]:
code = handler + '() { echo plain; return 0; }; set -- ' + action + '\ncase "$1" in' + dispatch
self.assertEqual(self.run_shell(code).stdout.strip(), 'plain')
self.assertFalse(self.state()['calls'])
if __name__ == '__main__':
unittest.main(verbosity=2)
+5 -3
View File
@@ -18,9 +18,11 @@ assert output('rpm', '-q', '--qf', '%{ARCH}', PACKAGE) == 'noarch'
assert output('rpm', '-q', '--scripts', PACKAGE) == ''
subprocess.run(['rpm', '-V', PACKAGE], check=True)
expected = {AGENT: ROOT / 'refactored/managed-container'}
for path in (ROOT / 'refactored/managed-container.d').glob('*.sh'):
for path in (ROOT / 'refactored/managed-container.d').glob('*'):
if not path.is_file():
continue
expected[PROVIDER / 'managed-container.d' / path.name] = path
assert len(expected) == 10
assert len(expected) == 11
for path, source in expected.items():
assert path.read_bytes() == source.read_bytes(), path
info = path.stat()
@@ -32,7 +34,7 @@ allowed = {str(path) for path in expected} | {str(PROVIDER), str(PROVIDER / 'man
unexpected = {path for path in owned - allowed if not path.startswith('/usr/share/doc/' + PACKAGE)}
assert not unexpected, unexpected
reference = output(str(ROOT / 'reference/managed-container'), 'meta-data')
reference = output(str(ROOT / 'refactored/managed-container'), 'meta-data')
installed = subprocess.check_output([str(AGENT), 'meta-data'], cwd='/tmp', text=True)
assert installed == reference
print('PASS installed RPM: noarch, file integrity, root ownership, modes, payload, no scriptlets, metadata/module loading')
+94
View File
@@ -0,0 +1,94 @@
#!/usr/bin/env python3
"""Run the production eligibility SQL against controlled PostgreSQL 18 fixtures.
Explicit opt-in integration test: creates/removes only its own disposable Podman
container. No host ports, volumes, cluster credentials or live databases used.
The postgres:18 image must already be present locally.
"""
from pathlib import Path
import os
import json
import shlex
import subprocess
import tempfile
import time
import uuid
ROOT = Path(__file__).resolve().parents[1]
with tempfile.TemporaryDirectory(prefix='mc-score-sql-') as directory:
query_file = Path(directory) / 'query.sql'
env = dict(os.environ, QUERY_FILE=str(query_file))
extract = '. ' + shlex.quote(str(ROOT / 'refactored/managed-container.d/postgres.sh')) + '''
as_service_user() { printf '%s' "${@: -1}" > "$QUERY_FILE"; echo t; }
mc_pg_promotion_eligible
'''
subprocess.run(['bash', '-c', extract], env=env, check=True)
query = query_file.read_text()
subprocess.run(['bash', '-c', extract.replace('mc_pg_promotion_eligible', 'mc_pg_standby_observation')], env=env, check=True)
observation_query = query_file.read_text()
name = 'mc-score-sql-' + uuid.uuid4().hex[:12]
subprocess.run(['podman', 'run', '-d', '--pull=never', '--name', name,
'--network=none', '--tmpfs', '/var/lib/postgresql',
'-e', 'POSTGRES_HOST_AUTH_METHOD=trust',
'docker.io/library/postgres:18'], check=True, capture_output=True)
try:
for _ in range(40):
ready = subprocess.run(['podman', 'exec', name, 'pg_isready', '-U', 'postgres'],
capture_output=True)
if ready.returncode == 0:
break
time.sleep(0.5)
else:
raise RuntimeError('Disposable PostgreSQL did not become ready')
def sql(text):
return subprocess.check_output(['podman', 'exec', '-i', name, 'psql',
'-X', '-U', 'postgres', '-v', 'ON_ERROR_STOP=1',
'-Atq'], input=text, text=True).strip()
assert sql('SHOW server_version_num;').startswith('18')
sql('''
CREATE SCHEMA score_test;
CREATE TABLE score_test.state (recovery bool, paused bool, replay pg_lsn);
INSERT INTO score_test.state VALUES (true,false,'0/20');
CREATE TABLE score_test.pg_stat_wal_receiver (status text, flushed_lsn pg_lsn);
CREATE FUNCTION score_test.pg_is_in_recovery() RETURNS bool LANGUAGE sql
AS 'SELECT recovery FROM score_test.state';
CREATE FUNCTION score_test.pg_is_wal_replay_paused() RETURNS bool LANGUAGE plpgsql AS $$
BEGIN
IF NOT (SELECT recovery FROM score_test.state) THEN
RAISE EXCEPTION 'replay pause is undefined on a primary';
END IF;
RETURN (SELECT paused FROM score_test.state);
END $$;
CREATE FUNCTION score_test.pg_last_wal_replay_lsn() RETURNS pg_lsn LANGUAGE sql
AS 'SELECT replay FROM score_test.state';
''')
cases = [
('caught-up streaming standby', True, False, '0/20', 'streaming', '0/20', 't'),
('replay ahead of receiver snapshot', True, False, '0/30', 'streaming', '0/20', 't'),
('replay lag', True, False, '0/10', 'streaming', '0/20', 'f'),
('paused replay', True, True, '0/20', 'streaming', '0/20', 'f'),
('no replay position', True, False, None, 'streaming', '0/20', 'f'),
('unknown flushed position', True, False, '0/20', 'streaming', None, 'f'),
('receiver stopped', True, False, '0/20', 'stopping', '0/20', 'f'),
('no receiver after primary failure', True, False, '0/20', None, None, 'f'),
('primary skips recovery-only function', False, False, '0/20', None, None, 'f'),
]
def literal(value):
return 'NULL' if value is None else "'" + value + "'"
for label, recovery, paused, replay, status, flushed, expected in cases:
setup = f'''SET search_path=score_test,pg_catalog;
UPDATE score_test.state SET recovery={str(recovery).lower()},paused={str(paused).lower()},replay={literal(replay)};
TRUNCATE score_test.pg_stat_wal_receiver;
'''
if status is not None:
setup += f'INSERT INTO score_test.pg_stat_wal_receiver VALUES ({literal(status)},{literal(flushed)});\n'
result = sql(setup + query)
assert result == expected, (label, result, expected)
observation = json.loads(sql(setup + observation_query))
assert (observation['strict'] is True) == (expected == 't'), (label, observation)
assert observation['identity'][0].isdigit() and observation['identity'][1].startswith('/')
assert observation['recovery'] == recovery
assert observation['receiver'] == status
print('PASS strict and observation SQL:', label)
finally:
subprocess.run(['podman', 'rm', '-f', name], check=True, capture_output=True)
+40
View File
@@ -0,0 +1,40 @@
#!/usr/bin/env python3
"""Explicit disposable-container-only test of the root helper CLI and /run state."""
import json
import os
from pathlib import Path
import subprocess
import sys
import uuid
from test_disconnected_state import CONNECTED, DISCONNECTED
assert os.geteuid() == 0 and os.environ.get('MC_DISPOSABLE_STATE_TEST') == '1', (
'Run only in a disposable container with MC_DISPOSABLE_STATE_TEST=1')
helper = (Path(sys.argv[1]) if len(sys.argv) > 1 else
Path(__file__).resolve().parents[1] / 'refactored/managed-container.d/promotion_state.py')
resource = 'mc-disposable-' + uuid.uuid4().hex
args = ['python3', str(helper), resource, 'test-owner', 'test.service', 'test-container', '300']
def run(obs, expected):
result = subprocess.run(args, input=json.dumps(obs), text=True, capture_output=True, timeout=3)
assert result.returncode == 0 and result.stdout.strip() == expected, result
run(DISCONNECTED, '-INFINITY')
run(CONNECTED, '100')
files = list(Path('/run/managed-container/promotion').glob('*.json'))
record_path = next(p for p in files if json.loads(p.read_text())['context'][0] == resource)
record = json.loads(record_path.read_text())
assert record['replay'] == CONNECTED['replay'] and record['flush'] == CONNECTED['flush']
for path, mode in [(record_path, 0o600), (record_path.parent, 0o700), (record_path.parent.parent, 0o700)]:
info = path.stat()
assert (info.st_uid, info.st_gid, info.st_mode & 0o777) == (0, 0, mode), path
run(DISCONNECTED, '50')
assert json.loads(record_path.read_text()) == record # Disconnection never renews evidence.
record_path.write_text('{malformed')
run(DISCONNECTED, '-INFINITY')
assert not record_path.exists()
run(CONNECTED, '100')
run(dict(DISCONNECTED, identity=['999', '/new', '1789201000.123456']), '-INFINITY')
assert not record_path.exists()
print('PASS root CLI: fixed /run location, ownership/modes, persistence, disconnect, malformed and replaced data')
+19 -5
View File
@@ -41,13 +41,19 @@ for path in (REFERENCE, ROOT / 'reference/managed-container.pre-refactor-5eae1f3
assert hashlib.sha256(path.read_bytes()).hexdigest() == expected_hash
print('PASS both reference hashes')
assert functions(SOURCE) == functions('\n'.join(p.read_text() for p in MODULES))
current_functions = functions('\n'.join(p.read_text() for p in MODULES))
for name, definition in functions(SOURCE).items():
assert current_functions[name] == definition, name
assert set(current_functions) - set(functions(SOURCE)) == {
'mc_promotion_score_set', 'mc_promotion_score_from_monitor',
'mc_promotion_monitor', 'mc_promotion_action', 'mc_promotion_validate', 'mc_pg_promotion_eligible',
'mc_promotion_metadata', 'mc_pg_standby_observation', 'mc_pg_standby_score',
}
assert len(functions(SOURCE)) == 29
common = (AGENT.parent / 'managed-container.d/common.sh').read_text()
defaults = SOURCE[SOURCE.index('OCF_RESKEY_service_user_default='):SOURCE.index('\n###############################################################################\n# Metadata')].strip()
assert defaults in common
assert SOURCE[SOURCE.index('case "$1" in'):] == AGENT.read_text()[AGENT.read_text().index('case "$1" in'):]
print('PASS all 29 function bodies, defaults/globals and dispatcher identical')
print('PASS all 29 original function bodies and defaults/globals identical; nine scoring helpers added')
for path in [AGENT, *MODULES]:
subprocess.run(['bash', '-n', str(path)], check=True)
@@ -58,11 +64,19 @@ for path in (REFERENCE, AGENT):
result = subprocess.run([str(path), 'meta-data'], cwd='/tmp', capture_output=True, text=True)
assert result.returncode == 0 and not result.stderr, result
metadata.append(result.stdout)
assert metadata[0] == metadata[1]
extended = ET.fromstring(metadata[1])
parameters = extended.find('parameters')
added = parameters.find("parameter[@name='disconnected_max_age']")
assert added is not None and added.find('content').get('default') == '300'
parameters.remove(added)
def normalize(element):
return (element.tag, element.attrib, (element.text or '').strip(),
[normalize(child) for child in element])
assert normalize(extended) == normalize(ET.fromstring(metadata[0]))
xml = ET.fromstring(metadata[0])
assert len(xml.findall('./parameters/parameter')) == 12
assert len(xml.findall('./actions/action')) == 8
print('PASS metadata byte equality and XML parsing from unrelated cwd')
print('PASS original metadata preserved plus freshness parameter and XML parsing from unrelated cwd')
# Sourcing production modules in a fresh shell tests their real loading order.
# No dispatcher executes. All operational boundaries are overridden below.