#!/usr/bin/env python3 """Real-process Maintenance log leases; files stay inside disposable directories.""" import json import os from pathlib import Path import signal import subprocess import sys import tempfile import time import unittest SDP = Path(__file__).resolve().parents[1] CHILD = r''' import json, os, signal, sys, time from pathlib import Path sys.path[:0] = [sys.argv[1], sys.argv[2]] import sdputils import sdp_structured_log as logger root = Path(os.environ['LOGS']) name, mode = sys.argv[3:5] (root / (name + '.requested')).touch() while not (root / 'start').exists(): time.sleep(.01) if mode == 'handled-term': def terminate(signum, frame): print(name + '-terminated', flush=True) raise SystemExit(128 + signum) signal.signal(signal.SIGTERM, terminate) with logger.run('concurrent.py', instance='fixture'), sdputils.maintenance_run('concurrent.py', 'fixture'): print(name + '-start', flush=True) fd = sys.stdout.logfile.fileno() (root / (name + '.ready')).write_text(json.dumps({'inode': os.fstat(fd).st_ino, 'links': os.fstat(fd).st_nlink})) while not (root / (name + '.go')).exists(): time.sleep(.01) print(name + '-finish', flush=True) (root / (name + '.end')).write_text(json.dumps({'inode': os.fstat(fd).st_ino, 'links': os.fstat(fd).st_nlink})) if mode == 'failure': raise RuntimeError(name + '-specific-failure') ''' class MaintenanceLogConcurrency(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory(prefix='sdp-log-lease-test-') self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.children = [] self.addCleanup(self.reap) def reap(self): for child in self.children: if child.poll() is None: child.kill() child.communicate(timeout=5) def wait_file(self, path): deadline = time.monotonic() + 10 while not path.exists(): if time.monotonic() >= deadline: self.fail('Timed out waiting for ' + str(path)) time.sleep(.01) def start(self, name, mode='success', keep='1', simultaneous=False): env = dict(os.environ, LOGS=str(self.root), SDP_RUN_LOG_MAX_BYTES='0', SDP_LOG_FORMAT='jsonl') if keep is None: env.pop('SDP_RUN_LOG_KEEP', None) else: env['SDP_RUN_LOG_KEEP'] = keep child = subprocess.Popen([sys.executable, '-c', CHILD, str(SDP / 'Maintenance'), str(SDP / 'Server/Unix/p4/common/bin'), name, mode], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) self.children.append(child) self.wait_file(self.root / (name + '.requested')) if not simultaneous: (self.root / 'start').touch() self.wait_file(self.root / (name + '.ready')) return child def finish(self, name, child, code=0): (self.root / (name + '.go')).touch() out, err = child.communicate(timeout=10) self.assertEqual(child.returncode, code, out + err) return out, err def logs(self): return {p.name: p.read_text() for p in self.root.iterdir() if p.name == 'concurrent.log' or p.name.startswith('concurrent.log.run-')} def test_overlapping_runs_keep_active_inode_and_all_diagnostics_at_one_and_default_retention(self): for keep in ('1', None): with self.subTest(keep=keep): for path in self.root.iterdir(): path.unlink() owner = self.start('owner', keep=keep) inode = json.loads((self.root / 'owner.ready').read_text())['inode'] # Exceeds the default seven archives while the first run stays active. for index in range(9): name = 'peer' + str(index) peer = self.start(name, keep=keep) self.finish(name, peer) self.assertEqual((self.root / 'concurrent.log').stat().st_ino, inode) self.assertEqual(len(self.logs()), 1) self.finish('owner', owner) self.assertEqual(json.loads((self.root / 'owner.end').read_text())['links'], 1) after = self.start('after', keep=keep) self.finish('after', after) archived, = (content for name, content in self.logs().items() if '.run-' in name) for name in ['owner'] + ['peer' + str(i) for i in range(9)]: self.assertIn(name + '-start', archived) self.assertIn(name + '-finish', archived) events = [json.loads(line) for line in (self.root / 'concurrent.jsonl').read_text().splitlines()] self.assertEqual(len(events), 11) self.assertTrue(all(row['status'] == 'success' for row in events)) self.assertEqual(len({row['run_id'] for row in events}), 11) if keep == '1': latest = self.start('latest', keep=keep) self.finish('latest', latest) self.assertEqual(len(self.logs()), 2) self.assertNotIn('owner-start', '\n'.join(self.logs().values())) def test_simultaneous_preparation_opens_one_shared_inode(self): peers = [(str(index), self.start(str(index), simultaneous=True)) for index in range(8)] (self.root / 'start').touch() for name, _ in peers: self.wait_file(self.root / (name + '.ready')) inodes = {json.loads((self.root / (name + '.ready')).read_text())['inode'] for name, _ in peers} self.assertEqual(len(inodes), 1) for name, child in peers: self.finish(name, child) self.assertEqual(len(self.logs()), 1) text = (self.root / 'concurrent.log').read_text() for name, _ in peers: self.assertIn(name + '-finish', text) def test_failure_and_handled_or_abrupt_term_release_leases_and_preserve_evidence(self): for mode, code in [('failure', 1), ('handled-term', 143), ('abrupt-term', -signal.SIGTERM)]: with self.subTest(mode=mode): for path in self.root.iterdir(): path.unlink() first = self.start('first', mode) peer = self.start('peer') if mode == 'failure': self.finish('first', first, code) else: first.send_signal(signal.SIGTERM) first.communicate(timeout=10) self.assertEqual(first.returncode, code) self.finish('peer', peer) after = self.start('after') self.finish('after', after) archived, = (content for name, content in self.logs().items() if '.run-' in name) self.assertIn('first-start', archived) self.assertIn('peer-finish', archived) if mode == 'failure': self.assertIn('first-specific-failure', archived) if mode == 'handled-term': self.assertIn('first-terminated', archived) self.assertIn('failed (exit 143)', archived) if __name__ == '__main__': unittest.main()