From ba95b9bbc2941c72f5c93fac78d406b6cdf09713 Mon Sep 17 00:00:00 2001 From: Josh Patterson Date: Fri, 2 Oct 2026 10:35:30 -0400 Subject: [PATCH] Address review feedback on so-push-drainer result tracking Result checks walked dispatch records oldest-first with a cap of five lookups per pass, counting records whose push was still running. Five long-running pushes therefore used every slot on every 15s pass and newer, finished pushes were not reported until one cleared. Check the least recently checked records first and back off on pushes that are still running (30s for the first two minutes, then age/4 up to 5 minutes), recording checked_at in the dispatch record. Catch any exception when writing a dispatch record so a failed write cannot skip intent cleanup and re-dispatch the same intents every pass. Log both output streams when no jid is found, and stop logging a traceback when a record has already been removed. Scope the test's salt mock to the drainer import. Run from the repo root, 'salt' resolves to this repo's salt/ directory as a namespace package, so setdefault left it in place and test_load_push_cfg failed. Verified on a 3.4.0 managersearch + sensor: a pushed highstate with soc and telegraf pushes dispatched into it all reported success, with 25 result lookups across the three pushes instead of one per record per pass. --- salt/manager/tools/sbin/so-push-drainer | 44 ++++++++----- .../tools/sbin/so-push-drainer_test.py | 65 ++++++++++++++----- 2 files changed, 78 insertions(+), 31 deletions(-) diff --git a/salt/manager/tools/sbin/so-push-drainer b/salt/manager/tools/sbin/so-push-drainer index 48c3e5b7e..a09d1cb43 100644 --- a/salt/manager/tools/sbin/so-push-drainer +++ b/salt/manager/tools/sbin/so-push-drainer @@ -48,6 +48,7 @@ DISPATCHED_DIR = '/opt/so/state/push_dispatched' HIGHSTATE_SENTINEL = '__highstate__' RESULT_CHECK_DELAY = 30 +RESULT_RECHECK_MAX = 300 RESULT_MAX_AGE = 7200 RESULT_CHECKS_PER_PASS = 5 TEXT_LIMIT = 500 @@ -133,10 +134,11 @@ def _dispatch(actions, log): except Exception: log.exception('dispatch raised') return None - match = JID_RE.search('{}\n{}'.format(result.stderr or '', result.stdout or '')) + output = '{}\n{}'.format(result.stderr or '', result.stdout or '') + match = JID_RE.search(output) if not match: - log.warning('dispatch accepted but no jid found, result will not be tracked: stderr=%s', - _trim(result.stderr)) + log.warning('dispatch accepted but no jid found, result will not be tracked: output=%s', + _trim(output)) return '' log.info('dispatch accepted: jid=%s', match.group(1)) return match.group(1) @@ -155,21 +157,26 @@ def _trim(value): def _unlink(path, log): try: os.unlink(path) + except FileNotFoundError: + pass except OSError: log.exception('failed to remove %s', path) -def _record_dispatch(jid, actions, paths, log): - record = {'jid': jid, 'dispatched_at': time.time(), 'actions': actions, 'paths': paths} - path = os.path.join(DISPATCHED_DIR, '{}.json'.format(jid)) +def _write_record(path, record, log): try: os.makedirs(DISPATCHED_DIR, exist_ok=True) tmp_path = path + '.tmp' with open(tmp_path, 'w') as f: json.dump(record, f) os.rename(tmp_path, path) - except OSError: - log.exception('failed to record dispatch %s', jid) + except Exception: + log.exception('failed to record dispatch %s', record.get('jid')) + + +def _record_dispatch(jid, actions, paths, log): + record = {'jid': jid, 'dispatched_at': time.time(), 'actions': actions, 'paths': paths} + _write_record(os.path.join(DISPATCHED_DIR, '{}.json'.format(jid)), record, log) def _lookup_jid(jid, log): @@ -235,23 +242,30 @@ def _orch_failures(ret): return failures +def _recheck_delay(age): + return min(RESULT_RECHECK_MAX, max(RESULT_CHECK_DELAY, age / 4)) + + def _check_dispatched(log, now): - checked = 0 - for path in sorted(glob.glob(os.path.join(DISPATCHED_DIR, '*.json'))): - if checked >= RESULT_CHECKS_PER_PASS: - break + due = [] + for path in glob.glob(os.path.join(DISPATCHED_DIR, '*.json')): record = _read_intent(path, log) if not isinstance(record, dict) or not record.get('jid'): _unlink(path, log) continue age = now - record.get('dispatched_at', 0) - if age < RESULT_CHECK_DELAY: - continue - checked += 1 + last_check = record.get('checked_at', record.get('dispatched_at', 0)) + if now - last_check >= _recheck_delay(age): + due.append((last_check, path, record, age)) + # Least recently checked first, so pushes that are still running can't starve finished ones. + for _, path, record, age in sorted(due, key=lambda item: item[:2])[:RESULT_CHECKS_PER_PASS]: jid = record['jid'] try: if _report_result(record, age, log): _unlink(path, log) + else: + record['checked_at'] = now + _write_record(path, record, log) except Exception: # Drop the record so one unreadable result can't fail every pass ahead of the drain. log.exception('cannot evaluate result for jid=%s; no longer tracking', jid) diff --git a/salt/manager/tools/sbin/so-push-drainer_test.py b/salt/manager/tools/sbin/so-push-drainer_test.py index 0c415710e..002011810 100644 --- a/salt/manager/tools/sbin/so-push-drainer_test.py +++ b/salt/manager/tools/sbin/so-push-drainer_test.py @@ -16,17 +16,17 @@ import unittest from importlib.machinery import SourceFileLoader from unittest.mock import MagicMock, patch -# salt is not installed where these tests run; the drainer only needs salt.client.Caller. -_salt = MagicMock() -sys.modules.setdefault('salt', _salt) -sys.modules.setdefault('salt.client', _salt.client) - HERE = os.path.dirname(os.path.abspath(__file__)) SCRIPT = os.path.join(HERE, 'so-push-drainer') _loader = SourceFileLoader('so_push_drainer', SCRIPT) _spec = importlib.util.spec_from_loader('so_push_drainer', _loader) drainer = importlib.util.module_from_spec(_spec) -_loader.exec_module(drainer) + +# salt is not installed where these tests run; the drainer only needs salt.client.Caller. +# Mocked only while the drainer loads: run from the repo root, 'salt' is this repo's salt/ directory. +_salt = MagicMock() +with patch.dict(sys.modules, {'salt': _salt, 'salt.client': _salt.client}): + _loader.exec_module(drainer) MASTER = 'manager.localdomain_master' JID = '20260930171554259426' @@ -181,8 +181,10 @@ class TestHelpers(DrainerTestCase): 'salt.exceptions.AuthenticationError: Authentication error occurred.') self.assertEqual(drainer._trim('line one\n line two\n'), 'line one line two') - def test_unlink_missing_logs(self): + def test_unlink(self): drainer._unlink(os.path.join(self.tmpdir, 'missing'), self.log) + self.log.exception.assert_not_called() + drainer._unlink(self.tmpdir, self.log) self.log.exception.assert_called_once() @@ -205,9 +207,9 @@ class TestDispatch(DrainerTestCase): self.assertEqual(jid, JID) def test_no_jid(self): - jid, _ = self.run_dispatch(return_value=MagicMock(stdout='', stderr=None)) + jid, _ = self.run_dispatch(return_value=MagicMock(stdout='unexpected output', stderr=None)) self.assertEqual(jid, '') - self.log.warning.assert_called_once() + self.assertIn('output=unexpected output', self.logged('warning')) def test_failures_return_none(self): for exc in (subprocess.CalledProcessError(1, 'salt-run', 'out', 'err'), @@ -224,10 +226,12 @@ class TestDispatch(DrainerTestCase): self.assertEqual(record['paths'], ['audit:soc.config.licenseKey']) self.assertIn('dispatched_at', record) - def test_record_dispatch_oserror(self): + def test_record_dispatch_errors(self): with patch.object(drainer.os, 'makedirs', side_effect=OSError('ro')): drainer._record_dispatch(JID, [], [], self.log) - self.log.exception.assert_called_once() + drainer._record_dispatch(JID, [object()], [], self.log) + self.assertEqual(self.log.exception.call_count, 2) + self.assertFalse(os.path.exists(os.path.join(self.dispatched, JID + '.json'))) class TestResults(DrainerTestCase): @@ -311,7 +315,8 @@ class TestResults(DrainerTestCase): drainer._check_dispatched(self.log, now) self.assertTrue(os.path.exists(young)) - self.assertTrue(os.path.exists(paths['3_pending'])) + with open(paths['3_pending']) as f: + self.assertEqual(json.load(f)['checked_at'], now) for jid in ('1_failed', '2_ok', '4_expired'): self.assertFalse(os.path.exists(paths[jid]), jid) self.assertFalse(os.path.exists(bad)) @@ -339,13 +344,41 @@ class TestResults(DrainerTestCase): self.assertIn('jid=1_bad', self.log.exception.call_args[0][0] % self.log.exception.call_args[0][1:]) self.assertIn('push succeeded jid=2_ok', self.logged('info')) - def test_check_dispatched_limit(self): + def test_recheck_delay(self): + self.assertEqual(drainer._recheck_delay(10), drainer.RESULT_CHECK_DELAY) + self.assertEqual(drainer._recheck_delay(400), 100) + self.assertEqual(drainer._recheck_delay(drainer.RESULT_MAX_AGE), drainer.RESULT_RECHECK_MAX) + + def test_check_dispatched_limit_rotates(self): now = time.time() - for i in range(drainer.RESULT_CHECKS_PER_PASS + 2): - self.record('{:02d}'.format(i), 60, now) + limit = drainer.RESULT_CHECKS_PER_PASS + jids = ['{:02d}'.format(i) for i in range(limit + 2)] + for jid in jids: + self.record(jid, 60, now) with patch.object(drainer, '_lookup_jid', return_value={}) as lookup: drainer._check_dispatched(self.log, now) - self.assertEqual(lookup.call_count, drainer.RESULT_CHECKS_PER_PASS) + self.assertEqual([c.args[0] for c in lookup.call_args_list], jids[:limit]) + lookup.reset_mock() + drainer._check_dispatched(self.log, now + 40) + self.assertEqual([c.args[0] for c in lookup.call_args_list], jids[limit:] + jids[:limit - 2]) + + def test_check_dispatched_not_blocked_by_running(self): + now = time.time() + for i in range(drainer.RESULT_CHECKS_PER_PASS): + self.record('1_running{}'.format(i), 600, now) + done = self.record('2_done', 60, now) + + def lookup(jid, log): + return SUCCESS_RET if jid == '2_done' else {} + + with patch.object(drainer, '_lookup_jid', side_effect=lookup) as lookup_jid: + drainer._check_dispatched(self.log, now) + self.assertTrue(os.path.exists(done)) + lookup_jid.reset_mock() + drainer._check_dispatched(self.log, now + 15) + self.assertEqual([c.args[0] for c in lookup_jid.call_args_list], ['2_done']) + self.assertFalse(os.path.exists(done)) + self.assertIn('push succeeded jid=2_done', self.logged('info')) class TestMain(DrainerTestCase):