Merge pull request #37086 from cachedout/mm_req

Make salt-call a first-class citizen for multi-master

salt/minion.py

@@ -505,6 +505,15 @@ class MinionBase(object):
             if opts['random_master']:
                 shuffle(opts['local_masters'])
             last_exc = None
+            opts['master_uri_list'] = list()
+
+            # This sits outside of the connection loop below because it needs to set
+            # up a list of master URIs regardless of which masters are available
+            # to connect _to_. This is primarily used for masterless mode, when
+            # we need a list of master URIs to fire calls back to.
+            for master in opts['local_masters']:
+                opts['master'] = master
+                opts['master_uri_list'].append(resolve_dns(opts)['master_uri'])
 
             while True:
                 attempts += 1

salt/modules/event.py

@@ -63,6 +63,13 @@ def fire_master(data, tag, preload=None):
                     ip=salt.utils.ip_bracket(__opts__['interface']),
                     port=__opts__.get('ret_port', '4506')  # TODO, no fallback
                     )
+        masters = list()
+        ret = True
+        if 'master_uri_list' in __opts__:
+            for master_uri in __opts__['master_uri_list']:
+                masters.append(master_uri)
+        else:
+            masters.append(__opts__['master_uri'])
         auth = salt.crypt.SAuth(__opts__)
         load = {'id': __opts__['id'],
                 'tag': tag,
@@ -73,12 +80,13 @@ def fire_master(data, tag, preload=None):
         if isinstance(preload, dict):
             load.update(preload)
 
-        channel = salt.transport.Channel.factory(__opts__)
-        try:
-            channel.send(load)
-        except Exception:
-            pass
-        return True
+        for master in masters:
+            channel = salt.transport.Channel.factory(__opts__, master_uri=master)
+            try:
+                channel.send(load)
+            except Exception:
+                ret = False
+        return ret
     else:
         # Usually, we can send the event via the minion, which is faster
         # because it is already authenticated

salt/modules/publish.py

@@ -13,7 +13,7 @@ import salt.crypt
 import salt.payload
 import salt.transport
 import salt.utils.args
-from salt.exceptions import SaltReqTimeoutError
+from salt.exceptions import SaltReqTimeoutError, SaltInvocationError
 
 log = logging.getLogger(__name__)
 
@@ -46,7 +46,8 @@ def _publish(
         returner='',
         timeout=5,
         form='clean',
-        wait=False):
+        wait=False,
+        via_master=None):
     '''
     Publish a command from the minion out to other minions, publications need
     to be enabled on the Salt master and the minion needs to have permission
@@ -75,7 +76,35 @@ def _publish(
 
     arg = _parse_args(arg)
 
-    log.info('Publishing \'{0}\' to {master_uri}'.format(fun, **__opts__))
+    if via_master:
+        if 'master_uri_list' not in __opts__:
+            raise SaltInvocationError(message='Could not find list of masters \
+                    in minion configuration but `via_master` was specified.')
+        else:
+            # Find the master in the list of master_uris generated by the minion base class
+            matching_master_uris = [master for master
+                    in __opts__['master_uri_list']
+                    if '//{0}:'.format(via_master)
+                    in master]
+
+            if not matching_master_uris:
+                raise SaltInvocationError('Could not find match for {0} in \
+                list of configured masters {1} when using `via_master` option'.format(
+                    via_master, __opts__['master_uri_list']))
+
+            if len(matching_master_uris) > 1:
+                # If we have multiple matches, consider this a non-fatal error
+                # and continue with whatever we found first.
+                log.warning('The `via_master` flag found \
+                        more than one possible match found for {0} when evaluating \
+                        list {1}'.format(via_master, __opts__['master_uri_list']))
+            master_uri = matching_master_uris.pop()
+    else:
+        # If no preference is expressed by the user, just publish to the first master
+        # in the list.
+        master_uri = __opts__['master_uri']
+
+    log.info('Publishing \'{0}\' to {1}'.format(fun, master_uri))
     auth = salt.crypt.SAuth(__opts__)
     tok = auth.gen_token('salt')
     load = {'cmd': 'minion_pub',
@@ -89,7 +118,7 @@ def _publish(
             'form': form,
             'id': __opts__['id']}
 
-    channel = salt.transport.Channel.factory(__opts__)
+    channel = salt.transport.Channel.factory(__opts__, master_uri=master_uri)
     try:
         peer_data = channel.send(load)
     except SaltReqTimeoutError:
@@ -145,7 +174,7 @@ def _publish(
             return ret
 
 
-def publish(tgt, fun, arg=None, expr_form='glob', returner='', timeout=5):
+def publish(tgt, fun, arg=None, expr_form='glob', returner='', timeout=5, via_master=None):
     '''
     Publish a command from the minion out to other minions.
 
@@ -204,7 +233,9 @@ def publish(tgt, fun, arg=None, expr_form='glob', returner='', timeout=5):
             salt '*' publish.publish test.kwarg arg="['cheese=spam','spam=cheese']"
 
 
-
+    When running via salt-call, the `via_master` flag may be set to specific which
+    master the publication should be sent to. Only one master may be specified. If
+    unset, the publication will be sent only to the first master in minion configuration.
     '''
     return _publish(tgt,
                     fun,
@@ -213,7 +244,8 @@ def publish(tgt, fun, arg=None, expr_form='glob', returner='', timeout=5):
                     returner=returner,
                     timeout=timeout,
                     form='clean',
-                    wait=True)
+                    wait=True,
+                    via_master=via_master)
 
 
 def full_data(tgt, fun, arg=None, expr_form='glob', returner='', timeout=5):

salt/modules/saltutil.py

@@ -959,17 +959,26 @@ def revoke_auth(preserve_minion_cache=False):
 
         salt '*' saltutil.revoke_auth
     '''
-    channel = salt.transport.Channel.factory(__opts__)
-    tok = channel.auth.gen_token('salt')
-    load = {'cmd': 'revoke_auth',
-            'id': __opts__['id'],
-            'tok': tok,
-            'preserve_minion_cache': preserve_minion_cache}
+    masters = list()
+    ret = True
+    if 'master_uri_list' in __opts__:
+        for master_uri in __opts__['master_uri_list']:
+            masters.append(master_uri)
+    else:
+        masters.append(__opts__['master_uri'])
 
-    try:
-        return channel.send(load)
-    except SaltReqTimeoutError:
-        return False
+    for master in masters:
+        channel = salt.transport.Channel.factory(__opts__, master_uri=master)
+        tok = channel.auth.gen_token('salt')
+        load = {'cmd': 'revoke_auth',
+                'id': __opts__['id'],
+                'tok': tok,
+                'preserve_minion_cache': preserve_minion_cache}
+        try:
+            channel.send(load)
+        except SaltReqTimeoutError:
+            ret = False
+    return ret
 
 
 def _get_ssh_or_api_client(cfgfile, ssh=False):

salt/utils/__init__.py

@@ -3010,3 +3010,11 @@ def str_version_to_evr(verstring):
         release = ''
 
     return epoch, version, release
+
+
+def substr_in_list(string_to_search_for, list_to_search):
+    '''
+    Return a boolean value that indicates whether or not a given
+    string is present in any of the strings which comprise a list
+    '''
+    return any(string_to_search_for in s for s in list_to_search)