Postgres: pg_cron over the local socket, so_maint_telegraf role, repair and soup 3.5.0

Merge pull request #16389 from Security-Onion-Solutions/reyesj2-patch-1
update elastic-agent artifact repo directory
2026-11-13 08:00:07 +01:00 · 2026-11-12 19:07:57 -05:00 · 2026-11-11 14:32:59 -06:00 · 2026-11-10 14:53:30 -05:00 · 2026-11-10 08:26:19 -05:00 · 2026-11-09 12:32:06 -06:00
19 changed files with 543 additions and 1047 deletions
@@ -131,8 +131,6 @@ def beacon(config):  # noqa: C901
            'setting_id': setting_id,
            'node_id': node_id,
        })
-        log.info('postgres_pillar_beacon: audit_settings id=%d setting_id=%s node_id=%s',
-                 row_id, setting_id, node_id)
        if row_id > max_id:
            max_id = row_id

@@ -16,12 +16,9 @@ For each intent file under /opt/so/state/push_pending/*.json whose last_touch
 is older than debounce_seconds, this script:
  * concatenates the actions lists from every ready intent
  * dedupes by (state or __highstate__, tgt, tgt_type)
-  * drops actions whose target matches no accepted minion (e.g. a role with no nodes yet)
  * dispatches a single `salt-run state.orchestrate orch.push_batch --async`
    with the deduped actions list passed as pillar kwargs
  * deletes the contributed intent files on successful dispatch
-  * records the orchestration jid under /opt/so/state/push_dispatched and, on
-    later passes, looks up its result and logs success or per-minion failures

 Reactor sls files (push_files, push_pillar) write intents
 but never dispatch directly
@@ -33,36 +30,18 @@ import json
 import logging
 import logging.handlers
 import os
-import re
 import subprocess
 import sys
 import time

 import salt.client
-import salt.config
-import salt.utils.minions

 PENDING_DIR = '/opt/so/state/push_pending'
 LOCK_FILE = os.path.join(PENDING_DIR, '.lock')
 LOG_FILE = '/opt/so/log/salt/so-push-drainer.log'
-MASTER_CONFIG = '/etc/salt/master'
-
-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
-
-# Lead-in salt puts on the comment of an orchestration step that raised.
-STEP_RAISED = 'An exception occurred in this state:'
-
-# salt-run --async reports the jid only in a log line (stderr by default).
-JID_RE = re.compile(r'salt/run/(\d{20})')
-

 def _make_logger():
    logger = logging.getLogger('so-push-drainer')
@@ -117,29 +96,6 @@ def _dedupe_actions(actions):
    return deduped


-def _targeted(actions, log):
-    """Drops actions whose target matches no accepted minion (e.g. a role with no nodes yet), which
-    the orchestration would otherwise report as failed with 'No minions responded'."""
-    try:
-        ckminions = salt.utils.minions.CkMinions(salt.config.master_config(MASTER_CONFIG))
-    except Exception:
-        log.exception('cannot load master config; dispatching without checking targets')
-        return actions
-    targeted = []
-    for action in actions:
-        try:
-            matched = ckminions.check_minions(action['tgt'], action.get('tgt_type', 'compound')).get('minions')
-        except Exception:
-            log.exception('cannot resolve tgt=%s; dispatching it anyway', action.get('tgt'))
-            matched = True
-        if matched:
-            targeted.append(action)
-        else:
-            log.info('skipping %s: no minions match tgt=%s',
-                     'highstate' if action.get('highstate') else action.get('state'), action.get('tgt'))
-    return targeted
-
-
 def _dispatch(actions, log):
    pillar_arg = json.dumps({'actions': actions})
    cmd = [
@@ -157,189 +113,14 @@ def _dispatch(actions, log):
    except subprocess.CalledProcessError as exc:
        log.error('dispatch failed (rc=%s): stdout=%s stderr=%s',
                  exc.returncode, exc.stdout, exc.stderr)
-        return None
+        return False
    except subprocess.TimeoutExpired:
        log.error('dispatch timed out after 60s')
-        return None
+        return False
    except Exception:
        log.exception('dispatch raised')
-        return None
-    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: output=%s',
-                    _trim(output))
-        return ''
-    log.info('dispatch accepted: jid=%s', match.group(1))
-    return match.group(1)
-
-
-def _trim(value):
-    text = value if isinstance(value, str) else json.dumps(value, default=str)
-    lines = [line.strip() for line in text.splitlines() if line.strip()]
-    if 'Traceback (most recent call last):' in text:
-        # Keep the lead-in and the raised exception; the frames are noise in a log line.
-        lines = [text.split('Traceback (most recent call last):', 1)[0].strip(), lines[-1]]
-    text = ' '.join(line for line in lines if line)
-    return text if len(text) <= TEXT_LIMIT else text[:TEXT_LIMIT] + '...'
-
-
-def _unlink(path, log):
-    try:
-        os.unlink(path)
-    except FileNotFoundError:
-        pass
-    except OSError:
-        log.exception('failed to remove %s', path)
-
-
-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 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):
-    """Returns the job cache entry for jid, {} while it is still running, or None on error."""
-    cmd = ['salt-run', 'jobs.lookup_jid', jid, '--out=json']
-    try:
-        result = subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=60)
-        return json.loads(result.stdout or '{}')
-    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, ValueError) as exc:
-        log.warning('lookup of jid %s failed: %s', jid, exc)
-        return None
-
-
-def _minion_failure(minion_ret):
-    if isinstance(minion_ret, dict):
-        return '; '.join(
-            '{}: {}'.format(state.get('__id__', state_key), _trim(state.get('comment', '')))
-            for state_key, state in minion_ret.items()
-            if isinstance(state, dict) and state.get('result') is False
-        )
-    # A state run rejected before it starts (e.g. another state run is in
-    # progress) returns a list of error strings instead of state results.
-    if isinstance(minion_ret, (list, str)):
-        return _trim(minion_ret)
-    return ''
-
-
-def _step_failures(step):
-    if not isinstance(step, dict) or step.get('result') is not False:
-        return []
-    failures = ['{}: {}'.format(step.get('__id__', step.get('name')), _trim(step.get('comment', '')))]
-    changes = step.get('changes')
-    minion_rets = changes.get('ret') if isinstance(changes, dict) else None
-    if isinstance(minion_rets, dict):
-        for minion, minion_ret in minion_rets.items():
-            text = _minion_failure(minion_ret)
-            if text:
-                failures.append('{}: {}'.format(minion, text))
-    return failures
-
-
-def _orch_failures(ret):
-    if not isinstance(ret, dict):
-        return [_trim(ret)]
-    failures = []
-    for job in ret.values():
-        if not isinstance(job, dict):
-            continue
-        job_ret = job.get('return')
-        data = job_ret.get('data') if isinstance(job_ret, dict) else {}
-        if not isinstance(data, dict):
-            if data:
-                failures.append(_trim(data))
-            data = {}
-        for steps in data.values():
-            if not isinstance(steps, dict):
-                failures.append(_trim(steps))
-                continue
-            for step in steps.values():
-                failures.extend(_step_failures(step))
-        if job.get('success') is False and not failures:
-            failures.append('orchestration reported failure: {}'.format(_trim(job.get('return'))))
-    return failures
-
-
-def _failed_steps(ret):
-    steps = []
-    for job in ret.values() if isinstance(ret, dict) else []:
-        job_ret = job.get('return') if isinstance(job, dict) else None
-        data = job_ret.get('data') if isinstance(job_ret, dict) else None
-        for group in data.values() if isinstance(data, dict) else []:
-            if isinstance(group, dict):
-                steps.extend(step for step in group.values() if isinstance(step, dict) and step.get('result') is False)
-    return steps
-
-
-def _result_unknown(ret):
-    # A step that raised (e.g. salt-master restarted while it waited on a queued state run)
-    # never collected the minion's return, so the state run may still have completed.
-    steps = _failed_steps(ret)
-    return bool(steps) and all(str(step.get('comment', '')).startswith(STEP_RAISED) for step in steps)
-
-
-def _recheck_delay(age):
-    return min(RESULT_RECHECK_MAX, max(RESULT_CHECK_DELAY, age / 4))
-
-
-def _check_dispatched(log, now):
-    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)
-        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)
-            _unlink(path, log)
-
-
-def _report_result(record, age, log):
-    """Logs the outcome of a dispatched push. Returns True once the record is finished with."""
-    jid = record['jid']
-    paths = record.get('paths', [])
-    ret = _lookup_jid(jid, log)
-    if not ret:
-        if age > RESULT_MAX_AGE:
-            log.warning('no result for jid=%s after %ds, no longer tracking; paths=%s', jid, age, paths)
-            return True
        return False
-    failures = _orch_failures(ret)
-    if failures and _result_unknown(ret):
-        log.warning('push result unknown jid=%s paths=%s; the orchestration lost track of the state run, which may '
-                    'still have completed (if not, the change will be applied at the next scheduled highstate): %s',
-                    jid, paths, ' | '.join(failures))
-    elif failures:
-        log.error('push failed jid=%s paths=%s; change will be applied at the next scheduled highstate: %s',
-                  jid, paths, ' | '.join(failures))
-    else:
-        log.info('push succeeded jid=%s paths=%s', jid, paths)
+    log.info('dispatch accepted: %s', (result.stdout or '').strip())
    return True


@@ -362,9 +143,6 @@ def main():

    debounce_seconds = int(push.get('debounce_seconds', 30))

-    # Outside the lock: lookups are slow and the reactors take the same lock.
-    _check_dispatched(log, time.time())
-
    os.makedirs(PENDING_DIR, exist_ok=True)
    lock_fd = os.open(LOCK_FILE, os.O_CREAT | os.O_RDWR, 0o644)
    try:
@@ -430,20 +208,10 @@ def main():
            len(ready), len(deduped), len(combined_actions),
            debounce_duration, all_paths[:20],
        )
-        for action in deduped:
-            log.info('action: %s tgt=%s', 'highstate' if action.get('highstate') else action.get('state'),
-                     action.get('tgt'))

-        targeted = _targeted(deduped, log)
-        if targeted:
-            jid = _dispatch(targeted, log)
-            if jid is None:
-                log.warning('dispatch failed; leaving intent files in place for retry')
-                return 1
-            if jid:
-                _record_dispatch(jid, targeted, all_paths[:20], log)
-        else:
-            log.info('no action matches a minion; nothing to dispatch')
+        if not _dispatch(deduped, log):
+            log.warning('dispatch failed; leaving intent files in place for retry')
+            return 1

        for path, _ in ready:
            try:
@@ -1,569 +0,0 @@
-# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
-# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
-# https://securityonion.net/license; you may not use this file except in compliance with the
-# Elastic License 2.0.
-
-import importlib.util
-import json
-import logging
-import os
-import shutil
-import subprocess
-import sys
-import tempfile
-import time
-import unittest
-from importlib.machinery import SourceFileLoader
-from unittest.mock import MagicMock, patch
-
-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)
-
-# salt is not installed where these tests run; the drainer only needs salt.client.Caller and CkMinions.
-# 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, 'salt.config': _salt.config,
-                              'salt.utils': _salt.utils, 'salt.utils.minions': _salt.utils.minions}):
-    _loader.exec_module(drainer)
-
-MASTER = 'manager.localdomain_master'
-JID = '20260930171554259426'
-ASYNC_STDERR = ('[WARNING ] Running in asynchronous mode. Results of this execution may be collected '
-                'by attaching to the master event bus or by examining the master job cache, if '
-                'configured. This execution is running under tag salt/run/{}\n'.format(JID))
-CONFLICT = ('The function "state.sls" is running as PID 372218 and was started at '
-            '2026, Sep 30 17:15:40.466233 with jid 20260930171540466233')
-
-
-def _orch_ret(steps, success=True):
-    return {MASTER: {
-        'fun': 'runner.state.orchestrate',
-        'jid': JID,
-        'return': {'data': {MASTER: steps}, 'outputter': 'highstate', 'retcode': 0 if success else 1},
-        'success': success,
-    }}
-
-
-REFRESH_STEP = {
-    'salt_|-refresh_pillar_1_|-saltutil.refresh_pillar_|-function': {
-        '__id__': 'refresh_pillar_1', 'result': True,
-        'changes': {'ret': {'manager_standalone': True}},
-        'comment': 'Function ran successfully.',
-    },
-}
-
-CONFLICT_RET = _orch_ret(dict(REFRESH_STEP, **{
-    'salt_|-apply_soc_1_|-apply_soc_1_|-state': {
-        '__id__': 'apply_soc_1', 'result': False,
-        'changes': {'out': 'highstate', 'ret': {'manager_standalone': [CONFLICT]}},
-        'comment': 'Run failed on minions: manager_standalone',
-    },
-}), success=False)
-
-STATE_FAIL_RET = _orch_ret({
-    'salt_|-apply_hydra_1_|-apply_hydra_1_|-state': {
-        '__id__': 'apply_hydra_1', 'result': False,
-        'changes': {'out': 'highstate', 'ret': {'manager_standalone': {
-            'test_|-no_license_|-no_license_|-fail_without_changes': {
-                '__id__': 'hydra.enabled_no_license_detected', 'result': False,
-                'comment': 'This is a feature supported only for customers with a valid license.',
-            },
-            'file_|-hydra_conf_|-/opt/so/conf/hydra_|-managed': {'result': True, 'comment': 'ok'},
-        }}},
-        'comment': 'Run failed on minions: manager_standalone',
-    },
-}, success=False)
-
-RAISED = ('An exception occurred in this state: Traceback (most recent call last):\n'
-          '  File "salt/client/__init__.py", line 1934, in pub\n'
-          '    raise AuthenticationError(err_msg)\n'
-          'salt.exceptions.AuthenticationError: Authentication error occurred.\n')
-
-RAISED_RET = _orch_ret(dict(REFRESH_STEP, **{
-    'salt_|-apply_hydra_1_|-apply_hydra_1_|-state': {
-        '__id__': 'apply_hydra_1', 'result': False, 'changes': {}, 'comment': RAISED,
-    },
-}), success=False)
-
-SUCCESS_RET = _orch_ret(dict(REFRESH_STEP, **{
-    'salt_|-apply_telegraf_1_|-apply_telegraf_1_|-state': {
-        '__id__': 'apply_telegraf_1', 'result': True,
-        'changes': {'out': 'highstate', 'ret': {'manager_standalone': {
-            'file_|-tgrafconf_|-/opt/so/conf/telegraf/etc/telegraf.conf_|-managed': {'result': True},
-        }}},
-        'comment': 'States ran successfully.',
-    },
-}))
-
-
-class DrainerTestCase(unittest.TestCase):
-
-    def setUp(self):
-        self.tmpdir = tempfile.mkdtemp()
-        self.pending = os.path.join(self.tmpdir, 'push_pending')
-        self.dispatched = os.path.join(self.tmpdir, 'push_dispatched')
-        os.makedirs(self.pending)
-        for name, value in (
-            ('PENDING_DIR', self.pending),
-            ('LOCK_FILE', os.path.join(self.pending, '.lock')),
-            ('DISPATCHED_DIR', self.dispatched),
-            ('LOG_FILE', os.path.join(self.tmpdir, 'log', 'so-push-drainer.log')),
-        ):
-            patcher = patch.object(drainer, name, value)
-            patcher.start()
-            self.addCleanup(patcher.stop)
-        self.log = MagicMock()
-
-    def tearDown(self):
-        shutil.rmtree(self.tmpdir, ignore_errors=True)
-
-    def write_json(self, directory, name, data):
-        os.makedirs(directory, exist_ok=True)
-        path = os.path.join(directory, name)
-        with open(path, 'w') as f:
-            if isinstance(data, str):
-                f.write(data)
-            else:
-                json.dump(data, f)
-        return path
-
-    def logged(self, level):
-        return ' '.join(c.args[0] % c.args[1:] for c in getattr(self.log, level).call_args_list)
-
-
-class TestHelpers(DrainerTestCase):
-
-    def test_make_logger_adds_handler_once(self):
-        logger = logging.getLogger('so-push-drainer')
-
-        def close_handlers():
-            for handler in logger.handlers:
-                handler.close()
-            logger.handlers.clear()
-
-        self.addCleanup(close_handlers)
-        logger.handlers.clear()
-        self.assertIs(drainer._make_logger(), logger)
-        drainer._make_logger()
-        self.assertEqual(len(logger.handlers), 1)
-        self.assertTrue(os.path.isdir(os.path.dirname(drainer.LOG_FILE)))
-
-    def test_load_push_cfg(self):
-        with patch.object(drainer.salt.client, 'Caller') as caller:
-            caller.return_value.cmd.return_value = {'enabled': False}
-            self.assertEqual(drainer._load_push_cfg(), {'enabled': False})
-            caller.return_value.cmd.return_value = 'garbage'
-            self.assertEqual(drainer._load_push_cfg(), {})
-
-    def test_read_intent(self):
-        good = self.write_json(self.pending, 'good.json', {'a': 1})
-        bad = self.write_json(self.pending, 'bad.json', '{nope')
-        self.assertEqual(drainer._read_intent(good, self.log), {'a': 1})
-        self.assertIsNone(drainer._read_intent(bad, self.log))
-        with patch('builtins.open', side_effect=RuntimeError('boom')):
-            self.assertIsNone(drainer._read_intent(good, self.log))
-        self.log.exception.assert_called_once()
-
-    def test_dedupe_actions(self):
-        actions = [
-            'not a dict',
-            {'state': 'soc'},
-            {'state': 'soc', 'tgt': '*'},
-            {'state': 'soc', 'tgt': '*', 'tgt_type': 'compound'},
-            {'highstate': True, 'tgt': '*'},
-            {'state': 'soc', 'tgt': 'node1', 'tgt_type': 'glob'},
-        ]
-        self.assertEqual(drainer._dedupe_actions(actions), [actions[2], actions[4], actions[5]])
-
-    def test_trim(self):
-        self.assertEqual(drainer._trim('  text \n'), 'text')
-        self.assertEqual(drainer._trim(['a']), '["a"]')
-        self.assertEqual(drainer._trim(None), 'null')
-        self.assertEqual(drainer._trim('x' * 600), 'x' * drainer.TEXT_LIMIT + '...')
-
-    def test_trim_traceback(self):
-        comment = ('An exception occurred in this state: Traceback (most recent call last):\n'
-                   '  File "salt/client/__init__.py", line 1934, in pub\n'
-                   '    raise AuthenticationError(err_msg)\n'
-                   'salt.exceptions.AuthenticationError: Authentication error occurred.\n')
-        self.assertEqual(drainer._trim(comment), 'An exception occurred in this state: '
-                         'salt.exceptions.AuthenticationError: Authentication error occurred.')
-        self.assertEqual(drainer._trim('line one\n  line two\n'), 'line one line two')
-
-    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()
-
-
-class TestDispatch(DrainerTestCase):
-
-    def run_dispatch(self, **kwargs):
-        with patch.object(drainer.subprocess, 'run', **kwargs) as run:
-            jid = drainer._dispatch([{'state': 'soc', 'tgt': '*'}], self.log)
-        return jid, run
-
-    def test_jid_parsed_from_stderr(self):
-        jid, run = self.run_dispatch(return_value=MagicMock(stdout='', stderr=ASYNC_STDERR))
-        self.assertEqual(jid, JID)
-        cmd = run.call_args[0][0]
-        self.assertEqual(cmd[:3], ['salt-run', 'state.orchestrate', 'orch.push_batch'])
-        self.assertIn('--async', cmd)
-
-    def test_jid_parsed_from_stdout(self):
-        jid, _ = self.run_dispatch(return_value=MagicMock(stdout=ASYNC_STDERR, stderr=None))
-        self.assertEqual(jid, JID)
-
-    def test_no_jid(self):
-        jid, _ = self.run_dispatch(return_value=MagicMock(stdout='unexpected output', stderr=None))
-        self.assertEqual(jid, '')
-        self.assertIn('output=unexpected output', self.logged('warning'))
-
-    def test_failures_return_none(self):
-        for exc in (subprocess.CalledProcessError(1, 'salt-run', 'out', 'err'),
-                    subprocess.TimeoutExpired('salt-run', 60),
-                    RuntimeError('boom')):
-            jid, _ = self.run_dispatch(side_effect=exc)
-            self.assertIsNone(jid)
-
-    def test_record_dispatch(self):
-        drainer._record_dispatch(JID, [{'state': 'soc'}], ['audit:soc.config.licenseKey'], self.log)
-        with open(os.path.join(self.dispatched, JID + '.json')) as f:
-            record = json.load(f)
-        self.assertEqual(record['jid'], JID)
-        self.assertEqual(record['paths'], ['audit:soc.config.licenseKey'])
-        self.assertIn('dispatched_at', record)
-
-    def test_record_dispatch_errors(self):
-        with patch.object(drainer.os, 'makedirs', side_effect=OSError('ro')):
-            drainer._record_dispatch(JID, [], [], self.log)
-        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):
-
-    def test_lookup_jid(self):
-        with patch.object(drainer.subprocess, 'run') as run:
-            run.return_value = MagicMock(stdout=json.dumps(SUCCESS_RET))
-            self.assertEqual(drainer._lookup_jid(JID, self.log), SUCCESS_RET)
-            self.assertEqual(run.call_args[0][0], ['salt-run', 'jobs.lookup_jid', JID, '--out=json'])
-            run.return_value = MagicMock(stdout='')
-            self.assertEqual(drainer._lookup_jid(JID, self.log), {})
-            run.return_value = MagicMock(stdout='not json')
-            self.assertIsNone(drainer._lookup_jid(JID, self.log))
-            run.side_effect = subprocess.TimeoutExpired('salt-run', 60)
-            self.assertIsNone(drainer._lookup_jid(JID, self.log))
-
-    def test_minion_failure_shapes(self):
-        self.assertEqual(drainer._minion_failure([CONFLICT]), json.dumps([CONFLICT]))
-        self.assertEqual(drainer._minion_failure('Rendering SLS failed'), 'Rendering SLS failed')
-        self.assertEqual(drainer._minion_failure(True), '')
-        self.assertEqual(drainer._minion_failure({'a': {'result': True}}), '')
-
-    def test_orch_failures_conflict(self):
-        failures = drainer._orch_failures(CONFLICT_RET)
-        self.assertEqual(failures[0], 'apply_soc_1: Run failed on minions: manager_standalone')
-        self.assertIn('manager_standalone', failures[1])
-        self.assertIn('is running as PID 372218', failures[1])
-        self.assertEqual(len(failures), 2)
-
-    def test_orch_failures_failed_state(self):
-        failures = drainer._orch_failures(STATE_FAIL_RET)
-        self.assertEqual(len(failures), 2)
-        self.assertIn('hydra.enabled_no_license_detected: This is a feature', failures[1])
-        self.assertNotIn('hydra_conf', failures[1])
-
-    def test_orch_failures_success(self):
-        self.assertEqual(drainer._orch_failures(SUCCESS_RET), [])
-
-    def test_orch_failures_render_error(self):
-        ret = {MASTER: {'return': {'data': {MASTER: ['Rendering SLS failed']}}, 'success': False}}
-        self.assertEqual(drainer._orch_failures(ret), ['["Rendering SLS failed"]'])
-
-    def test_orch_failures_not_a_dict(self):
-        self.assertEqual(drainer._orch_failures(['No minions matched']), ['["No minions matched"]'])
-        self.assertEqual(drainer._orch_failures('Runner error'), ['Runner error'])
-
-    def test_orch_failures_data_not_a_dict(self):
-        ret = {MASTER: {'return': {'data': ["Rendering SLS 'orch.push_batch' failed"]}, 'success': False}}
-        self.assertEqual(drainer._orch_failures(ret), ['["Rendering SLS \'orch.push_batch\' failed"]'])
-
-    def test_orch_failures_odd_changes(self):
-        for changes in ('Run failed', {'ret': ['manager_standalone']}):
-            ret = _orch_ret({'salt_|-apply_soc_1_|-apply_soc_1_|-state': {
-                '__id__': 'apply_soc_1', 'result': False, 'changes': changes, 'comment': 'Run failed on minions',
-            }}, success=False)
-            self.assertEqual(drainer._orch_failures(ret), ['apply_soc_1: Run failed on minions'])
-
-    def test_orch_failures_unparsed(self):
-        self.assertEqual(drainer._orch_failures({MASTER: 'odd'}), [])
-        ret = {MASTER: {'return': 'Exception occurred', 'success': False}}
-        self.assertEqual(drainer._orch_failures(ret), ['orchestration reported failure: Exception occurred'])
-
-    def test_result_unknown(self):
-        self.assertTrue(drainer._result_unknown(RAISED_RET))
-        for ret in (CONFLICT_RET, STATE_FAIL_RET, SUCCESS_RET, ['No minions matched'], {MASTER: 'odd'},
-                    {MASTER: {'return': {'data': {MASTER: ['Rendering SLS failed']}}, 'success': False}}):
-            self.assertFalse(drainer._result_unknown(ret), ret)
-        mixed = _orch_ret(dict(RAISED_RET[MASTER]['return']['data'][MASTER],
-                               **CONFLICT_RET[MASTER]['return']['data'][MASTER]), success=False)
-        self.assertFalse(drainer._result_unknown(mixed))
-
-    def record(self, jid, age, now):
-        return self.write_json(self.dispatched, jid + '.json', {
-            'jid': jid, 'dispatched_at': now - age, 'actions': [], 'paths': ['audit:' + jid],
-        })
-
-    def test_check_dispatched(self):
-        now = time.time()
-        results = {
-            '1_failed': CONFLICT_RET,
-            '2_ok': SUCCESS_RET,
-            '3_pending': {},
-            '4_expired': None,
-            '6_unknown': RAISED_RET,
-        }
-        young = self.record('0_young', 5, now)
-        paths = {jid: self.record(jid, 60, now) for jid in results}
-        paths['4_expired'] = self.record('4_expired', drainer.RESULT_MAX_AGE + 1, now)
-        bad = self.write_json(self.dispatched, '5_bad.json', '{nope')
-        with patch.object(drainer, '_lookup_jid', side_effect=lambda jid, log: results[jid]):
-            drainer._check_dispatched(self.log, now)
-
-        self.assertTrue(os.path.exists(young))
-        with open(paths['3_pending']) as f:
-            self.assertEqual(json.load(f)['checked_at'], now)
-        for jid in ('1_failed', '2_ok', '4_expired', '6_unknown'):
-            self.assertFalse(os.path.exists(paths[jid]), jid)
-        self.assertFalse(os.path.exists(bad))
-        self.assertIn('push failed jid=1_failed', self.logged('error'))
-        self.assertIn('is running as PID 372218', self.logged('error'))
-        self.assertIn('push succeeded jid=2_ok', self.logged('info'))
-        self.assertIn('no result for jid=4_expired', self.logged('warning'))
-        self.assertIn('push result unknown jid=6_unknown', self.logged('warning'))
-        self.assertIn('AuthenticationError: Authentication error occurred.', self.logged('warning'))
-        self.assertNotIn('6_unknown', self.logged('error'))
-
-    def test_check_dispatched_survives_bad_result(self):
-        now = time.time()
-        bad = self.record('1_bad', 60, now)
-        good = self.record('2_ok', 60, now)
-
-        def orch_failures(ret):
-            if ret == 'boom':
-                raise ValueError('unexpected shape')
-            return []
-
-        with patch.object(drainer, '_lookup_jid', side_effect=lambda jid, log: 'boom' if jid == '1_bad' else SUCCESS_RET), \
-                patch.object(drainer, '_orch_failures', side_effect=orch_failures):
-            drainer._check_dispatched(self.log, now)
-        self.assertFalse(os.path.exists(bad))
-        self.assertFalse(os.path.exists(good))
-        self.log.exception.assert_called_once()
-        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_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()
-        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([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 TestTargeted(DrainerTestCase):
-
-    def setUp(self):
-        super().setUp()
-        self.ckminions = MagicMock()
-        patcher = patch.object(drainer.salt.utils.minions, 'CkMinions', return_value=self.ckminions)
-        self.CkMinions = patcher.start()
-        self.addCleanup(patcher.stop)
-
-    def test_drops_unmatched(self):
-        matches = {'G@role:so-sensor': ['sensor_sensor'], 'manager_manager': ['manager_manager']}
-        self.ckminions.check_minions.side_effect = lambda tgt, tgt_type: {'minions': matches.get(tgt, [])}
-        actions = [
-            {'state': 'strelka', 'tgt': 'G@role:so-sensor'},
-            {'state': 'suricata', 'tgt': 'G@role:so-heavynode'},
-            {'highstate': True, 'tgt': 'G@role:so-idh'},
-            {'state': 'soc', 'tgt': 'manager_manager', 'tgt_type': 'list'},
-        ]
-        self.assertEqual(drainer._targeted(actions, self.log), [actions[0], actions[3]])
-        self.CkMinions.assert_called_once()
-        self.assertEqual([c.args for c in self.ckminions.check_minions.call_args_list], [
-            ('G@role:so-sensor', 'compound'), ('G@role:so-heavynode', 'compound'),
-            ('G@role:so-idh', 'compound'), ('manager_manager', 'list'),
-        ])
-        info = self.logged('info')
-        self.assertIn('skipping suricata: no minions match tgt=G@role:so-heavynode', info)
-        self.assertIn('skipping highstate: no minions match tgt=G@role:so-idh', info)
-        self.assertNotIn('strelka', info)
-
-    def test_master_config_error_keeps_actions(self):
-        actions = [{'state': 'soc', 'tgt': '*'}]
-        with patch.object(drainer.salt.config, 'master_config', side_effect=OSError('no config')):
-            self.assertEqual(drainer._targeted(actions, self.log), actions)
-        self.log.exception.assert_called_once()
-        self.CkMinions.assert_not_called()
-
-    def test_check_error_keeps_action(self):
-        actions = [{'state': 'soc', 'tgt': 'G@role:'}]
-        self.ckminions.check_minions.side_effect = RuntimeError('bad target')
-        self.assertEqual(drainer._targeted(actions, self.log), actions)
-        self.log.exception.assert_called_once()
-
-
-class TestMain(DrainerTestCase):
-
-    def setUp(self):
-        super().setUp()
-        self.cfg = {'enabled': True, 'debounce_seconds': 30}
-        for name, kwargs in (
-            ('_make_logger', {'return_value': self.log}),
-            ('_load_push_cfg', {'side_effect': lambda: self.cfg}),
-            ('_check_dispatched', {}),
-            ('_targeted', {'side_effect': lambda actions, log: actions}),
-        ):
-            patcher = patch.object(drainer, name, **kwargs)
-            setattr(self, name, patcher.start())
-            self.addCleanup(patcher.stop)
-
-    def intent(self, name, age=60, actions=None, paths=None):
-        now = time.time()
-        return self.write_json(self.pending, name, {
-            'first_touch': now - age - 5, 'last_touch': now - age,
-            'actions': [{'state': 'soc', 'tgt': '*'}] if actions is None else actions,
-            'paths': paths or ['audit:soc.config.licenseKey'],
-        })
-
-    def test_no_pending_dir(self):
-        shutil.rmtree(self.pending)
-        self.assertEqual(drainer.main(), 0)
-        self._load_push_cfg.assert_not_called()
-
-    def test_cfg_error(self):
-        self._load_push_cfg.side_effect = RuntimeError('no salt')
-        self.assertEqual(drainer.main(), 1)
-
-    def test_disabled(self):
-        self.cfg['enabled'] = False
-        self.assertEqual(drainer.main(), 0)
-        self._check_dispatched.assert_not_called()
-
-    def test_no_intents_still_checks_results(self):
-        self.assertEqual(drainer.main(), 0)
-        self._check_dispatched.assert_called_once()
-
-    def test_debounce_and_broken(self):
-        young = self.intent('young.json', age=1)
-        broken = self.write_json(self.pending, 'broken.json', '{nope')
-        with patch.object(drainer, '_dispatch') as dispatch:
-            self.assertEqual(drainer.main(), 0)
-        dispatch.assert_not_called()
-        self.assertTrue(os.path.exists(young))
-        self.assertFalse(os.path.exists(broken))
-
-    def test_broken_unlink_error_ignored(self):
-        self.write_json(self.pending, 'broken.json', '{nope')
-        with patch.object(drainer.os, 'unlink', side_effect=OSError('busy')):
-            self.assertEqual(drainer.main(), 0)
-
-    def test_no_usable_actions(self):
-        path = self.intent('empty.json', actions=[{'state': 'soc'}])
-        self.assertEqual(drainer.main(), 0)
-        self.assertFalse(os.path.exists(path))
-        self.intent('empty.json', actions=[{'state': 'soc'}])
-        with patch.object(drainer.os, 'unlink', side_effect=OSError('busy')):
-            self.assertEqual(drainer.main(), 0)
-
-    def test_dispatch_failure_keeps_intents(self):
-        path = self.intent('pillar_soc.json')
-        with patch.object(drainer, '_dispatch', return_value=None):
-            self.assertEqual(drainer.main(), 1)
-        self.assertTrue(os.path.exists(path))
-
-    def test_dispatch_records_jid(self):
-        soc = self.intent('pillar_soc.json')
-        hs = self.intent('pillar_global.json', actions=[{'highstate': True, 'tgt': '*'}], paths=['audit:global.x'])
-        with patch.object(drainer, '_dispatch', return_value=JID) as dispatch, \
-                patch.object(drainer, '_record_dispatch') as record:
-            self.assertEqual(drainer.main(), 0)
-        self.assertEqual(len(dispatch.call_args[0][0]), 2)
-        record.assert_called_once()
-        self.assertEqual(record.call_args[0][0], JID)
-        self.assertEqual(sorted(record.call_args[0][2]), ['audit:global.x', 'audit:soc.config.licenseKey'])
-        self.assertFalse(os.path.exists(soc))
-        self.assertFalse(os.path.exists(hs))
-        self.assertIn('action: highstate tgt=*', self.logged('info'))
-
-    def test_dispatch_without_jid_not_recorded(self):
-        self.intent('pillar_soc.json')
-        with patch.object(drainer, '_dispatch', return_value=''), \
-                patch.object(drainer, '_record_dispatch') as record, \
-                patch.object(drainer.os, 'unlink', side_effect=OSError('busy')):
-            self.assertEqual(drainer.main(), 0)
-        record.assert_not_called()
-        self.log.exception.assert_called_once()
-
-    def test_untargeted_not_dispatched(self):
-        path = self.intent('strelka.json', actions=[{'state': 'strelka', 'tgt': 'G@role:so-sensor'}])
-        self._targeted.side_effect = None
-        self._targeted.return_value = []
-        with patch.object(drainer, '_dispatch') as dispatch, \
-                patch.object(drainer, '_record_dispatch') as record:
-            self.assertEqual(drainer.main(), 0)
-        dispatch.assert_not_called()
-        record.assert_not_called()
-        self.assertFalse(os.path.exists(path))
-        self.assertIn('no action matches a minion; nothing to dispatch', self.logged('info'))
-
-    def test_dispatches_only_targeted(self):
-        kept = {'state': 'soc', 'tgt': '*'}
-        self.intent('mixed.json', actions=[kept, {'state': 'strelka', 'tgt': 'G@role:so-sensor'}])
-        self._targeted.side_effect = None
-        self._targeted.return_value = [kept]
-        with patch.object(drainer, '_dispatch', return_value=JID) as dispatch, \
-                patch.object(drainer, '_record_dispatch') as record:
-            self.assertEqual(drainer.main(), 0)
-        self.assertEqual(dispatch.call_args[0][0], [kept])
-        self.assertEqual(record.call_args[0][1], [kept])
-
-
-if __name__ == '__main__':
-    unittest.main()
@@ -546,6 +546,7 @@ postupgrade_changes() {
    [[ "$POSTVERSION" == "3.1.0" ]] && post_to_3.2.0
    [[ "$POSTVERSION" == "3.2.0" ]] && post_to_3.3.0
    [[ "$POSTVERSION" == "3.3.0" ]] && post_to_3.4.0
+    [[ "$POSTVERSION" == "3.4.0" ]] && post_to_3.5.0
    # All applicable post-upgrade steps completed; clear the resume marker.
    rm -f "$POSTVERSION_FILE"
    true
@@ -1065,17 +1066,31 @@ telegraf_repair() {
  # when there is something to repair, so everyone else is left alone.
  local repair=/usr/sbin/so-telegraf-repair
  [[ -x "$repair" ]] || return 0
-  docker ps --format '{{.Names}}' | grep -qx so-postgres || return 0
+  local state
+  state=$(docker inspect -f '{{.State.Running}} {{.State.Restarting}}' so-postgres 2>/dev/null) || return 0
+  if [[ "$state" != "true false" ]]; then
+    echo "  so-postgres is not running; skipping the Telegraf partition check."
+    FINAL_MESSAGE_QUEUE+=("WARNING: so-postgres is not running, so Telegraf partitions were not checked. If /nsm is full, follow \"Telegraf Partition Repair\" at:")
+    FINAL_MESSAGE_QUEUE+=("    - https://securityonion.net/docs/postgresql")
+    return 0
+  fi

  echo "Checking Telegraf metric partitions."
-  local status=0
-  "$repair" --check >> "$SOUP_LOG" 2>&1 || status=$?
+  local out status=0
+  out=$("$repair" --check 2>&1) || status=$?
+  echo "$out" >> "$SOUP_LOG"
  case "$status" in
    0) echo "  Telegraf partitions are healthy; nothing to repair." ;;
    1) echo "  Repairing stalled Telegraf partitions."
-       "$repair" --yes \
-         || echo "  warning: so-telegraf-repair failed; run it manually" >&2 ;;
-    *) echo "  Skipping; Telegraf is not storing metrics in Postgres on this host." ;;
+       if ! "$repair" --yes; then
+         echo "  warning: so-telegraf-repair failed; run it manually" >&2
+         FINAL_MESSAGE_QUEUE+=("WARNING: Telegraf partition repair failed; Postgres metrics will keep filling /nsm until it is repaired:")
+         FINAL_MESSAGE_QUEUE+=("    - sudo so-telegraf-repair")
+       fi ;;
+    2) echo "  Skipping: $(tail -n 1 <<<"$out")" ;;
+    *) echo "  warning: could not check Telegraf partitions" >&2
+       FINAL_MESSAGE_QUEUE+=("WARNING: Telegraf partitions could not be checked during the upgrade; run:")
+       FINAL_MESSAGE_QUEUE+=("    - sudo so-telegraf-repair --check") ;;
  esac
 }

@@ -1202,6 +1217,15 @@ add_detection_runtime_fields() {
 }
 ### 3.4.0 End ###

+### 3.5.0 Scripts ###
+
+post_to_3.5.0() {
+  telegraf_repair
+
+  set_postversion 3.5.0
+}
+### 3.5.0 End ###
+

 repo_sync() {
  echo "Sync the local repo."
@@ -54,9 +54,9 @@ so-nginx:
      - /opt/so/log/nginx/:/var/log/nginx:rw
      - /opt/so/tmp/nginx/:/var/lib/nginx:rw
      - /opt/so/tmp/nginx/:/run:rw
-      - /nsm/elastic-fleet/so_agent-installers/:/opt/socore/html/packages
-      - /nsm/elastic-fleet/artifacts/:/opt/socore/html/artifacts 
+      - /nsm/elastic-fleet/artifacts/:/opt/socore/elastic-artifacts/artifacts:ro
      {% if GLOBALS.is_manager %}
+      - /nsm/elastic-fleet/so_agent-installers/:/opt/socore/html/packages
      - /etc/pki/managerssl.crt:/etc/pki/nginx/server.crt:ro
      - /etc/pki/managerssl.key:/etc/pki/nginx/server.key:ro
      # ATT&CK Navigator binds
@@ -4,6 +4,8 @@
 {%- set role = grains.id.split('_') | last %}
 {%- set influxpass = salt['pillar.get']('secrets:influx_pass') %}
 {%- set influxauth = ('so:' + influxpass) | base64_encode %}
+{%- set all_names = [GLOBALS.hostname, GLOBALS.url_base] + NGINXMERGED.ssl.alt_names %}
+{%- set full_server_name = all_names | unique | join(' ') %}

 worker_processes auto;
 error_log /var/log/nginx/error.log;
@@ -39,52 +41,31 @@ http {

 	include /etc/nginx/conf.d/*.conf;

-	{%- if role in ['fleet'] %}
-
-	server {
-		listen 8443;
-		server_name {{ GLOBALS.hostname }};
-		root /opt/socore/html;
-		location /artifacts/ {
-			try_files $uri =206;
-			proxy_read_timeout    90;
-			proxy_connect_timeout 90;
-			proxy_set_header      Host $host;
-			proxy_set_header      X-Real-IP $remote_addr;
-			proxy_set_header      X-Forwarded-For $proxy_add_x_forwarded_for;
-			proxy_set_header      Proxy "";
-			proxy_set_header      X-Forwarded-Proto $scheme;
-		}
-	}
-
-    {%- endif %}
-
 	{%- if GLOBALS.is_manager %}
-	{%- set all_names = [GLOBALS.hostname, GLOBALS.url_base] + NGINXMERGED.ssl.alt_names %}
-	{%- set full_server_name = all_names | unique | join(' ') %}
-
 	server {
 		listen 80 default_server;
 		server_name _;
 		return 307 https://{{ GLOBALS.url_base }}$request_uri;
 	}
-
+  {%- endif %}
+  {%- if GLOBALS.is_manager or role == 'fleet' %}
 	server {
 		listen 8443;
+    {%- if GLOBALS.is_manager %}
 		server_name {{ full_server_name }};
-		root /opt/socore/html;
+    {%- else %}
+    server_name {{ GLOBALS.hostname }};
+    {%- endif %}
+		root /opt/socore/elastic-artifacts;
 		location /artifacts/ {
-			try_files $uri =206;
-			proxy_read_timeout    90;
-			proxy_connect_timeout 90;
-			proxy_set_header      Host $host;
-			proxy_set_header      X-Real-IP $remote_addr;
-			proxy_set_header      X-Forwarded-For $proxy_add_x_forwarded_for;
-			proxy_set_header      Proxy "";
-			proxy_set_header      X-Forwarded-Proto $scheme;
+			try_files $uri =404;
+		}
+		location / {
+			return 404;
 		}
 	}
-
+  {%- endif %}
+  {%- if GLOBALS.is_manager %}
 	server {
 		listen 443 ssl default_server;
 		http2 on;
@@ -3,8 +3,6 @@
 {% set BATCH = AUTOAPPLY.batch %}
 {% set BATCH_WAIT = AUTOAPPLY.batch_wait %}

-{# queue must be a top-level salt.state arg (kwarg is ignored); an int is max_queue and still fails on conflict #}
-
 {% for action in actions %}
 {%   if action.get('highstate') %}
 apply_highstate_{{ loop.index }}:
@@ -14,7 +12,8 @@ apply_highstate_{{ loop.index }}:
    - highstate: True
    - batch: {{ action.get('batch', BATCH) }}
    - batch_wait: {{ action.get('batch_wait', BATCH_WAIT) }}
-    - queue: True
+    - kwarg:
+        queue: 2
 {%   else %}
 refresh_pillar_{{ loop.index }}:
  salt.function:
@@ -30,7 +29,8 @@ apply_{{ action.state | replace('.', '_') }}_{{ loop.index }}:
      - {{ action.state }}
    - batch: {{ action.get('batch', BATCH) }}
    - batch_wait: {{ action.get('batch_wait', BATCH_WAIT) }}
-    - queue: True
+    - kwarg:
+        queue: 2
    - require:
      - salt: refresh_pillar_{{ loop.index }}
 {%   endif %}
@@ -17,3 +17,4 @@ postgres:
    log_min_messages: 'warning'
    shared_preload_libraries: pg_cron
    cron.database_name: postgres
+    cron.host: /run/postgresql
@@ -87,3 +87,9 @@ postgres:
      global: True
      advanced: True
      helpLink: postgres
+    cron.host:
+      description: Where pg_cron connects to run scheduled jobs such as Telegraf partition maintenance. Must be the container's PostgreSQL socket directory, /run/postgresql, because pg_cron connects without a password and only local socket connections are trusted. Any other value makes every scheduled job fail. Changing it restarts PostgreSQL.
+      readonly: True
+      global: True
+      advanced: True
+      helpLink: postgres
@@ -52,7 +52,7 @@ if [ -n "$FILTER_HOST" ] && ! [[ "$FILTER_HOST" =~ $HOST_RE ]]; then
 fi

 so_psql() {
-  docker exec so-postgres psql -U postgres -d so_telegraf -At -F $'\t' "$@"
+  docker exec -e PGOPTIONS='-c application_name=so-stats-show' so-postgres psql -U so_maint_telegraf -d so_telegraf -At -F $'\t' "$@"
 }

 if ! docker exec so-postgres psql -U postgres -lqt 2>/dev/null | cut -d\| -f1 | grep -qw so_telegraf; then
@@ -60,6 +60,11 @@ if ! docker exec so-postgres psql -U postgres -lqt 2>/dev/null | cut -d\| -f1 |
  exit 2
 fi

+if ! docker exec so-postgres psql -U postgres -tAc "SELECT 1 FROM pg_roles WHERE rolname = 'so_maint_telegraf'" 2>/dev/null | grep -q 1; then
+  echo "The so_maint_telegraf role is missing; run a highstate."
+  exit 1
+fi
+
 table_exists() {
  local table="$1"
  [ -n "$(so_psql -c "SELECT 1 FROM information_schema.tables WHERE table_schema='${SCHEMA}' AND table_name='${table}' LIMIT 1;")" ]
@@ -4,15 +4,18 @@ set -e
 # Provision Telegraf state inside the so-postgres container.
 # Usage: so-telegraf-postgres <subcommand>
 #   create_db    Ensure the so_telegraf database exists.
-#   group_role   Provision the so_telegraf group role, telegraf/partman schemas,
-#                pg_partman, the so_admin maintenance routines, and the hourly
-#                pg_cron maintenance job.
+#   group_role   Provision the so_telegraf group role, the so_maint_telegraf
+#                maintenance login role, telegraf/partman schemas, pg_partman,
+#                the so_admin maintenance routines, and the hourly pg_cron
+#                maintenance job.
 #   user         Create or update a per-minion login role granted to so_telegraf.
 #                Env: ROLE_USER, ROLE_PASS.
 #   retention    Reconcile partman retention and premake on telegraf parents.
 #                Env: RETENTION_DAYS.
-#   maintenance  Drain default partitions and run partman maintenance.
-#   check        Report partition health. Non-zero if any parent is unhealthy.
+#   maintenance  Drain default partitions and run partman maintenance, as
+#                so_maint_telegraf.
+#   check        Report partition health, as so_maint_telegraf. Non-zero if any
+#                parent is unhealthy.
 #
 # A default partition holding rows for a day blocks creating that day's child,
 # so maintenance drains defaults before calling partman. Use so-telegraf-repair
@@ -20,6 +23,10 @@ set -e

 cmd="${1:?subcommand required}"

+# Long enough for an hourly maintenance run to release its locks, short enough
+# that soup and admins are not left hanging.
+LOCK_TIMEOUT=60s
+
 case "$cmd" in
  create_db)
    if ! docker exec so-postgres psql -U postgres -tAc \
@@ -30,7 +37,9 @@ case "$cmd" in
    ;;

  group_role)
-    docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -U postgres -d so_telegraf <<'EOSQL'
+    docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -v lock_timeout="$LOCK_TIMEOUT" \
+      -U postgres -d so_telegraf <<'EOSQL'
+SET lock_timeout = :'lock_timeout';
 DO $$
 BEGIN
    IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = 'so_telegraf') THEN
@@ -38,6 +47,17 @@ BEGIN
    END IF;
 END
 $$;
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = 'so_maint_telegraf') THEN
+        CREATE ROLE so_maint_telegraf;
+    END IF;
+END
+$$;
+ALTER ROLE so_maint_telegraf LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION NOBYPASSRLS PASSWORD NULL;
+GRANT so_telegraf TO so_maint_telegraf;
+ALTER ROLE so_maint_telegraf SET role = so_telegraf;
+ALTER ROLE so_maint_telegraf SET search_path = pg_catalog;
 GRANT CONNECT ON DATABASE so_telegraf TO so_telegraf;
 CREATE SCHEMA IF NOT EXISTS telegraf AUTHORIZATION so_telegraf;
 GRANT USAGE, CREATE ON SCHEMA telegraf TO so_telegraf;
@@ -58,44 +78,82 @@ ALTER DEFAULT PRIVILEGES IN SCHEMA partman
 ALTER DEFAULT PRIVILEGES IN SCHEMA partman
    GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO so_telegraf;

-- pg_cron runs these as postgres, so they must not sit in a schema any
-- Telegraf role can create objects in.
+-- Owned by postgres so no Telegraf role can create objects here; the routines
+-- run with the caller's privileges.
 CREATE SCHEMA IF NOT EXISTS so_admin AUTHORIZATION postgres;
 REVOKE ALL ON SCHEMA so_admin FROM PUBLIC;

+-- Resolved through pg_inherits, never by name, so a view or a table that only
+-- carries the _default name is never read or written.
+CREATE OR REPLACE FUNCTION so_admin.telegraf_defaults()
+RETURNS TABLE (
+    parent_table  regclass,
+    default_part  regclass,
+    has_rows      boolean,
+    default_size  bigint
+)
+LANGUAGE plpgsql
+STABLE
+SECURITY INVOKER
+AS $func$
+DECLARE
+    r  record;
+BEGIN
+    FOR r IN
+        SELECT to_regclass(pc.parent_table) AS pt
+        FROM partman.part_config pc
+        WHERE pc.parent_table LIKE 'telegraf.%'
+        ORDER BY pc.parent_table
+    LOOP
+        CONTINUE WHEN r.pt IS NULL;
+        parent_table := r.pt;
+
+        SELECT c.oid::regclass INTO default_part
+        FROM pg_catalog.pg_inherits i
+        JOIN pg_catalog.pg_class c ON c.oid = i.inhrelid
+        WHERE i.inhparent = r.pt
+          AND c.relkind = 'r' AND c.relispartition
+          AND pg_catalog.pg_get_expr(c.relpartbound, c.oid) = 'DEFAULT';
+
+        IF default_part IS NULL THEN
+            has_rows     := false;
+            default_size := NULL;
+        ELSE
+            EXECUTE format('SELECT EXISTS (SELECT 1 FROM %s LIMIT 1)', default_part)
+                INTO has_rows;
+            default_size := pg_catalog.pg_total_relation_size(default_part);
+        END IF;
+
+        RETURN NEXT;
+    END LOOP;
+END;
+$func$;
+
 CREATE OR REPLACE PROCEDURE so_admin.telegraf_maintenance()
 LANGUAGE plpgsql
+SECURITY INVOKER
+-- No SET clause: it would forbid the COMMITs below, so so_maint_telegraf pins search_path instead.
 AS $proc$
 DECLARE
-    r          record;
-    v_default  text;
-    v_rows     bigint;
+    r  record;
 BEGIN
    -- No per-parent EXCEPTION handler: partition_data_proc commits internally,
    -- and COMMIT is illegal while a subtransaction is active. A failing parent
    -- aborts the run and the next pass retries.
    FOR r IN
-        SELECT parent_table, retention
-        FROM partman.part_config
-        WHERE parent_table LIKE 'telegraf.%'
-        ORDER BY parent_table
+        SELECT pc.parent_table, pc.retention, d.default_part, d.default_size
+        FROM so_admin.telegraf_defaults() d
+        JOIN partman.part_config pc ON to_regclass(pc.parent_table) = d.parent_table
+        WHERE d.default_part IS NOT NULL AND d.has_rows
+        ORDER BY pc.parent_table
    LOOP
-        v_default := format('%I.%I',
-                            split_part(r.parent_table, '.', 1),
-                            split_part(r.parent_table, '.', 2) || '_default');
-
-        CONTINUE WHEN to_regclass(v_default) IS NULL;
-
-        EXECUTE format('SELECT count(*) FROM %s', v_default) INTO v_rows;
-        CONTINUE WHEN v_rows = 0;
-
-        RAISE WARNING 'so_admin.telegraf_maintenance: % rows stranded in %, draining',
-                      v_rows, v_default;
+        RAISE WARNING 'so_admin.telegraf_maintenance: % holds rows (%), draining',
+                      r.default_part, pg_size_pretty(r.default_size);

        -- Cheaper to delete expired rows than to repartition and then drop them.
        IF r.retention IS NOT NULL THEN
            EXECUTE format('DELETE FROM %s WHERE "time" < now() - %L::interval',
-                           v_default, r.retention);
+                           r.default_part, r.retention);
            COMMIT;
        END IF;

@@ -103,7 +161,7 @@ BEGIN
        CALL partman.partition_data_proc(
            p_parent_table := r.parent_table,
            p_loop_count   := 200,
-            p_source_table := v_default
+            p_source_table := r.default_part::text
        );
        COMMIT;
    END LOOP;
@@ -112,58 +170,75 @@ BEGIN
 END;
 $proc$;

-CREATE OR REPLACE FUNCTION so_admin.telegraf_partition_status()
+DROP FUNCTION IF EXISTS so_admin.telegraf_partition_status();
+CREATE FUNCTION so_admin.telegraf_partition_status()
 RETURNS TABLE (
-    parent_table  text,
-    oldest_child  date,
-    newest_child  date,
-    days_ahead    int,
-    retention     text,
-    default_rows  bigint,
-    default_size  text
+    parent_table      text,
+    oldest_child      date,
+    newest_child      date,
+    days_ahead        int,
+    retention         text,
+    default_has_rows  boolean,
+    default_size      text
 )
 LANGUAGE plpgsql
+STABLE
 AS $func$
 DECLARE
-    r          record;
-    v_default  regclass;
+    r  record;
 BEGIN
    FOR r IN
-        SELECT pc.parent_table AS pt, pc.retention AS ret
-        FROM partman.part_config pc
-        WHERE pc.parent_table LIKE 'telegraf.%'
-        ORDER BY pc.parent_table
+        SELECT d.parent_table AS pt, d.has_rows, d.default_size AS size, pc.retention AS ret
+        FROM so_admin.telegraf_defaults() d
+        JOIN partman.part_config pc ON to_regclass(pc.parent_table) = d.parent_table
+        ORDER BY d.parent_table::text
    LOOP
-        parent_table := r.pt;
+        parent_table := r.pt::text;
        retention    := r.ret;

-        SELECT min(d), max(d) INTO oldest_child, newest_child
+        SELECT min(x), max(x) INTO oldest_child, newest_child
        FROM (
-            SELECT to_date(substring(c.relname FROM '_p(\d{8})$'), 'YYYYMMDD') AS d
-            FROM pg_inherits i
-            JOIN pg_class c ON c.oid = i.inhrelid
-            WHERE i.inhparent = r.pt::regclass
-              AND pg_get_expr(c.relpartbound, c.oid) <> 'DEFAULT'
+            SELECT to_date(substring(c.relname FROM '_p(\d{8})$'), 'YYYYMMDD') AS x
+            FROM pg_catalog.pg_inherits i
+            JOIN pg_catalog.pg_class c ON c.oid = i.inhrelid
+            WHERE i.inhparent = r.pt
+              AND pg_catalog.pg_get_expr(c.relpartbound, c.oid) <> 'DEFAULT'
        ) s;

-        days_ahead := newest_child - current_date;
-
-        v_default := to_regclass(format('%I.%I',
-                                        split_part(r.pt, '.', 1),
-                                        split_part(r.pt, '.', 2) || '_default'));
-        IF v_default IS NULL THEN
-            default_rows := NULL;
-            default_size := NULL;
-        ELSE
-            EXECUTE format('SELECT count(*) FROM %s', v_default::text) INTO default_rows;
-            default_size := pg_size_pretty(pg_total_relation_size(v_default));
-        END IF;
+        days_ahead       := newest_child - current_date;
+        default_has_rows := r.has_rows;
+        default_size     := pg_size_pretty(r.size);

        RETURN NEXT;
    END LOOP;
 END;
 $func$;

+GRANT USAGE ON SCHEMA so_admin TO so_telegraf;
+GRANT EXECUTE ON FUNCTION so_admin.telegraf_defaults() TO so_telegraf;
+GRANT EXECUTE ON PROCEDURE so_admin.telegraf_maintenance() TO so_telegraf;
+GRANT EXECUTE ON FUNCTION so_admin.telegraf_partition_status() TO so_telegraf;
+
+-- Partitions created by a superuser session (an older repair) belong to
+-- postgres; partman acting as so_telegraf could not drop them.
+DO $$
+DECLARE
+    r  record;
+BEGIN
+    FOR r IN
+        SELECT n.nspname, c.relname
+        FROM pg_catalog.pg_class c
+        JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+        WHERE c.relkind IN ('r', 'p')
+          AND (n.nspname = 'telegraf'
+               OR (n.nspname = 'partman' AND c.relname LIKE 'template\_telegraf\_%'))
+          AND c.relowner <> 'so_telegraf'::regrole
+    LOOP
+        EXECUTE format('ALTER TABLE %I.%I OWNER TO so_telegraf', r.nspname, r.relname);
+    END LOOP;
+END
+$$;
+
 -- Drop the registration older releases left in so_telegraf.
 SELECT CASE
         WHEN current_setting('cron.database_name', true) IS DISTINCT FROM current_database()
@@ -175,20 +250,69 @@ DROP EXTENSION pg_cron CASCADE;
 \endif
 EOSQL

+    stalled=false
+    probe_failed=false
+    if out=$(docker exec -e PGOPTIONS="-c lock_timeout=$LOCK_TIMEOUT" so-postgres \
+        psql -U so_maint_telegraf -d so_telegraf -tAc \
+        "SELECT coalesce(bool_or(has_rows), false) FROM so_admin.telegraf_defaults()"); then
+      case "$out" in
+        t) stalled=true ;;
+        f) stalled=false ;;
+        *) probe_failed=true ;;
+      esac
+    else
+      probe_failed=true
+    fi
+    if $probe_failed; then
+      echo "ERROR: could not check Telegraf default partitions; the maintenance job was not changed." >&2
+      exit 1
+    fi
+
    # Guarded on the live GUC so applying this before the postgresql.conf change
    # has restarted the container skips rather than failing.
-    docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -U postgres -d postgres <<'EOSQL'
+    docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -v stalled="$stalled" \
+      -U postgres -d postgres <<'EOSQL'
 SELECT CASE WHEN current_setting('cron.database_name', true) = current_database()
            THEN 'true' ELSE 'false' END AS cron_here \gset
 \if :cron_here
 CREATE EXTENSION IF NOT EXISTS pg_cron;
-- cron.schedule_in_database is idempotent by jobname.
+-- A highstate reconciles the job's definition but never its active flag; only so-telegraf-repair turns it on.
+BEGIN;
+SELECT coalesce(max(jobid)::text, '') AS jobid,
+       count(*) > 0 AS job_exists,
+       coalesce(bool_and(active), true) AS job_active
+FROM cron.job
+WHERE jobname = 'telegraf-partman-maintenance' AND username = 'so_maint_telegraf' \gset
+SELECT cron.unschedule(jobid) FROM cron.job
+WHERE jobname = 'telegraf-partman-maintenance' AND username <> 'so_maint_telegraf';
+\if :job_exists
+SELECT cron.alter_job(:jobid,
+  schedule := '17 * * * *',
+  command  := 'CALL so_admin.telegraf_maintenance()',
+  database := 'so_telegraf');
+\else
+-- Registered inactive on a stalled grid so nothing drains the default
+-- partitions before so-telegraf-repair truncates them.
 SELECT cron.schedule_in_database(
  'telegraf-partman-maintenance',
  '17 * * * *',
  'CALL so_admin.telegraf_maintenance()',
-  'so_telegraf'
+  'so_telegraf',
+  'so_maint_telegraf',
+  NOT :'stalled'::boolean
 );
+\endif
+UPDATE cron.job SET nodename = current_setting('cron.host')
+WHERE jobname = 'telegraf-partman-maintenance'
+  AND nodename IS DISTINCT FROM current_setting('cron.host');
+COMMIT;
+SELECT coalesce(bool_and(active), true) AS job_active
+FROM cron.job
+WHERE jobname = 'telegraf-partman-maintenance' AND username = 'so_maint_telegraf' \gset
+\if :job_active
+\else
+\echo 'WARNING: the Telegraf maintenance job is inactive; run ''sudo so-telegraf-repair''.'
+\endif
 \else
 \echo 'pg_cron metadata database is not `postgres` yet; skipping job registration.'
 \endif
@@ -247,7 +371,7 @@ EOSQL
    ;;

  maintenance)
-    docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -U postgres -d so_telegraf <<'EOSQL'
+    docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -U so_maint_telegraf -d so_telegraf <<'EOSQL'
 SELECT CASE WHEN to_regproc('so_admin.telegraf_maintenance') IS NOT NULL
            THEN 'true' ELSE 'false' END AS has_proc \gset
 \if :has_proc
@@ -259,7 +383,7 @@ EOSQL
    ;;

  check)
-    docker exec -i so-postgres psql -U postgres -d so_telegraf <<'EOSQL'
+    docker exec -i so-postgres psql -U so_maint_telegraf -d so_telegraf <<'EOSQL'
 \pset border 2
 SELECT * FROM so_admin.telegraf_partition_status();
 EOSQL
@@ -278,9 +402,9 @@ LIMIT 5;
 \echo 'pg_cron is not installed in this database.'
 \endif
 EOSQL
-    unhealthy=$(docker exec so-postgres psql -U postgres -d so_telegraf -tAc \
+    unhealthy=$(docker exec so-postgres psql -U so_maint_telegraf -d so_telegraf -tAc \
      "SELECT count(*) FROM so_admin.telegraf_partition_status()
-       WHERE coalesce(default_rows, 0) > 0 OR coalesce(days_ahead, -1) < 1")
+       WHERE default_has_rows OR coalesce(days_ahead, -1) < 1")
    if [ "${unhealthy:-1}" != "0" ]; then
      echo "so-telegraf-postgres check: $unhealthy telegraf parent(s) unhealthy" >&2
      exit 1
@@ -11,26 +11,38 @@
 # maintenance once. Healthy grids are reported and left alone.
 #
 # Usage: so-telegraf-repair [--check] [--yes] [--no-restart]
-#   --check       Report health and change nothing.
+#   --check       Report health and change nothing. Also reports a missing
+#                 so_maint_telegraf role and a maintenance job that is missing
+#                 or inactive.
 #   --yes         Skip the confirmation prompt (for soup and other automation).
 #   --no-restart  Never restart so-postgres, even if pg_cron's launcher is dead.
 #
+# Repair mode (without --check) registers the so_maint_telegraf role and the
+# maintenance job if they are missing, pauses the job while it repairs, and
+# always leaves it active afterwards. Data work runs as so_maint_telegraf.
+#
 # Exit status:
 #   0  healthy, or repair completed
-#   1  repair is needed (--check only)
-#   2  cannot run here: so-postgres, so_telegraf or pg_partman is missing
+#   1  repair is needed (--check), or the repair failed after it started
+#   2  skipped: Telegraf output is not Postgres, or so-postgres, so_telegraf or pg_partman is missing
+#   3  could not check, for example a connection error

-set -e
+set -eE

 # Matches p_premake in telegraf.conf's create_parent template.
 PREMAKE=7
 JOB_NAME=telegraf-partman-maintenance
+MAINT_ROLE=so_maint_telegraf
+# Long enough for an hourly maintenance run to release its locks, short enough
+# that soup and admins are not left hanging.
+LOCK_TIMEOUT=60s
+PG_OPTS="-c lock_timeout=$LOCK_TIMEOUT -c application_name=so-telegraf-repair"

 CHECK_ONLY=false
 ASSUME_YES=false
 NO_RESTART=false

-usage() { sed -n '/^# Usage:/,/^#   2 /p' "$0" | sed 's/^# \?//'; }
+usage() { sed -n '/^# Usage:/,/^#   3 /p' "$0" | sed 's/^# \?//'; }

 while [[ $# -gt 0 ]]; do
  case "$1" in
@@ -38,70 +50,153 @@ while [[ $# -gt 0 ]]; do
    --yes|-y)          ASSUME_YES=true ;;
    --no-restart)      NO_RESTART=true ;;
    -h|--help)         usage; exit 0 ;;
-    *) echo "Unknown option: $1" >&2; usage >&2; exit 2 ;;
+    *) echo "Unknown option: $1" >&2; usage >&2; exit 3 ;;
  esac
  shift
 done

-skip() { echo "$*"; exit 2; }
+# Strips C0 (except tab and newline), DEL and C1 characters, so names and
+# messages written with Telegraf credentials can't drive the terminal.
+sanitise() { LC_ALL=C tr -d '\000-\010\013-\037\177' | LC_ALL=C sed 's/\xc2[\x80-\x9f]//g'; }

-psql_tg() { docker exec -i so-postgres psql -U postgres -d so_telegraf "$@"; }
-psql_pg() { docker exec -i so-postgres psql -U postgres -d postgres "$@"; }
+ERR_FILE=$(mktemp)
+ERR_STATUS=3
+ERR_PREFIX="Could not check Telegraf partitions"
+job_id=
+restore_active=
+
+on_err() {
+  # Command substitutions inherit this trap; only the top level reports.
+  [[ $BASH_SUBSHELL -eq 0 ]] || exit "$ERR_STATUS"
+  local line
+  line=$(tail -n 1 "$ERR_FILE" | sanitise)
+  echo "$ERR_PREFIX: ${line:-command failed}" >&2
+  exit "$ERR_STATUS"
+}
+
+on_exit() {
+  if [[ -n "$restore_active" ]]; then
+    psql_cron -tAc "SELECT cron.alter_job($job_id, active := $restore_active)" >/dev/null || true
+  fi
+  rm -f "$ERR_FILE"
+}
+
+trap on_err ERR
+trap on_exit EXIT
+
+finish() { trap - ERR; exit "$1"; }
+skip() { echo "$*"; finish 2; }
+
+# Keeps the last stderr line for on_err while still showing all of it.
+capture() {
+  local rc=0
+  : >"$ERR_FILE"
+  "$@" 2>"$ERR_FILE" || rc=$?
+  sanitise <"$ERR_FILE" >&2
+  return "$rc"
+}
+
+psql_tg() {
+  capture docker exec -i -e PGOPTIONS="$PG_OPTS" so-postgres psql -U "$MAINT_ROLE" -d so_telegraf "$@"
+}
+psql_pg() {
+  capture docker exec -i -e PGOPTIONS="$PG_OPTS" so-postgres psql -U postgres -d postgres "$@"
+}
+psql_cron() {
+  capture docker exec -i -e PGOPTIONS="$PG_OPTS" so-postgres psql -U postgres -d "$cron_db" "$@"
+}

-# query_to_xml so the per-table row counts need no helper function installed.
 REPORT="
-WITH parents AS (
-    SELECT pc.parent_table,
-           pc.premake,
-           split_part(pc.parent_table, '.', 1) AS sch,
-           split_part(pc.parent_table, '.', 2) AS tbl
-    FROM partman.part_config pc
-    WHERE pc.parent_table LIKE 'telegraf.%'
-), children AS (
-    SELECT p.parent_table,
-           max(to_date(substring(c.relname FROM '_p(\d{8})\$'), 'YYYYMMDD')) AS newest_child
-    FROM parents p
-    JOIN pg_class pt ON pt.oid = p.parent_table::regclass
-    JOIN pg_inherits i ON i.inhparent = pt.oid
-    JOIN pg_class c ON c.oid = i.inhrelid
-    WHERE pg_get_expr(c.relpartbound, c.oid) <> 'DEFAULT'
-    GROUP BY p.parent_table
-), defaults AS (
-    SELECT p.parent_table,
-           p.premake,
-           format('%I.%I', p.sch, p.tbl || '_default') AS default_table,
-           to_regclass(format('%I.%I', p.sch, p.tbl || '_default')) AS default_oid
-    FROM parents p
-)
 SELECT d.parent_table,
       c.newest_child,
       (c.newest_child - current_date) AS days_ahead,
-       d.premake,
-       CASE WHEN d.default_oid IS NULL THEN NULL ELSE
-            (xpath('/row/cnt/text()',
-                   query_to_xml(format('SELECT count(*) AS cnt FROM %s', d.default_table),
-                                false, true, '')))[1]::text::bigint
-       END AS default_rows,
-       CASE WHEN d.default_oid IS NULL THEN NULL
-            ELSE pg_size_pretty(pg_total_relation_size(d.default_oid)) END AS default_size
-FROM defaults d
-LEFT JOIN children c ON c.parent_table = d.parent_table
-ORDER BY 1
+       pc.premake,
+       d.has_rows AS default_has_rows,
+       pg_size_pretty(d.default_size) AS default_size
+FROM so_admin.telegraf_defaults() d
+JOIN partman.part_config pc ON to_regclass(pc.parent_table) = d.parent_table
+LEFT JOIN LATERAL (
+    SELECT max(to_date(substring(ch.relname FROM '_p(\d{8})\$'), 'YYYYMMDD')) AS newest_child
+    FROM pg_inherits i
+    JOIN pg_class ch ON ch.oid = i.inhrelid
+    WHERE i.inhparent = d.parent_table
+      AND pg_get_expr(ch.relpartbound, ch.oid) <> 'DEFAULT'
+) c ON true
+ORDER BY d.parent_table::text
 "

-docker ps --format '{{.Names}}' | grep -qx so-postgres \
+print_report() {
+  local report
+  report=$(psql_tg -c "$REPORT")
+  sanitise <<<"$report"
+}
+
+has_cron() {
+  [[ -n "$cron_db" ]] || return 0
+  psql_cron -tAc "SELECT 1 FROM pg_extension WHERE extname = 'pg_cron'"
+}
+
+lookup_job() {
+  local cron
+  cron=$(has_cron)
+  [[ -n "$cron" ]] || return 0
+  psql_cron -tA -F '|' -c \
+    "SELECT jobid, active FROM cron.job
+      WHERE jobname = '$JOB_NAME' AND username = '$MAINT_ROLE'
+      ORDER BY jobid LIMIT 1"
+}
+
+output=$(capture salt-call pillar.get telegraf:output --out=newline_values_only)
+output=$(tr -d '[:space:]' <<<"$output" | tr '[:lower:]' '[:upper:]')
+output=${output:-INFLUXDB}
+case "$output" in
+  POSTGRES|BOTH) ;;
+  *) skip "Telegraf output is $(sanitise <<<"$output"); not checking Postgres partitions." ;;
+esac
+
+containers=$(capture docker ps --format '{{.Names}}')
+grep -qx so-postgres <<<"$containers" \
  || skip "so-postgres is not running; nothing to repair."
-docker exec so-postgres psql -U postgres -tAc \
-  "SELECT 1 FROM pg_database WHERE datname='so_telegraf'" | grep -q 1 \
+has_db=$(psql_pg -tAc "SELECT 1 FROM pg_database WHERE datname = 'so_telegraf'")
+[[ -n "$has_db" ]] \
  || skip "The so_telegraf database does not exist; Telegraf is not writing to Postgres."
-psql_tg -tAc "SELECT 1 FROM pg_extension WHERE extname='pg_partman'" | grep -q 1 \
+
+# so_telegraf before the postgres state lands, postgres after.
+cron_db=$(psql_pg -tAc "SELECT current_setting('cron.database_name', true)")
+cron_db=$(tr -d '[:space:]' <<<"$cron_db")
+
+has_role=$(psql_pg -tAc "SELECT 1 FROM pg_roles WHERE rolname = '$MAINT_ROLE'")
+job=
+[[ "$cron_db" == postgres ]] && job=$(lookup_job)
+
+if [[ -z "$has_role" ]] && $CHECK_ONLY; then
+  echo "Repair is needed:"
+  echo "  * maintenance role $MAINT_ROLE is missing"
+  echo
+  echo "Re-run without --check to repair."
+  finish 1
+fi
+
+if ! $CHECK_ONLY && [[ -z "$has_role" || ( "$cron_db" == postgres && -z "$job" ) ]]; then
+  echo "Registering the maintenance role and job."
+  if ! /usr/sbin/so-telegraf-postgres group_role; then
+    echo "Could not register the maintenance job; run a highstate and try again." >&2
+    finish 3
+  fi
+  [[ "$cron_db" == postgres ]] && job=$(lookup_job)
+fi
+
+has_partman=$(psql_tg -tAc "SELECT 1 FROM pg_extension WHERE extname = 'pg_partman'")
+[[ -n "$has_partman" ]] \
  || skip "pg_partman is not installed in so_telegraf; nothing to repair."

 parents=$(psql_tg -tAc \
  "SELECT count(*) FROM partman.part_config WHERE parent_table LIKE 'telegraf.%'")
-stranded=$(psql_tg -tAc "SELECT coalesce(sum(default_rows), 0) FROM ( $REPORT ) t")
-behind=$(psql_tg -tAc \
-  "SELECT count(*) FROM ( $REPORT ) t WHERE coalesce(days_ahead, -1) < 1")
+counts=$(psql_tg -tAc \
+  "SELECT count(*) FILTER (WHERE default_has_rows) || '|' ||
+          count(*) FILTER (WHERE coalesce(days_ahead, -1) < 1)
+     FROM ( $REPORT ) t")
+IFS='|' read -r stranded behind <<<"$counts"
 # premake < 7, or infinite_time_partitions off: without the latter partman
 # refuses to premake forward across the gap the stall left behind.
 misconfigured=$(psql_tg -tAc \
@@ -115,16 +210,24 @@ launcher=$(psql_pg -tAc \
  "SELECT count(*) FROM pg_stat_activity
    WHERE backend_type ILIKE '%pg_cron%' OR application_name ILIKE '%pg_cron%'")

-# so_telegraf before the postgres state lands, postgres after.
-cron_db=$(docker exec so-postgres psql -U postgres -tAc \
-  "SELECT current_setting('cron.database_name', true)" | tr -d '[:space:]')
 last_run=never
-if [[ -n "$cron_db" ]]; then
-  last_run=$(docker exec so-postgres psql -U postgres -d "$cron_db" -tAc \
-    "SELECT coalesce(max(d.start_time)::text, 'never')
+cron_installed=$(has_cron)
+if [[ -n "$cron_installed" ]]; then
+  run=$(psql_cron -tAc \
+    "SELECT d.status || ' at ' || d.start_time
       FROM cron.job_run_details d JOIN cron.job j USING (jobid)
-      WHERE j.jobname = '$JOB_NAME'" 2>/dev/null | tr -d '[:space:]' || echo unknown)
-  [[ -n "$last_run" ]] || last_run=never
+      WHERE j.jobname = '$JOB_NAME'
+      ORDER BY d.start_time DESC LIMIT 1")
+  if [[ -n "$run" ]]; then
+    message=$(psql_cron -tAc \
+      "SELECT coalesce(d.return_message, '')
+         FROM cron.job_run_details d JOIN cron.job j USING (jobid)
+        WHERE j.jobname = '$JOB_NAME'
+        ORDER BY d.start_time DESC LIMIT 1")
+    message=$(sanitise <<<"$message")
+    message=${message//$'\n'/ }
+    last_run="$(sanitise <<<"$run"): ${message:0:200}"
+  fi
 fi

 # A grid that has never written a metric has nothing to recover, and an empty
@@ -132,43 +235,67 @@ fi
 restart_needed=false
 [[ "$launcher" -eq 0 && "$parents" -gt 0 && -n "$cron_db" ]] && restart_needed=true

+# group_role registers the job only in the postgres database.
+job_inactive=false
+if [[ "$cron_db" == postgres ]]; then
+  [[ -z "$job" || "${job##*|}" == f ]] && job_inactive=true
+fi
+
 repair_needed=false
 [[ "$stranded" -gt 0 ]] && repair_needed=true
 [[ "$behind" -gt 0 ]] && repair_needed=true
 [[ "$misconfigured" -gt 0 ]] && repair_needed=true
 $restart_needed && repair_needed=true
+$job_inactive && repair_needed=true

 echo "Telegraf partition status:"
-psql_tg -c "$REPORT"
-echo "Rows stranded in default partitions: $stranded"
-echo "pg_cron metadata database: ${cron_db:-unset}"
+print_report
+echo "Default partitions with rows: $stranded"
+echo "pg_cron metadata database: $(sanitise <<<"${cron_db:-unset}")"
+[[ "$cron_db" == postgres ]] \
+  || echo "pg_cron metadata database is $(sanitise <<<"${cron_db:-unset}"); not checking the maintenance job."
 echo "pg_cron launcher running: $([[ "$launcher" -gt 0 ]] && echo yes || echo no)"
 echo "Last $JOB_NAME run: $last_run"
 echo

 if ! $repair_needed; then
  echo "Telegraf partitions are healthy. Nothing to do."
-  exit 0
+  finish 0
 fi

 if $CHECK_ONLY; then
  echo "Repair is needed:"
-  [[ "$stranded" -gt 0 ]]    && echo "  * $stranded row(s) stranded in default partitions"
+  [[ "$stranded" -gt 0 ]]    && echo "  * $stranded default partition(s) hold rows"
  [[ "$behind" -gt 0 ]]      && echo "  * $behind parent(s) with no partition for the current window"
  [[ "$misconfigured" -gt 0 ]] && echo "  * $misconfigured parent(s) with stale partman settings"
  $restart_needed            && echo "  * pg_cron's launcher is dead; maintenance is not running at all"
+  $job_inactive              && echo "  * the maintenance job is missing or inactive"
  echo
  echo "Re-run without --check to repair."
-  exit 1
+  finish 1
 fi

 if [[ "$stranded" -gt 0 ]] && ! $ASSUME_YES; then
-  echo "This will permanently discard the $stranded stranded row(s) above."
+  echo "This will permanently discard the rows in the $stranded default partition(s) above."
  $restart_needed && ! $NO_RESTART && \
    echo "so-postgres will also be restarted, which briefly interrupts SOC."
-  [[ -t 0 ]] || { echo "Not a terminal; re-run with --yes to confirm." >&2; exit 2; }
+  [[ -t 0 ]] || { echo "Not a terminal; re-run with --yes to confirm." >&2; finish 1; }
  read -r -p "Continue? [y/N] " answer
-  [[ "$answer" =~ ^[Yy]$ ]] || { echo "Aborted."; exit 0; }
+  [[ "$answer" =~ ^[Yy]$ ]] || { echo "Aborted."; finish 0; }
+fi
+
+ERR_STATUS=1
+ERR_PREFIX="Telegraf partition repair failed"
+
+if [[ "$cron_db" == postgres ]]; then
+  if [[ -z "$job" ]]; then
+    echo "Could not register the maintenance job; run a highstate and try again." >&2
+    finish 3
+  fi
+  job_id=${job%%|*}
+  [[ "${job##*|}" == t ]] && restore_active=true || restore_active=false
+  echo "Pausing the $JOB_NAME job during the repair."
+  psql_cron -tAc "SELECT cron.alter_job($job_id, active := false)" >/dev/null
 fi

 if [[ "$misconfigured" -gt 0 ]]; then
@@ -186,27 +313,24 @@ if [[ "$stranded" -gt 0 ]]; then
  # One transaction: Telegraf is still writing, so a default emptied without
  # its partition in place is refilled before maintenance can attach one.
  psql_tg -v ON_ERROR_STOP=1 <<'EOSQL'
+BEGIN;
 DO $$
 DECLARE
    r record;
 BEGIN
    FOR r IN
-        SELECT pc.parent_table,
-               format('%I.%I', n.nspname, c.relname) AS default_table
-        FROM partman.part_config pc
-        JOIN pg_class p ON p.oid = pc.parent_table::regclass
-        JOIN pg_inherits i ON i.inhparent = p.oid
-        JOIN pg_class c ON c.oid = i.inhrelid
-        JOIN pg_namespace n ON n.oid = c.relnamespace
-        WHERE pc.parent_table LIKE 'telegraf.%'
-          AND pg_get_expr(c.relpartbound, c.oid) = 'DEFAULT'
+        SELECT pc.parent_table, d.default_part
+        FROM so_admin.telegraf_defaults() d
+        JOIN partman.part_config pc ON to_regclass(pc.parent_table) = d.parent_table
+        WHERE d.default_part IS NOT NULL
    LOOP
-        EXECUTE format('TRUNCATE TABLE %s', r.default_table);
+        EXECUTE format('TRUNCATE TABLE %s', r.default_part);
        PERFORM partman.create_partition_time(
            r.parent_table, ARRAY[date_trunc('day', now())]::timestamptz[]);
    END LOOP;
 END
 $$;
+COMMIT;
 EOSQL
 fi

@@ -222,7 +346,7 @@ if $restart_needed; then
      sleep 2
    done
    docker exec so-postgres pg_isready -U postgres -q \
-      || { echo "so-postgres did not come back; check 'docker logs so-postgres'." >&2; exit 1; }
+      || { echo "so-postgres did not come back; check 'docker logs so-postgres'." >&2; finish 1; }
  fi
 fi

@@ -238,8 +362,14 @@ CALL partman.run_maintenance_proc();
 \endif
 EOSQL

+if [[ -n "$job_id" ]]; then
+  psql_cron -tAc "SELECT cron.alter_job($job_id, active := true)" >/dev/null
+  restore_active=
+  echo "The $JOB_NAME job is active."
+fi
+
 echo
 echo "Telegraf partition status after repair:"
-psql_tg -c "$REPORT"
+print_report
 echo "The $JOB_NAME job runs hourly at :17. Confirm it fired with:"
 echo "  so-telegraf-repair --check"
@@ -7,7 +7,7 @@
 # Two branches, keyed on node_id:
 #   A) node_id populated -> the change is scoped to that one minion. Look up the
 #      app in pillar_push_map.yaml and write an intent that runs the app's mapped
-#      state(s) on just that node, and only if the node is in the app's target.
+#      state(s) targeted to just that node.
 #   B) node_id empty -> grid-wide app change. Look up the app in
 #      pillar_push_map.yaml and write an intent with the entry's actions as-is.
 #
@@ -116,17 +116,15 @@ def _app_from_setting(setting_id):


 def _node_actions(entry, node_id):
-    # Copy the app's mapped actions but narrow each one to the single node.
+    # Copy the app's mapped actions but retarget each one to the single node.
    # Preserves the state/highstate selection and any batch/batch_wait overrides.
-    # The app's own target is kept so the state is not applied to a node that does not
-    # run the app (e.g. an elasticsearch setting saved for a sensor fails to render there).
    actions = []
    for action in entry:
        if not isinstance(action, dict):
            continue
        node_action = dict(action)
-        node_action['tgt'] = 'L@{} and ( {} )'.format(node_id, action.get('tgt', '*'))
-        node_action['tgt_type'] = 'compound'
+        node_action['tgt'] = node_id
+        node_action['tgt_type'] = 'glob'
        actions.append(node_action)
    return actions

@@ -140,7 +138,6 @@ def run():
    # top level so the reactor is robust to either shape.
    event = data.get('data', data)  # noqa: F821 -- data provided by reactor
    setting_id = event.get('setting_id', '')
-    audit_id = event.get('id')
    node_id = (event.get('node_id') or '').strip()

    app = _app_from_setting(setting_id)
@@ -153,8 +150,8 @@ def run():
    if not entry:
        LOG.warning(
            'push_pillar: app "%s" is not in pillar_push_map.yaml; change will be '
-            'picked up at the next scheduled highstate (setting_id=%s audit_id=%s)',
-            app, setting_id, audit_id,
+            'picked up at the next scheduled highstate (setting_id=%s)',
+            app, setting_id,
        )
        return {}

@@ -168,12 +165,12 @@ def run():
            'node_{}_{}'.format(node_id, app), actions,
            'audit:{}@{}'.format(setting_id, node_id),
        )
-        LOG.info('push_pillar: per-node intent updated for %s on %s (setting_id=%s audit_id=%s)',
-                 app, node_id, setting_id, audit_id)
+        LOG.info('push_pillar: per-node intent updated for %s on %s (setting_id=%s)',
+                 app, node_id, setting_id)
        return {}

    # Branch B: grid-wide app change -> use the map entry's actions as-is.
    actions = list(entry)  # copy to avoid mutating the cache
    _write_intent('pillar_{}'.format(app), actions, 'audit:{}'.format(setting_id))
-    LOG.info('push_pillar: app intent updated for %s (setting_id=%s audit_id=%s)', app, setting_id, audit_id)
+    LOG.info('push_pillar: app intent updated for %s (setting_id=%s)', app, setting_id)
    return {}
@@ -138,6 +138,10 @@
 #                            AGGREGATOR PLUGINS                               #
 ###############################################################################

+[[aggregators.merge]]
+  drop_original = true
+  namepass = ["system", "swap", "kafka_partition"]
+
 ###############################################################################
 #                            INPUT PLUGINS                                    #
 ###############################################################################
@@ -289,7 +293,7 @@
  password = "{{ salt['pillar.get']('elasticsearch:auth:users:so_logstash_user:pass') }}"
 {%- endif %}

-{% if grains.role in ['so-manager','so-managersearch','so-standalone','so-receiver'] and GLOBALS.pipeline == "KAFKA" -%}
+{% if grains.role in ['so-manager','so-managerhype','so-managersearch','so-standalone','so-receiver'] and GLOBALS.pipeline == "KAFKA" -%}
 [[inputs.jolokia2_agent]]
  name_prefix= "kafka_"
  urls = ["http://{{ NODEIP }}:8778/jolokia"]
@@ -346,6 +350,26 @@
   interval = "60s"
 {%- endif %}

+{#- These report state that changes slowly; collecting them every interval only stores duplicate rows #}
+{%- set SLOWSCRIPTS = [] %}
+{%- for script in ['agentstatus.sh', 'os.sh', 'features.sh', 'raid.sh', 'zeekloss.sh'] %}
+{%-   if script in TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]] %}
+{%-     do TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]].remove(script) %}
+{%-     do SLOWSCRIPTS.append(script) %}
+{%-   endif %}
+{%- endfor %}
+{%- if SLOWSCRIPTS | length > 0 %}
+[[inputs.exec]]
+   commands = [
+{%-   for script in SLOWSCRIPTS %}
+      ["/scripts/{{script}}"]{% if not loop.last %},{% endif %}
+{%-   endfor %}
+   ]
+   data_format = "influx"
+   timeout = "15s"
+   interval = "5m"
+{%- endif %}
+
 {%- if TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]] | length > 0 %}
 [[inputs.exec]]
   commands = [
@@ -358,12 +382,6 @@
  timeout = "15s"
 {%- endif %}

-{%- if salt['pillar.get']('healthcheck:enabled', False) %}
-[[inputs.file]]
-  files = ["/host/nsm/zeek/logs/zeek_restart.log"]
-  data_format = "influx"
-{%- endif %}
-
 [[inputs.file]]
  files = ["/etc/telegraf/node_config.json"]
  name_override = "node_config"
@@ -18,7 +18,7 @@

 {% if GLOBALS.pipeline != 'REDIS' %}
 {# When global pipeline is not REDIS remove redis.sh script. KAFKA metrics are collected via jolokia agent. Config in telegraf.conf #}
-{%   if GLOBALS.role in ['so-standalone', 'so-manager', 'so-managersearch', 'so-receiver', 'so-heavynode'] %}
+{%   if GLOBALS.role in ['so-standalone', 'so-manager', 'so-managerhype', 'so-managersearch', 'so-receiver', 'so-heavynode'] %}
 {%     do TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]].remove('redis.sh') %}
 {%   endif %}
 {% endif %}
@@ -10,7 +10,7 @@
 # if this script isn't already running
 if [[ ! "`pidof -x $(basename $0) -o %PPID`" ]]; then

-    FILES=$(ls -1x /host/nsm/strelka/unprocessed | wc -l)
+    FILES=$(find /host/nsm/strelka/unprocessed -maxdepth 1 -type f | wc -l)
    echo "faffiles files=$FILES"
    
 fi
@@ -12,8 +12,9 @@ if [[ ! "`pidof -x $(basename $0) -o %PPID`" ]]; then

    CONTAINERSLOG=/var/log/somon/containers.log
    # the collector rewrites this every minute; report nothing rather than repeating a stale
-    # file as if it were current, in case a run was skipped or the collector is wedged
-    MAXAGE=150
+    # file as if it were current, in case a run was skipped or the collector is wedged.
+    # Kept under two collector cycles so a 60s telegraf interval can't emit the same snapshot twice.
+    MAXAGE=90

    if [ -r "$CONTAINERSLOG" ]; then
        AGE=$(( $(date +%s) - $(stat -c %Y "$CONTAINERSLOG") ))
@@ -11,27 +11,40 @@
 if [[ ! "`pidof -x $(basename $0) -o %PPID`" ]]; then

    PREVCOUNTFILE='/tmp/eps.txt'
-    EVENTCOUNTCURRENT="$(curl -s localhost:9600/_node/stats | jq '.events.in')"
+    STATS="$(curl -s localhost:9600/_node/stats)"

-    if [ ! -z "$EVENTCOUNTCURRENT" ]; then
-
-      if [ -f "$PREVCOUNTFILE" ]; then
-        EVENTCOUNTPREVIOUS=`cat $PREVCOUNTFILE`
-      else
-        echo "${EVENTCOUNTCURRENT}" > $PREVCOUNTFILE
-        exit 0
-      fi
-
-      echo "${EVENTCOUNTCURRENT}" > $PREVCOUNTFILE
-      # the division by 30 is because the agent interval is 30 seconds
-      EVENTS=$(((EVENTCOUNTCURRENT - EVENTCOUNTPREVIOUS)/30))
-      if [ "$EVENTS" -lt 0 ]; then
-        EVENTS=0
-      fi
-
-      echo "consumptioneps eps=${EVENTS%%.*}"
+    if echo "$STATS" | jq -e '.pipelines.search' > /dev/null 2>&1; then
+      STAGE=search
+      COUNT="$(echo "$STATS" | jq '.pipelines.search.events.out')"
+    elif echo "$STATS" | jq -e '.pipelines.receiver' > /dev/null 2>&1; then
+      STAGE=receiver
+      COUNT="$(echo "$STATS" | jq '.pipelines.receiver.events.in')"
+    else
+      exit 0
    fi

+    if ! [[ "$COUNT" =~ ^[0-9]+$ ]]; then
+      exit 0
+    fi
+
+    NOW=$(date +%s)
+    read -r PREVEPOCH PREVCOUNT 2>/dev/null < "$PREVCOUNTFILE"
+    echo "${NOW} ${COUNT}" > "$PREVCOUNTFILE"
+
+    if ! [[ "$PREVEPOCH" =~ ^[0-9]+$ && "$PREVCOUNT" =~ ^[0-9]+$ ]]; then
+      exit 0
+    fi
+
+    ELAPSED=$((NOW - PREVEPOCH))
+    DELTA=$((COUNT - PREVCOUNT))
+    if [ "$ELAPSED" -le 0 ] || [ "$DELTA" -lt 0 ]; then
+      EPS=0
+    else
+      EPS=$((DELTA / ELAPSED))
+    fi
+
+    echo "consumptioneps,stage=${STAGE} eps=${EPS}"
+
 fi

 exit 0
@@ -10,8 +10,7 @@ if [[ ! "`pidof -x $(basename $0) -o %PPID`" ]]; then
    FPS_ENABLED=$(cat /var/log/sostatus/fps_enabled)
    LKS_ENABLED=$(cat /var/log/sostatus/lks_enabled)

-    echo "features fps=$FPS_ENABLED"
-    echo "features lks=$LKS_ENABLED"
+    echo "features fps=$FPS_ENABLED,lks=$LKS_ENABLED"
 fi

 exit 0
Author	SHA1	Message	Date
Mike Reeves	ae394ce709	Postgres: pg_cron over the local socket, so_maint_telegraf role, repair and soup 3.5.0	2026-11-12 19:07:57 -05:00
Jorge Reyes	40a30f6908	Merge pull request #16389 from Security-Onion-Solutions/reyesj2-patch-1 update elastic-agent artifact repo directory	2026-11-11 14:32:59 -06:00
Mike Reeves	1fd37251c9	Merge pull request #16390 from Security-Onion-Solutions/foundry/mwaq014injcipm Telegraf: fix EPS double counting and duplicate metric rows	2026-11-10 14:53:30 -05:00
Mike Reeves	4a5253f688	Telegraf: fix EPS double counting and duplicate metric rows	2026-11-10 08:26:19 -05:00
reyesj2	ec220572f0	dedup artifact registry config & split ea artifacts out of /opt/socore/html/ into dedicated /opt/socore/elastic-artifacts/artifacts/	2026-11-09 12:32:06 -06:00