Gathers the data from the specified minions' mine

        :param dict load: A payload received from a minion

        :rtype: dict
        :return: Mine data from the specified minions
510547
def _mine_get(self, load):
        '''
        
        '''
        load = self.__verify_load(load, ('id', 'tgt', 'fun', 'tok'))
        if load is False:
            return {}
        else:
            return self.masterapi._mine_get(load, skip_verify=True)
182481
def save_load(jid, load, minions=None):
    '''
    
    '''
    options = _get_options()

    index = options['master_job_cache_index']
    doc_type = options['master_job_cache_doc_type']

    if options['index_date']:
        index = '{0}-{1}'.format(index,
            datetime.date.today().strftime('%Y.%m.%d'))

    _ensure_index(index)

    # addressing multiple types (bool, string, dict, ...) issue in master_job_cache index for return key (#20826)
    if not load.get('return', None) is None:
        # if load.return is not a dict, moving the result to load.return.<job_fun_escaped>.return
        if not isinstance(load['return'], dict):
            job_fun_escaped = load['fun'].replace('.', '_')
            load['return'] = {job_fun_escaped: {'return': load['return']}}
        # rename load.return to load.data in order to have the same key in all indices (master_job_cache, job)
        load['data'] = load.pop('return')

    data = {
        'jid': jid,
        'load': load,
    }

    ret = __salt__['elasticsearch.document_create'](index=index,
                                                    doc_type=doc_type,
                                                    id=jid,
                                                    body=salt.utils.json.dumps(data))
754821
def publish(self, load):
        '''
        
        '''
        payload = {'enc': 'aes'}
        crypticle = salt.crypt.Crypticle(self.opts, salt.master.SMaster.secrets['aes']['secret'].value)
        payload['load'] = crypticle.dumps(load)
        if self.opts['sign_pub_messages']:
            master_pem_path = os.path.join(self.opts['pki_dir'], 'master.pem')
            log.debug("Signing data packet")
            payload['sig'] = salt.crypt.sign_message(master_pem_path, payload['load'])
        int_payload = {'payload': self.serial.dumps(payload)}

        # add some targeting stuff for lists only (for now)
        if load['tgt_type'] == 'list':
            int_payload['topic_lst'] = load['tgt']

        # If zmq_filtering is enabled, target matching has to happen master side
        match_targets = ["pcre", "glob", "list"]
        if self.opts['zmq_filtering'] and load['tgt_type'] in match_targets:
            # Fetch a list of minions that match
            _res = self.ckminions.check_minions(load['tgt'],
                                                tgt_type=load['tgt_type'])
            match_ids = _res['minions']

            log.debug("Publish Side Match: %s", match_ids)
            # Send list of miions thru so zmq can target them
            int_payload['topic_lst'] = match_ids
        payload = self.serial.dumps(int_payload)
        log.debug(
            'Sending payload to publish daemon. jid=%s size=%d',
            load.get('jid', None), len(payload),
        )
        if not self.pub_sock:
            self.pub_connect()
        self.pub_sock.send(payload)
        log.debug('Sent payload to publish daemon.')
360699
def get_full_returns(self, jid, minions, timeout=None):
        '''
        
        '''
        # TODO: change this from ret to return... or the other way.
        #       Its inconsistent, we should pick one

        ret = {}
        # create the iterator-- since we want to get anyone in the middle
        event_iter = self.get_event_iter_returns(jid, minions, timeout=timeout)

        try:
            data = self.returners['{0}.get_jid'.format(self.opts['master_job_cache'])](jid)
        except Exception as exc:
            raise SaltClientError('Returner {0} could not fetch jid data. '
                                  'Exception details: {1}'.format(
                                      self.opts['master_job_cache'],
                                      exc))
        for minion in data:
            m_data = {}
            if 'return' in data[minion]:
                m_data['ret'] = data[minion].get('return')
            else:
                m_data['ret'] = data[minion].get('return')
            if 'out' in data[minion]:
                m_data['out'] = data[minion]['out']
            if minion in ret:
                ret[minion].update(m_data)
            else:
                ret[minion] = m_data

        # if we have all the minion returns, lets just return
        if len(set(ret).intersection(minions)) >= len(minions):
            return ret

        # otherwise lets use the listener we created above to get the rest
        for event_ret in event_iter:
            # if nothing in the event_ret, skip
            if event_ret == {}:
                time.sleep(0.02)
                continue
            for minion, m_data in six.iteritems(event_ret):
                if minion in ret:
                    ret[minion].update(m_data)
                else:
                    ret[minion] = m_data

            # are we done yet?
            if len(set(ret).intersection(minions)) >= len(minions):
                return ret

        # otherwise we hit the timeout, return what we have
        return ret
1032507
def get_minions():
    '''
    
    '''
    conn, mdb = _get_conn(ret=None)
    ret = []
    name = mdb.saltReturns.distinct('minion')
    ret.append(name)
    return ret
1362320
def __verify_minion_publish(self, load):
        '''
        
        '''
        # Verify that the load is valid
        if 'peer' not in self.opts:
            return False
        if not isinstance(self.opts['peer'], dict):
            return False
        if any(key not in load for key in ('fun', 'arg', 'tgt', 'ret', 'id')):
            return False
        # If the command will make a recursive publish don't run
        if re.match('publish.*', load['fun']):
            return False
        # Check the permissions for this minion
        perms = []
        for match in self.opts['peer']:
            if re.match(match, load['id']):
                # This is the list of funcs/modules!
                if isinstance(self.opts['peer'][match], list):
                    perms.extend(self.opts['peer'][match])
        if ',' in load['fun']:
            # 'arg': [['cat', '/proc/cpuinfo'], [], ['foo']]
            load['fun'] = load['fun'].split(',')
            arg_ = []
            for arg in load['arg']:
                arg_.append(arg.split())
            load['arg'] = arg_
        return self.ckminions.auth_check(
                perms,
                load['fun'],
                load['arg'],
                load['tgt'],
                load.get('tgt_type', 'glob'),
                publish_validate=True)
1137552
def _pillar(self, load):
        '''
        
        '''
        if any(key not in load for key in ('id', 'grains')):
            return False
#        pillar = salt.pillar.Pillar(
        log.debug('Master _pillar using ext: %s', load.get('ext'))
        pillar = salt.pillar.get_pillar(
                self.opts,
                load['grains'],
                load['id'],
                load.get('saltenv', load.get('env')),
                load.get('ext'),
                self.mminion.functions,
                pillar_override=load.get('pillar_override', {}))
        data = pillar.compile_pillar()
        if self.opts.get('minion_data_cache', False):
            self.cache.store('minions/{0}'.format(load['id']),
                             'data',
                             {'grains': load['grains'], 'pillar': data})
            if self.opts.get('minion_data_cache_events') is True:
                self.event.fire_event({'comment': 'Minion data cache refresh'}, salt.utils.event.tagify(load['id'], 'refresh', 'minion'))
        return data
129993
def get_minions():
    '''
    
    '''
    query = '''SELECT DISTINCT minion_id
               FROM {keyspace}.minions;'''.format(keyspace=_get_keyspace())

    ret = []

    # cassandra_cql.cql_query may raise a CommandExecutionError
    try:
        data = __salt__['cassandra_cql.cql_query'](query)
        if data:
            for row in data:
                minion = row.get('minion_id')
                if minion:
                    ret.append(minion)
    except CommandExecutionError:
        log.critical('Could not get the list of minions.')
        raise
    except Exception as e:
        log.critical(
            'Unexpected error while getting list of minions: %s', e)
        raise

    return ret
898416
def _target_load(self, load):
        '''
        
        '''
        mp_call = _metaproxy_call(self.opts, 'target_load')
        return mp_call(self, load)
361334
def fire_master(self, data, tag, preload=None):
        '''
        
        '''
        load = {}
        if preload:
            load.update(preload)

        load.update({
            'id': self.opts['id'],
            'tag': tag,
            'data': data,
            'cmd': '_minion_event',
            'tok': self.auth.gen_token(b'salt'),
        })

        channel = salt.transport.client.ReqChannel.factory(self.opts)
        try:
            channel.send(load)
        except Exception:
            pass
        finally:
            channel.close()
        return True
450704
def ext_pillar(minion_id,  # pylint: disable=W0613
               pillar,  # pylint: disable=W0613
               command):
    '''
    
    '''
    try:
        command = command.replace('%s', minion_id)
        return salt.utils.json.loads(__salt__['cmd.run'](command))
    except Exception:
        log.critical('JSON data from %s failed to parse', command)
        return {}
1114157
def returner(load):
    '''
    
    '''
    cb_ = _get_connection()

    hn_key = '{0}/{1}'.format(load['jid'], load['id'])
    try:
        ret_doc = {'return': load['return'],
                   'full_ret': salt.utils.json.dumps(load)}

        cb_.add(hn_key,
               ret_doc,
               ttl=_get_ttl(),
               )
    except couchbase.exceptions.KeyExistsError:
        log.error(
            'An extra return was detected from minion %s, please verify '
            'the minion, this could be a replay attack', load['id']
        )
        return False
87442
def clear_cached_minion_data(self,
                                 clear_pillar=False,
                                 clear_grains=False,
                                 clear_mine=False,
                                 clear_mine_func=None):
        '''
        
        '''
        clear_what = []
        if clear_pillar:
            clear_what.append('pillar')
        if clear_grains:
            clear_what.append('grains')
        if clear_mine:
            clear_what.append('mine')
        if clear_mine_func is not None:
            clear_what.append('mine_func: \'{0}\''.format(clear_mine_func))
        if not clear_what:
            log.debug('No cached data types specified for clearing.')
            return False

        minion_ids = self._tgt_to_list()
        log.debug('Clearing cached %s data for: %s',
                  ', '.join(clear_what),
                  minion_ids)
        if clear_pillar == clear_grains:
            # clear_pillar and clear_grains are both True or both False.
            # This means we don't deal with pillar/grains caches at all.
            grains = {}
            pillars = {}
        else:
            # Unless both clear_pillar and clear_grains are True, we need
            # to read in the pillar/grains data since they are both stored
            # in the same file, 'data.p'
            grains, pillars = self._get_cached_minion_data(*minion_ids)
        try:
            c_minions = self.cache.list('minions')
            for minion_id in minion_ids:
                if not salt.utils.verify.valid_id(self.opts, minion_id):
                    continue

                if minion_id not in c_minions:
                    # Cache bank for this minion does not exist. Nothing to do.
                    continue
                bank = 'minions/{0}'.format(minion_id)
                minion_pillar = pillars.pop(minion_id, False)
                minion_grains = grains.pop(minion_id, False)
                if ((clear_pillar and clear_grains) or
                    (clear_pillar and not minion_grains) or
                    (clear_grains and not minion_pillar)):
                    # Not saving pillar or grains, so just delete the cache file
                    self.cache.flush(bank, 'data')
                elif clear_pillar and minion_grains:
                    self.cache.store(bank, 'data', {'grains': minion_grains})
                elif clear_grains and minion_pillar:
                    self.cache.store(bank, 'data', {'pillar': minion_pillar})
                if clear_mine:
                    # Delete the whole mine file
                    self.cache.flush(bank, 'mine')
                elif clear_mine_func is not None:
                    # Delete a specific function from the mine file
                    mine_data = self.cache.fetch(bank, 'mine')
                    if isinstance(mine_data, dict):
                        if mine_data.pop(clear_mine_func, False):
                            self.cache.store(bank, 'mine', mine_data)
        except (OSError, IOError):
            return True
        return True
1252787
def delete(fun):
    '''
    
    '''
    if __opts__['file_client'] == 'local':
        data = __salt__['data.get']('mine_cache')
        if isinstance(data, dict) and fun in data:
            del data[fun]
        return __salt__['data.update']('mine_cache', data)
    load = {
            'cmd': '_mine_delete',
            'id': __opts__['id'],
            'fun': fun,
    }
    return _mine_send(load, __opts__)
928726
def returner(ret):
    '''
    
    '''
    serv = _get_serv(ret)
    minion = ret['id']
    jid = ret['jid']
    fun = ret['fun']
    rets = salt.utils.json.dumps(ret)
    serv.set('{0}:{1}'.format(jid, minion), rets)  # cache for get_jid
    serv.set('{0}:{1}'.format(fun, minion), rets)  # cache for get_fun

    # The following operations are neither efficient nor atomic.
    # If there is a way to make them so, this should be updated.
    _append_list(serv, 'minions', minion)
    _append_list(serv, 'jids', jid)
1100335
def __verify_minion_publish(self, clear_load):
        '''
        
        '''
        # Verify that the load is valid
        if 'peer' not in self.opts:
            return False
        if not isinstance(self.opts['peer'], dict):
            return False
        if any(key not in clear_load for key in ('fun', 'arg', 'tgt', 'ret', 'tok', 'id')):
            return False
        # If the command will make a recursive publish don't run
        if clear_load['fun'].startswith('publish.'):
            return False
        # Check the permissions for this minion
        if not self.__verify_minion(clear_load['id'], clear_load['tok']):
            # The minion is not who it says it is!
            # We don't want to listen to it!
            log.warning(
                'Minion id %s is not who it says it is and is attempting '
                'to issue a peer command', clear_load['id']
            )
            return False
        clear_load.pop('tok')
        perms = []
        for match in self.opts['peer']:
            if re.match(match, clear_load['id']):
                # This is the list of funcs/modules!
                if isinstance(self.opts['peer'][match], list):
                    perms.extend(self.opts['peer'][match])
        if ',' in clear_load['fun']:
            # 'arg': [['cat', '/proc/cpuinfo'], [], ['foo']]
            clear_load['fun'] = clear_load['fun'].split(',')
            arg_ = []
            for arg in clear_load['arg']:
                arg_.append(arg.split())
            clear_load['arg'] = arg_

        # finally, check the auth of the load
        return self.ckminions.auth_check(
            perms,
            clear_load['fun'],
            clear_load['arg'],
            clear_load['tgt'],
            clear_load.get('tgt_type', 'glob'),
            publish_validate=True)
197634
def get_iter_returns(
            self,
            jid,
            minions,
            timeout=None,
            tgt='*',
            tgt_type='glob',
            expect_minions=False,
            block=True,
            **kwargs):
        '''
        
        '''
        if not isinstance(minions, set):
            if isinstance(minions, six.string_types):
                minions = set([minions])
            elif isinstance(minions, (list, tuple)):
                minions = set(list(minions))

        if timeout is None:
            timeout = self.opts['timeout']
        gather_job_timeout = int(kwargs.get('gather_job_timeout', self.opts['gather_job_timeout']))
        start = int(time.time())

        # timeouts per minion, id_ -> timeout time
        minion_timeouts = {}

        found = set()
        missing = set()
        # Check to see if the jid is real, if not return the empty dict
        try:
            if self.returners['{0}.get_load'.format(self.opts['master_job_cache'])](jid) == {}:
                log.warning('jid does not exist')
                yield {}
                # stop the iteration, since the jid is invalid
                raise StopIteration()
        except Exception as exc:
            log.warning('Returner unavailable: %s', exc, exc_info_on_loglevel=logging.DEBUG)
        # Wait for the hosts to check in
        last_time = False
        # iterator for this job's return
        if self.opts['order_masters']:
            # If we are a MoM, we need to gather expected minions from downstreams masters.
            ret_iter = self.get_returns_no_block('(salt/job|syndic/.*)/{0}'.format(jid), 'regex')
        else:
            ret_iter = self.get_returns_no_block('salt/job/{0}'.format(jid))
        # iterator for the info of this job
        jinfo_iter = []
        # open event jids that need to be un-subscribed from later
        open_jids = set()
        timeout_at = time.time() + timeout
        gather_syndic_wait = time.time() + self.opts['syndic_wait']
        # are there still minions running the job out there
        # start as True so that we ping at least once
        minions_running = True
        log.debug(
            'get_iter_returns for jid %s sent to %s will timeout at %s',
            jid, minions, datetime.fromtimestamp(timeout_at).time()
        )
        while True:
            # Process events until timeout is reached or all minions have returned
            for raw in ret_iter:
                # if we got None, then there were no events
                if raw is None:
                    break
                if 'minions' in raw.get('data', {}):
                    minions.update(raw['data']['minions'])
                    if 'missing' in raw.get('data', {}):
                        missing.update(raw['data']['missing'])
                    continue
                if 'return' not in raw['data']:
                    continue
                if kwargs.get('raw', False):
                    found.add(raw['data']['id'])
                    yield raw
                else:
                    found.add(raw['data']['id'])
                    ret = {raw['data']['id']: {'ret': raw['data']['return']}}
                    if 'out' in raw['data']:
                        ret[raw['data']['id']]['out'] = raw['data']['out']
                    if 'retcode' in raw['data']:
                        ret[raw['data']['id']]['retcode'] = raw['data']['retcode']
                    if 'jid' in raw['data']:
                        ret[raw['data']['id']]['jid'] = raw['data']['jid']
                    if kwargs.get('_cmd_meta', False):
                        ret[raw['data']['id']].update(raw['data'])
                    log.debug('jid %s return from %s', jid, raw['data']['id'])
                    yield ret

            # if we have all of the returns (and we aren't a syndic), no need for anything fancy
            if len(found.intersection(minions)) >= len(minions) and not self.opts['order_masters']:
                # All minions have returned, break out of the loop
                log.debug('jid %s found all minions %s', jid, found)
                break
            elif len(found.intersection(minions)) >= len(minions) and self.opts['order_masters']:
                if len(found) >= len(minions) and len(minions) > 0 and time.time() > gather_syndic_wait:
                    # There were some minions to find and we found them
                    # However, this does not imply that *all* masters have yet responded with expected minion lists.
                    # Therefore, continue to wait up to the syndic_wait period (calculated in gather_syndic_wait) to see
                    # if additional lower-level masters deliver their lists of expected
                    # minions.
                    break
            # If we get here we may not have gathered the minion list yet. Keep waiting
            # for all lower-level masters to respond with their minion lists

            # let start the timeouts for all remaining minions

            for id_ in minions - found:
                # if we have a new minion in the list, make sure it has a timeout
                if id_ not in minion_timeouts:
                    minion_timeouts[id_] = time.time() + timeout

            # if the jinfo has timed out and some minions are still running the job
            # re-do the ping
            if time.time() > timeout_at and minions_running:
                # since this is a new ping, no one has responded yet
                jinfo = self.gather_job_info(jid, list(minions - found), 'list', **kwargs)
                minions_running = False
                # if we weren't assigned any jid that means the master thinks
                # we have nothing to send
                if 'jid' not in jinfo:
                    jinfo_iter = []
                else:
                    jinfo_iter = self.get_returns_no_block('salt/job/{0}'.format(jinfo['jid']))
                timeout_at = time.time() + gather_job_timeout
                # if you are a syndic, wait a little longer
                if self.opts['order_masters']:
                    timeout_at += self.opts.get('syndic_wait', 1)

            # check for minions that are running the job still
            for raw in jinfo_iter:
                # if there are no more events, lets stop waiting for the jinfo
                if raw is None:
                    break
                try:
                    if raw['data']['retcode'] > 0:
                        log.error('saltutil returning errors on minion %s', raw['data']['id'])
                        minions.remove(raw['data']['id'])
                        break
                except KeyError as exc:
                    # This is a safe pass. We're just using the try/except to
                    # avoid having to deep-check for keys.
                    missing_key = exc.__str__().strip('\'"')
                    if missing_key == 'retcode':
                        log.debug('retcode missing from client return')
                    else:
                        log.debug(
                            'Passing on saltutil error. Key \'%s\' missing '
                            'from client return. This may be an error in '
                            'the client.', missing_key
                        )
                # Keep track of the jid events to unsubscribe from later
                open_jids.add(jinfo['jid'])

                # TODO: move to a library??
                if 'minions' in raw.get('data', {}):
                    minions.update(raw['data']['minions'])
                    continue
                if 'syndic' in raw.get('data', {}):
                    minions.update(raw['syndic'])
                    continue
                if 'return' not in raw.get('data', {}):
                    continue

                # if the job isn't running there anymore... don't count
                if raw['data']['return'] == {}:
                    continue

                # if the minion throws an exception containing the word "return"
                # the master will try to handle the string as a dict in the next
                # step. Check if we have a string, log the issue and continue.
                if isinstance(raw['data']['return'], six.string_types):
                    log.error("unexpected return from minion: %s", raw)
                    continue

                if 'return' in raw['data']['return'] and \
                    raw['data']['return']['return'] == {}:
                    continue

                # if we didn't originally target the minion, lets add it to the list
                if raw['data']['id'] not in minions:
                    minions.add(raw['data']['id'])
                # update this minion's timeout, as long as the job is still running
                minion_timeouts[raw['data']['id']] = time.time() + timeout
                # a minion returned, so we know its running somewhere
                minions_running = True

            # if we have hit gather_job_timeout (after firing the job) AND
            # if we have hit all minion timeouts, lets call it
            now = time.time()
            # if we have finished waiting, and no minions are running the job
            # then we need to see if each minion has timedout
            done = (now > timeout_at) and not minions_running
            if done:
                # if all minions have timeod out
                for id_ in minions - found:
                    if now < minion_timeouts[id_]:
                        done = False
                        break
            if done:
                break

            # don't spin
            if block:
                time.sleep(0.01)
            else:
                yield

        # If there are any remaining open events, clean them up.
        if open_jids:
            for jid in open_jids:
                self.event.unsubscribe(jid)

        if expect_minions:
            for minion in list((minions - found)):
                yield {minion: {'failed': True}}

        # Filter out any minions marked as missing for which we received
        # returns (prevents false events sent due to higher-level masters not
        # knowing about lower-level minions).
        missing -= found

        # Report on missing minions
        if missing:
            for minion in missing:
                yield {minion: {'failed': True}}