FIX: report push result as unknown when the orchestration raises

A queued push is tracked by an orchestration that waits on the minion's
state run. If salt-master restarts while it waits, the orchestration step
raises AuthenticationError and never collects the minion's return, but the
minion still runs the queued state. so-push-drainer logged these as
"push failed ... change will be applied at the next scheduled highstate",
which was wrong on both counts.

Seen on a fresh 3.4.0 standalone: hydra.enabled and telegraf.output were
saved just after SOC started a grid highstate. Both pushes queued behind
it. That highstate was the first with a license granting the vrt feature,
so salt.master wrote vrt_engine.conf and reactor_hypervisor.conf and
restarted salt-master as its last step. Both orchestrations raised, the
minion then ran both state runs with 0 failed states, and the drainer
logged two ERRORs.

When every failed step of the orchestration raised (salt's "An exception
occurred in this state:" comment, with no minion returns), log a WARNING
that the result is unknown and the state run may still have completed.
Conflicts, failed states, render errors, and results mixing a raised step
with a real failure still log ERROR.

salt/manager/tools/sbin/so-push-drainer

@@ -53,6 +53,9 @@ 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})')
 
@@ -242,6 +245,24 @@ def _orch_failures(ret):
     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))
 
@@ -283,7 +304,11 @@ def _report_result(record, age, log):
             return True
         return False
     failures = _orch_failures(ret)
-    if failures:
+    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:

salt/manager/tools/sbin/so-push-drainer_test.py

@@ -76,6 +76,17 @@ STATE_FAIL_RET = _orch_ret({
     },
 }, 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,
@@ -294,6 +305,15 @@ class TestResults(DrainerTestCase):
         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],
@@ -306,6 +326,7 @@ class TestResults(DrainerTestCase):
             '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}
@@ -317,13 +338,16 @@ class TestResults(DrainerTestCase):
         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'):
+        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()