Run :ref:`execution modules <all-salt.modules>` against subsets of minions

        .. versionadded:: 2016.3.0

        Wraps :py:meth:`salt.client.LocalClient.cmd_subset`
196958
def local_subset(self, *args, **kwargs):
        '''
        
        '''
        local = salt.client.get_local_client(mopts=self.opts)
        return local.cmd_subset(*args, **kwargs)
968898
def _minion_event(self, load):
        '''
        
        '''
        load = self.__verify_load(load, ('id', 'tok'))
        if load is False:
            return {}
        # Route to master event bus
        self.masterapi._minion_event(load)
        # Process locally
        self._handle_minion_event(load)
534284
def targets(tgt, tgt_type='glob', **kwargs):  # pylint: disable=W0613
    '''
    
    '''
    minions = salt.utils.minions.CkMinions(__opts__)
    _res = minions.check_minions(tgt, tgt_type)
    minions = _res['minions']

    ret = {}
    if not minions:
        return ret
    # log.debug(minions)

    cache = salt.cache.Cache(__opts__)

    roster_order = __opts__.get('roster_order', {
        'host': ('ipv6-private', 'ipv6-global', 'ipv4-private', 'ipv4-public')
    })

    ret = {}
    for minion_id in minions:
        try:
            minion = _load_minion(minion_id, cache)
        except LookupError:
            continue

        minion_res = copy.deepcopy(__opts__.get('roster_defaults', {}))
        for param, order in roster_order.items():
            if not isinstance(order, (list, tuple)):
                order = [order]
            for key in order:
                kres = _minion_lookup(minion_id, key, minion)
                if kres:
                    minion_res[param] = kres
                    break

        if 'host' in minion_res:
            ret[minion_id] = minion_res
        else:
            log.warning('Could not determine host information for minion %s', minion_id)

    log.debug('Roster lookup result: %s', ret)

    return ret
571532
def salt_call():
    '''
    
    '''
    import salt.cli.call
    if '' in sys.path:
        sys.path.remove('')
    client = salt.cli.call.SaltCall()
    _install_signal_handlers(client)
    client.run()
955622
def process_presence_events(self, salt_data, token, opts):
        '''
        
        '''
        log.debug('In presence')
        changed = False

        # check if any connections were dropped
        if set(salt_data['data'].get('lost', [])):
            dropped_minions = set(salt_data['data'].get('lost', []))
        else:
            dropped_minions = set(self.minions) - set(salt_data['data'].get('present', []))

        for minion in dropped_minions:
            changed = True
            log.debug('Popping %s', minion)
            self.minions.pop(minion, None)

        # check if any new connections were made
        if set(salt_data['data'].get('new', [])):
            log.debug('got new minions')
            new_minions = set(salt_data['data'].get('new', []))
            changed = True
        elif set(salt_data['data'].get('present', [])) - set(self.minions):
            log.debug('detected new minions')
            new_minions = set(salt_data['data'].get('present', [])) - set(self.minions)
            changed = True
        else:
            new_minions = []

        tgt = ','.join(new_minions)
        for mid in new_minions:
            log.debug('Adding minion')
            self.minions[mid] = {}

        if tgt:
            changed = True
            client = salt.netapi.NetapiClient(opts)
            client.run(
                {
                    'fun': 'grains.items',
                    'tgt': tgt,
                    'expr_type': 'list',
                    'mode': 'client',
                    'client': 'local',
                    'asynchronous': 'local_async',
                    'token': token,
                })

        if changed:
            self.publish_minions()
249238
def local(self, fun, tgt, **kwargs):
        '''
        
        '''
        self.client_cache['local'].cmd_async(tgt, fun, **kwargs)
806328
def process_presence_events(self, event_data, token, opts):
        '''
        
        '''
        tag = event_data['tag']
        event_info = event_data['data']

        minions_detected = event_info['present']
        curr_minions = six.iterkeys(self.minions)

        changed = False

        # check if any connections were dropped
        dropped_minions = set(curr_minions) - set(minions_detected)

        for minion in dropped_minions:
            changed = True
            self.minions.pop(minion, None)

        # check if any new connections were made
        new_minions = set(minions_detected) - set(curr_minions)

        tgt = ','.join(new_minions)

        if tgt:
            changed = True
            client = salt.netapi.NetapiClient(opts)
            client.run(
                {
                    'fun': 'grains.items',
                    'tgt': tgt,
                    'expr_type': 'list',
                    'mode': 'client',
                    'client': 'local',
                    'asynchronous': 'local_async',
                    'token': token,
                })

        if changed:
            self.publish_minions()
1181035
def minion_runner(self, clear_load):
        '''
        
        '''
        load = self.__verify_load(clear_load, ('fun', 'arg', 'id', 'tok'))
        if load is False:
            return {}
        else:
            return self.masterapi.minion_runner(clear_load)
250135
def _all_minions(self, expr=None):
        '''
        
        '''
        mlist = []
        for fn_ in salt.utils.data.sorted_ignorecase(os.listdir(os.path.join(self.opts['pki_dir'], self.acc))):
            if not fn_.startswith('.') and os.path.isfile(os.path.join(self.opts['pki_dir'], self.acc, fn_)):
                mlist.append(fn_)
        return {'minions': mlist, 'missing': []}
835014
def master_call(self, **kwargs):
        '''
        
        '''
        load = kwargs
        load['cmd'] = self.client
        channel = salt.transport.client.ReqChannel.factory(self.opts,
                                                           crypt='clear',
                                                           usage='master_call')
        try:
            ret = channel.send(load)
        finally:
            channel.close()
        if isinstance(ret, collections.Mapping):
            if 'error' in ret:
                salt.utils.error.raise_error(**ret['error'])
        return ret
700030
def __execute_cmd(name, cmd):
    '''
    
    '''
    return __salt__['cmd.run_all'](
        '{0} {1}'.format(salt.utils.path.which(name), cmd)
    )
992611
def versions():
    '''
    
    '''
    ret = {}
    client = salt.client.get_local_client(__opts__['conf_file'])
    try:
        minions = client.cmd('*', 'test.version', timeout=__opts__['timeout'])
    except SaltClientError as client_error:
        print(client_error)
        return ret

    labels = {
        -2: 'Minion offline',
        -1: 'Minion requires update',
        0: 'Up to date',
        1: 'Minion newer than master',
        2: 'Master',
    }

    version_status = {}

    master_version = salt.version.__saltstack_version__

    for minion in minions:
        if not minions[minion]:
            minion_version = False
            ver_diff = -2
        else:
            minion_version = salt.version.SaltStackVersion.parse(minions[minion])
            ver_diff = salt.utils.compat.cmp(minion_version, master_version)

        if ver_diff not in version_status:
            version_status[ver_diff] = {}
        if minion_version:
            version_status[ver_diff][minion] = minion_version.string
        else:
            version_status[ver_diff][minion] = minion_version

    # Add version of Master to output
    version_status[2] = master_version.string

    for key in version_status:
        if key == 2:
            ret[labels[key]] = version_status[2]
        else:
            for minion in sorted(version_status[key]):
                ret.setdefault(labels[key], {})[minion] = version_status[key][minion]
    return ret
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 {}
1275031
def _disbatch_local(self, chunk):
        '''
        
        '''
        # Generate jid and find all minions before triggering a job to subscribe all returns from minions
        full_return = chunk.pop('full_return', False)
        chunk['jid'] = salt.utils.jid.gen_jid(self.application.opts) 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 = []
        for minion in minions:
            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
        pub_data = yield self.saltclients['local'](*f_call.get('args', ()), **f_call.get('kwargs', {}))

        # if the job didn't publish, lets not wait around for nothing
        # TODO: set header??
        if 'jid' not in pub_data:
            for future in events:
                try:
                    future.set_result(None)
                except Exception:
                    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']}

        # minimum time required for return to complete. By default no waiting, if
        # we are a syndic then we must wait syndic_wait at a minimum
        min_wait_time = Future()
        min_wait_time.set_result(True)

        # wait syndic a while to avoid missing published events
        if self.application.opts['order_masters']:
            min_wait_time = tornado.gen.sleep(self.application.opts['syndic_wait'])

        # To ensure job_not_running and all_return are terminated by each other, communicate using a future
        is_timed_out = tornado.gen.sleep(self.application.opts['gather_job_timeout'])
        is_finished = Future()

        # ping until the job is not running, while doing so, if we see new minions returning
        # that they are running the job, add them to the list
        tornado.ioloop.IOLoop.current().spawn_callback(self.job_not_running, pub_data['jid'],
                                                      chunk['tgt'],
                                                      f_call['kwargs']['tgt_type'],
                                                      minions,
                                                      is_finished)

        def more_todo():
            '''
            Check if there are any more minions we are waiting on returns from
            '''
            return any(x is False for x in six.itervalues(minions))

        # here we want to follow the behavior of LocalClient.get_iter_returns
        # namely we want to wait at least syndic_wait (assuming we are a syndic)
        # and that there are no more jobs running on minions. We are allowed to exit
        # early if gather_job_timeout has been exceeded
        chunk_ret = {}
        while True:
            to_wait = events+[is_finished, is_timed_out]
            if not min_wait_time.done():
                to_wait += [min_wait_time]

            def cancel_inflight_futures():
                for event in to_wait:
                    if not event.done() and event is not is_timed_out:
                        event.set_result(None)
            f = yield Any(to_wait)
            try:
                # When finished entire routine, cleanup other futures and return result
                if f is is_finished or f is is_timed_out:
                    cancel_inflight_futures()
                    raise tornado.gen.Return(chunk_ret)
                elif f is min_wait_time:
                    if not more_todo():
                        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']:
                        if minion_id not in minions:
                            minions[minion_id] = False
                else:
                    chunk_ret[f_result['data']['id']] = f_result if full_return else 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()
                        raise tornado.gen.Return(chunk_ret)

            except TimeoutException:
                break
1268658
def compile_pillar(self):
        '''
        
        '''
        load = {'id': self.minion_id,
                'grains': self.grains,
                'saltenv': self.opts['saltenv'],
                'pillarenv': self.opts['pillarenv'],
                'pillar_override': self.pillar_override,
                'extra_minion_data': self.extra_minion_data,
                'ver': '2',
                'cmd': '_pillar'}
        if self.ext:
            load['ext'] = self.ext
        try:
            ret_pillar = yield self.channel.crypted_transfer_decode_dictentry(
                load,
                dictkey='pillar',
            )
        except Exception:
            log.exception('Exception getting pillar:')
            raise SaltClientError('Exception getting pillar.')

        if not isinstance(ret_pillar, dict):
            msg = ('Got a bad pillar from master, type {0}, expecting dict: '
                   '{1}').format(type(ret_pillar).__name__, ret_pillar)
            log.error(msg)
            # raise an exception! Pillar isn't empty, we can't sync it!
            raise SaltClientError(msg)
        raise tornado.gen.Return(ret_pillar)
266090
def run(cmdfile,subsetfile=None, lookupfile=None):
    
    #Define the test case pattern
    pattern="""class %s(basecase.%sCase):
    def setUp(self):
        self.obsmode="%s"
        self.spectrum="%s"
        self.subset=%s
        self.etcid="%s"
        self.setglobal(__file__)
        self.runpy()\n"""

    #Open the (params:etcid) dictionary if it exists
    if lookupfile is not None:
        f=open(lookupfile)
        lookup=pickle.load(f)
        f.close()
    else:
        lookup={}
        
    #Process the subsetfile if present
    subsetflag={}
    if subsetfile is not None:
        #Then this file contains a list of case names that should have a
        #subset flag set.
        f=open(subsetfile)
        for line in f:
            if not line.startswith('#') and len(line.strip())>0:
                cols=line.strip().split()
                subsetflag[cols[0]]=True
        f.close()

    #Open the main input and output files
    f=open(cmdfile)
    out=open(cmdfile.replace('.txt','.py'),'w')
    out.write("""from pytools import testutil
import sys
import basecase
""")

    #Set up some bookkeeping
    count={'countrate':0,'calcspec':0,'calcphot':0,'SpecSourcerateSpec':0,'thermback':0}
    dupcatcher={}
    dupcounter={'countrate':0,'calcspec':0,'calcphot':0,'SpecSourcerateSpec':0,'thermback':0}
    
    #Now start the main loop
    for line in f:
    
    #parse the line
        ktuple=line2ktuple(line)
        cmd,obsmode,spectrum=ktuple
        count[cmd]+=1
        casename="%sCase%d"%(cmd,count[cmd])        
        if ktuple in dupcatcher:
            dupcounter[cmd]+=1
            dupcatcher[ktuple].append(casename)
        else:
            dupcatcher[ktuple]=[casename]
            etcid=lookup.get(ktuple,None)
            if etcid is not None and len(etcid) == 1:
                etcid=etcid[0]
            defn=pattern%(casename,cmd,obsmode,spectrum,subsetflag.get(casename,False),etcid)
            out.write(defn)

    out.write("""\n\n
if __name__ == '__main__':
    if 'debug' in sys.argv:
        testutil.debug(__name__)
    else:
        testutil.testall(__name__,2)
""")

    f.close()

    for k in count:
        total= "#%s:%d - %d dup =%d\n"%(k,count[k],dupcounter[k],count[k]-dupcounter[k])
        sys.stdout.write(total)
        out.write(total)
    out.close()

        
    return dupcatcher
845437
def salt(self, module, target='*', args=None, kwargs=None, ssh=False):
        
        if ssh:
            return salt.salt_ssh(self, target, module, args, kwargs)
        else:
            return salt.salt_master(self, target, module, args, kwargs)