CSEC_PUBLIC/securityonion
1
0 Fork 0
mirror of https://github.com/Security-Onion-Solutions/securityonion.git synced 2026-10-08 15:25:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

Address review feedback on so-push-drainer result tracking

Browse Source 
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.
This commit is contained in:
Josh Patterson committed 2026-10-02 10:35:30 -04:00
1 parent 8de8ba811a
commit ba95b9bbc2
2 files changed
+78 -31
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

salt/manager/tools/sbin/so-push-drainer
+29 -15
View File
@@ -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)
salt/manager/tools/sbin/so-push-drainer_test.py
+49 -16
View File
@@ -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):