66 lines
3.1 KiB
Python
66 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Print (never execute) exact cleanup commands from a supplied CIB snapshot."""
|
|
import argparse
|
|
from pathlib import Path
|
|
import shlex
|
|
import sys
|
|
import xml.etree.ElementTree as ET
|
|
|
|
|
|
def plan(path, clone_id, preferred_node):
|
|
root = ET.parse(path).getroot()
|
|
clones = [c for c in root.findall('./configuration/resources/clone') if c.get('id') == clone_id]
|
|
if len(clones) != 1:
|
|
raise ValueError('Expected exactly one clone with the supplied ID')
|
|
for meta in clones[0].findall('meta_attributes'):
|
|
if meta.get('id-ref') or meta.find('rule') is not None:
|
|
raise ValueError('Clone metadata uses references/rules; inspect manually')
|
|
matches = []
|
|
for location in root.findall('./configuration/constraints/rsc_location'):
|
|
if location.get('rsc') != clone_id:
|
|
continue
|
|
for rule in location.findall('rule'):
|
|
children = list(rule)
|
|
if (rule.get('role') == 'Promoted' and rule.get('score') == '200'
|
|
and len(children) == 1 and children[0].tag == 'expression'
|
|
and children[0].get('attribute') == '#uname'
|
|
and children[0].get('operation') == 'eq'
|
|
and children[0].get('value') == preferred_node):
|
|
# Preserve unrelated rules if the location has more than one.
|
|
target = location.get('id') if len(list(location)) == 1 else rule.get('id')
|
|
if not target:
|
|
raise ValueError('Matching constraint/rule lacks an ID')
|
|
matches.append(target)
|
|
if len(matches) != 1:
|
|
raise ValueError('Expected exactly one simple Promoted/200/#uname preference; inspect the full CIB manually')
|
|
target_roles = [p.get('value') for m in clones[0].findall('meta_attributes')
|
|
for p in m.findall('nvpair') if p.get('name') == 'target-role']
|
|
if target_roles != ['Promoted']:
|
|
raise ValueError('Expected one direct clone target-role=Promoted; inspect inheritance/rules manually')
|
|
# target-role=Promoted is an intentional production invariant, not a
|
|
# temporary node preference. Never remove or rewrite it here.
|
|
return [shlex.join(['sudo', 'pcs', 'constraint', 'remove', matches[0]])]
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('cib', type=Path)
|
|
parser.add_argument('--clone', required=True)
|
|
parser.add_argument('--preferred-node', required=True)
|
|
parser.add_argument('--lab-validated', action='store_true', help='operator acknowledgement that this candidate passed lab validation')
|
|
args = parser.parse_args()
|
|
if not args.lab_validated:
|
|
parser.error('No rollout commands emitted: lab validation must be completed and acknowledged first')
|
|
try:
|
|
commands = plan(args.cib, args.clone, args.preferred_node)
|
|
except (ValueError, ET.ParseError, OSError) as error:
|
|
print('No commands emitted:', error, file=sys.stderr)
|
|
return 1
|
|
print('# Review against the current CIB before running. No commands have been executed.')
|
|
print('\n'.join(commands))
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|