Merge pull request #50756 from GwiYeong/local-client-fix

missing minion result data bug.

salt/netapi/rest_tornado/saltnado.py

@@ -921,14 +921,27 @@ class SaltAPIHandler(BaseSaltAPIHandler):  # pylint: disable=W0223
         '''
         Dispatch local client commands
         '''
-        # Generate jid before triggering a job to subscribe all returns from minions
+        # Generate jid and find all minions before triggering a job to subscribe all returns from minions
-        chunk['jid'] = salt.utils.jid.gen_jid()
+        chunk['jid'] = salt.utils.jid.gen_jid() if not chunk.get('jid', None) else chunk['jid']
+        minions = set(self.ckminions.check_minions(chunk['tgt'], chunk.get('tgt_type', 'glob')))
+
+        def subscribe_minion(minion):
+            salt_evt = self.application.event_listener.get_event(
+                     self,
+                     tag='salt/job/{}/ret/{}'.format(chunk['jid'], minion),
+                     matcher=EventListener.exact_matcher)
+            syndic_evt = self.application.event_listener.get_event(
+                    self,
+                    tag='syndic/job/{}/ret/{}'.format(chunk['jid'], minion),
+                    matcher=EventListener.exact_matcher)
+            return salt_evt, syndic_evt
 
         # start listening for the event before we fire the job to avoid races
-        events = [
+        events = []
-            self.application.event_listener.get_event(self, tag='salt/job/'+chunk['jid']),
+        for minion in minions:
-            self.application.event_listener.get_event(self, tag='syndic/job/'+chunk['jid']),
+            salt_evt, syndic_evt = subscribe_minion(minion)
-        ]
+            events.append(salt_evt)
+            events.append(syndic_evt)
 
         f_call = self._format_call_run_job_async(chunk)
         # fire a job off
@@ -947,6 +960,12 @@ class SaltAPIHandler(BaseSaltAPIHandler):  # pylint: disable=W0223
                     pass
             raise tornado.gen.Return('No minions matched the target. No command was sent, no jid was assigned.')
 
+        # get_event for missing minion
+        for minion in list(set(pub_data['minions']) - set(minions)):
+            salt_evt, syndic_evt = subscribe_minion(minion)
+            events.append(salt_evt)
+            events.append(syndic_evt)
+
         # Map of minion_id -> returned for all minions we think we need to wait on
         minions = {m: False for m in pub_data['minions']}
 
@@ -1001,7 +1020,10 @@ class SaltAPIHandler(BaseSaltAPIHandler):  # pylint: disable=W0223
                         cancel_inflight_futures()
                         raise tornado.gen.Return(chunk_ret)
                     continue
+
                 f_result = f.result()
+                if f in events:
+                    events.remove(f)
                 # if this is a start, then we need to add it to the pile
                 if f_result['tag'].endswith('/new'):
                     for minion_id in f_result['data']['minions']:
@@ -1011,7 +1033,6 @@ class SaltAPIHandler(BaseSaltAPIHandler):  # pylint: disable=W0223
                     chunk_ret[f_result['data']['id']] = f_result['data']['return']
                     # clear finished event future
                     minions[f_result['data']['id']] = True
-
                     # if there are no more minions to wait for, then we are done
                     if not more_todo() and min_wait_time.done():
                         cancel_inflight_futures()
@@ -1020,11 +1041,6 @@ class SaltAPIHandler(BaseSaltAPIHandler):  # pylint: disable=W0223
             except TimeoutException:
                 pass
 
-            if f == events[0]:
-                events[0] = self.application.event_listener.get_event(self, tag='salt/job/'+chunk['jid'])
-            else:
-                events[1] = self.application.event_listener.get_event(self, tag='syndic/job/'+chunk['jid'])
-
     @tornado.gen.coroutine
     def job_not_running(self, jid, tgt, tgt_type, minions, is_finished):
         '''