293 lines
13 KiB
Python
293 lines
13 KiB
Python
#!/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)
|