Allow saltutil.refresh_modules to wait for completion

Also added raet support to eventlisten as well as added functionality to connect to

salt/loader.py

@@ -20,6 +20,7 @@ from salt.exceptions import LoaderError
 from salt.template import check_render_pipe_str
 from salt.utils.decorators import Depends
 import salt.utils.lazy
+import salt.utils.event
 
 # Solve the Chicken and egg problem where grains need to run before any
 # of the modules are loaded and are generally available for any usage.
@@ -89,7 +90,7 @@ def _module_dirs(
     return cli_module_dirs + ext_type_types + [ext_types, sys_types]
 
 
-def minion_mods(opts, context=None, whitelist=None, include_errors=False, initial_load=False):
+def minion_mods(opts, context=None, whitelist=None, include_errors=False, initial_load=False, notify=False):
     '''
     Load execution modules
 
@@ -117,6 +118,9 @@ def minion_mods(opts, context=None, whitelist=None, include_errors=False, initia
                      pack={'__context__': context},
                      whitelist=whitelist)
     ret.pack['__salt__'] = ret
+    if notify:
+        evt = salt.utils.event.get_event('minion', opts=opts)
+        evt.fire_event({'complete': True}, tag='/salt/minion/minion_mod_complete')
 
     return ret
 

salt/minion.py

@@ -827,7 +827,7 @@ class Minion(MinionBase):
             mod_opts[key] = val
         return mod_opts
 
-    def _load_modules(self, force_refresh=False):
+    def _load_modules(self, force_refresh=False, notify=False):
         '''
         Return the functions and the returners loaded up from the loader
         module
@@ -852,9 +852,9 @@ class Minion(MinionBase):
         self.opts['grains'] = salt.loader.grains(self.opts, force_refresh)
         if self.opts.get('multimaster', False):
             s_opts = copy.copy(self.opts)
-            functions = salt.loader.minion_mods(s_opts)
+            functions = salt.loader.minion_mods(s_opts, notify=notify)
         else:
-            functions = salt.loader.minion_mods(self.opts)
+            functions = salt.loader.minion_mods(self.opts, notify=notify)
         returners = salt.loader.returners(self.opts, functions)
         errors = {}
         if '_errors' in functions:
@@ -1433,12 +1433,12 @@ class Minion(MinionBase):
         else:
             self.publish_port = auth.creds['publish_port']
 
-    def module_refresh(self, force_refresh=False):
+    def module_refresh(self, force_refresh=False, notify=False):
         '''
         Refresh the functions and returners.
         '''
-        log.debug('Refreshing modules')
-        self.functions, self.returners, _ = self._load_modules(force_refresh)
+        log.debug('Refreshing modules. Notify={0}'.format(notify))
+        self.functions, self.returners, _ = self._load_modules(force_refresh, notify=notify)
         self.schedule.functions = self.functions
         self.schedule.returners = self.returners
 
@@ -1559,7 +1559,8 @@ class Minion(MinionBase):
         '''
         log.debug('Handling event {0!r}'.format(package))
         if package.startswith('module_refresh'):
-            self.module_refresh()
+            tag, data = salt.utils.event.MinionEvent.unpack(package)
+            self.module_refresh(notify=data.get('notify', False))
         elif package.startswith('pillar_refresh'):
             self.pillar_refresh()
         elif package.startswith('manage_schedule'):
@@ -2718,7 +2719,7 @@ class ProxyMinion(Minion):
         '''
         return super(ProxyMinion, self)._prep_mod_opts()
 
-    def _load_modules(self, force_refresh=False):
+    def _load_modules(self, force_refresh=False, notify=False):
         '''
         Return the functions and the returners loaded up from the loader
         module

salt/modules/saltutil.py

@@ -40,6 +40,7 @@ import salt.runner
 import salt.utils
 import salt.utils.process
 import salt.utils.minion
+import salt.utils.event
 import salt.transport
 import salt.wheel
 from salt.exceptions import (
@@ -420,10 +421,14 @@ def refresh_pillar():
 pillar_refresh = refresh_pillar
 
 
-def refresh_modules():
+def refresh_modules(async=True):
     '''
     Signal the minion to refresh the module and grain data
 
+    The default is to refresh module asyncrhonously. To block
+    until the module refresh is complete, set the 'async' flag
+    to False.
+
     CLI Example:
 
     .. code-block:: bash
@@ -431,7 +436,16 @@ def refresh_modules():
         salt '*' saltutil.refresh_modules
     '''
     try:
-        ret = __salt__['event.fire']({}, 'module_refresh')
+        if async:
+            #  If we're going to block, first setup a listener
+            ret = __salt__['event.fire']({}, 'module_refresh')
+        else:
+            eventer = salt.utils.event.get_event('minion', opts=__opts__)
+            ret = __salt__['event.fire']({'notify': True}, 'module_refresh')
+            # Wait for the finish event to fire
+            log.trace('refresh_modules waiting for module refresh to complete')
+            # Blocks until we hear this event or until the timeout expires
+            eventer.get_event(tag='/salt/minion/minion_mod_complete', wait=30)
     except KeyError:
         log.error('Event module not available. Module refresh failed.')
         ret = False  # Effectively a no-op, since we can't really return without an event system

tests/eventlisten.py

@@ -46,6 +46,17 @@ def parse():
             help=('Return a count of the number of minions which have '
                  'replied to a job with a given func.'))
 
+    parser.add_option('-i',
+            '--id',
+            default='',
+            help=('If connecting to a live master or minion, pass in the id'))
+
+    parser.add_option('-t',
+            '--transport',
+            default='zeromq',
+            help=('Transport to use. (Default: \'zeromq\''))
+
+
     options, args = parser.parse_args()
 
     opts = {}
@@ -60,8 +71,10 @@ def parse():
         if args:
             opts['id'] = args[0]
             return opts
-
-        opts['id'] = options.node
+        if options.id:
+            opts['id'] = options.id
+        else:
+            opts['id'] = options.node
 
     return opts
 
@@ -85,9 +98,11 @@ def listen(opts):
     '''
     Attach to the pub socket and grab messages
     '''
-    event = salt.utils.event.SaltEvent(
+    event = salt.utils.event.get_event(
             opts['node'],
-            opts['sock_dir']
+            sock_dir=opts['sock_dir'],
+            transport=opts['transport'],
+            opts=opts
             )
     check_access_and_print_warning(opts['sock_dir'])
     print(event.puburi)