def ec2_tags():

    boto_version = StrictVersion(boto.__version__)
    required_boto_version = StrictVersion('2.8.0')
    if boto_version < required_boto_version:
        log.error("%s: installed boto version %s < %s, can't find ec2_tags",
                __name__, boto_version, required_boto_version)
        return None

    if not _on_ec2():
        log.info("%s: not an EC2 instance, skipping", __name__)
        return None

    (instance_id, region) = _get_instance_info()
    credentials = _get_credentials()
    if not credentials:
        log.error("%s: no AWS credentials found, see documentation for how to provide them.", __name__)
        return None

    # Connect to EC2 and parse the Roles tags for this instance
    conn = boto.ec2.connect_to_region(region,
            aws_access_key_id=credentials['access_key'],
            aws_secret_access_key=credentials['secret_key'])

    tags = {}
    try:
        reservation = conn.get_all_instances(instance_ids=[ instance_id ])[0]
        instance = reservation.instances[0]
        tags = instance.tags
    except IndexError, e:
        log.error("Couldn't retrieve instance information: %s", e)
        return None

def ec2_tags():
    boto_version = StrictVersion(boto.__version__)
    required_boto_version = StrictVersion('2.8.0')
    if boto_version < required_boto_version:
        log.error("Installed boto version %s < %s, can't find ec2_tags",
                  boto_version, required_boto_version)
        return None

    if not _on_ec2():
        log.info("Not an EC2 instance, skipping")
        return None

    instance_id, region = _get_instance_info()
    credentials = _get_credentials()

    # Connect to EC2 and parse the Roles tags for this instance
    if not (credentials['access_key'] and credentials['secret_key']):
        log.error("No AWS credentials found, see documentation for how to provide them.")
        return None

    try:
        conn = boto.ec2.connect_to_region(
            region,
            aws_access_key_id=credentials['access_key'],
            aws_secret_access_key=credentials['secret_key'],
        )
    except Exception, e:
        log.error("Could not get AWS connection: %s", e)
        return None

 def _handle_aes(self, load):
     '''
     Takes the aes encrypted load, decrypts is and runs the encapsulated
     instructions
     '''
     try:
         data = self.crypticle.loads(load)
     except AuthenticationError:
         self.authenticate()
         data = self.crypticle.loads(load)
     # Verify that the publication is valid
     if 'tgt' not in data or 'jid' not in data or 'fun' not in data \
        or 'arg' not in data:
         return
     # Verify that the publication applies to this minion
     if 'tgt_type' in data:
         if not getattr(self.matcher,
                        '{0}_match'.format(data['tgt_type']))(data['tgt']):
             return
     else:
         if not self.matcher.glob_match(data['tgt']):
             return
     # If the minion does not have the function, don't execute,
     # this prevents minions that could not load a minion module
     # from returning a predictable exception
     #if data['fun'] not in self.functions:
     #    return
     if 'user' in data:
         log.info(('User {0[user]} Executing command {0[fun]} with jid '
                   '{0[jid]}'.format(data)))
     else:
         log.info(('Executing command {0[fun]} with jid {0[jid]}'
                   .format(data)))
     log.debug('Command details {0}'.format(data))
     self._handle_decoded_payload(data)

 def start(self):
     '''
     we override start() just for our log message
     '''
     log.info("starting salt-eventsd daemon")
     # leave the startup to the supers daemon, thats where all 
     # the daemonizing and double-forking takes place
     super(SaltEventsDaemon, self).start()

 def run(self):
     '''
     the method automatically called by start() from
     our parent class 
     '''
     log.info("initializing event listener")
     self.pid = self._get_pid()
     self._write_state()
     self.listen()

 def _return_pub(self, ret, ret_cmd='_return'):
     '''
     Return the data from the executed command to the master server
     '''
     if self.opts['multiprocessing']:
         fn_ = os.path.join(self.proc_dir, ret['jid'])
         if os.path.isfile(fn_):
             try:
                 os.remove(fn_)
             except (OSError, IOError):
                 # The file is gone already
                 pass
     log.info('Returning information for job: {0}'.format(ret['jid']))
     sreq = salt.payload.SREQ(self.opts['master_uri'])
     if ret_cmd == '_syndic_return':
         load = {'cmd': ret_cmd,
                 'jid': ret['jid'],
                 'id': self.opts['id']}
         load['return'] = {}
         for key, value in ret.items():
             if key == 'jid' or key == 'fun':
                 continue
             load['return'][key] = value
     else:
         load = {'return': ret['return'],
                 'cmd': ret_cmd,
                 'jid': ret['jid'],
                 'id': self.opts['id']}
     try:
         if hasattr(self.functions[ret['fun']], '__outputter__'):
             oput = self.functions[ret['fun']].__outputter__
             if isinstance(oput, string_types):
                 load['out'] = oput
     except KeyError:
         pass
     try:
         ret_val = sreq.send('aes', self.crypticle.dumps(load))
     except SaltReqTimeoutError:
         ret_val = ''
     if isinstance(ret_val, string_types) and not ret_val:
         # The master AES key has changed, reauth
         self.authenticate()
         ret_val = sreq.send('aes', self.crypticle.dumps(load))
     if self.opts['cache_jobs']:
         # Local job cache has been enabled
         fn_ = os.path.join(
             self.opts['cachedir'],
             'minion_jobs',
             load['jid'],
             'return.p')
         jdir = os.path.dirname(fn_)
         if not os.path.isdir(jdir):
             os.makedirs(jdir)
         salt.utils.fopen(fn_, 'w+').write(self.serial.dumps(ret))
     return ret_val

    def __init__(self,
                 config='/etc/salt/eventsd'):

        # retrieve current settings from the config file
        self.opts = None
        self._read_yaml(config)

        # make sure we have a 'general' section
        if 'general' in self.opts.keys():
            self.gen_opts = self.opts['general']

        self._init_logger()
        log.info("loaded config from {0}".format(config))

    def _init_worker(self, qdata):
        '''
        write a collection of events to the database. every invocation of
        this methoed creates its own thread that writes into the database
        '''
        self.threads_cre += 1

        log.info("starting worker #{0}".format(self.threads_cre))

        # make sure we pass a copy of the list
        worker = SaltEventsdWorker(list(qdata),
                                   self.threads_cre,
                                   self.event_map,
                                   self.backends,
                                   **self.opts)

        worker.start()
        self.running_workers.append(worker)

def get_id():
    '''
    Guess the id of the minion.

    - If socket.getfqdn() returns us something other than localhost, use it
    - Check /etc/hosts for something that isn't localhost that maps to 127.*
    - Look for a routeable / public IP
    - A private IP is better than a loopback IP
    - localhost may be better than killing the minion
    '''

    log.debug('Guessing ID. The id can be explicitly in set {0}'
              .format('/etc/salt/minion'))
    fqdn = socket.getfqdn()
    if 'localhost' != fqdn:
        log.info('Found minion id from getfqdn(): {0}'.format(fqdn))
        return fqdn, False

    # Can /etc/hosts help us?
    try:
        # TODO Add Windows host file support
        with open('/etc/hosts') as f:
            line = f.readline()
            while line:
                names = line.split()
                ip = names.pop(0)
                if ip.startswith('127.'):
                    for name in names:
                        if name != 'localhost':
                            log.info('Found minion id in hosts file: {0}'
                                     .format(name))
                            return name, False
                line = f.readline()
    except Exception:
        pass

    # What IP addresses do we have?
    ip_addresses = [salt.utils.socket_util.IPv4Address(a) for a
                    in salt.utils.socket_util.ip4_addrs()
                    if not a.startswith('127.')]

    for a in ip_addresses:
        if not a.is_private:
            log.info('Using public ip address for id: {0}'.format(a))
            return str(a), True

    if ip_addresses:
        a = ip_addresses.pop(0)
        log.info('Using private ip address for id: {0}'.format(a))
        return str(a), True

    log.error('No id found, falling back to localhost')
    return 'localhost', False

    def _init_worker(self, qdata):
        '''
        The method dumps the data into a worker thread which
        handles pushing the data into different backends.
        '''
        self.threads_cre += 1

        log.info("Starting worker #{0}".format(self.threads_cre))

        # make sure we pass a copy of the list
        worker = SaltEventsdWorker(
            list(qdata),
            self.threads_cre,
            self.event_map,
            self.backends,
            **self.opts
        )

        worker.start()
        self.running_workers.append(worker)

    def stop(self,
             signal,
             frame):
        '''
        we override stop() to brake our main loop
        and have a pretty log message
        '''
        log.info("received signal {0}".format(signal))

        # if we have running workers, run through all and join() the ones
        # that have finished. if we still have running workers after that,
        # wait 5 secs for the rest and then exit. Maybe we should improv
        # this a litte bit more
        if( len(self.running_workers) > 0 ):
            clean_workers = []

            for count in range(0, 2):
                for worker in self.running_workers:
                    if worker.isAlive():
                        clean_workers.append(worker)
                    else:
                        worker.join()
                        log.debug("joined worker #{0}".format(worker.getName()))

                if( len(clean_workers) > 0 ):
                    log.info("waiting 5secs for remaining workers..")
                    time.sleep(5)
                else:
                    break

        log.info("salt-eventsd has shut down")

        # leave the cleanup to the supers stop
        super(SaltEventsDaemon, self).stop(signal, frame)

    def _write_state(self):
        '''
        writes a current status to the defined status-file
        this includes the current pid, events received/handled
        and threads created/joined
        '''
        try:
            # write the info to the specified log
            statf = open(self.state_file, 'w')
            statf.writelines(simplejson.dumps({'events_received':self.events_rec,
                                               'events_handled':self.events_han,
                                               'threads_created':self.threads_cre,
                                               'threads_joined':self.threads_join}
                                             ))
            # if we have the same pid as the pidfile, we are the running daemon
            # and also print the current counters to the logfile with 'info'
            if( os.getpid() == self.pid ):
                log.info("running with pid {0}".format(self.pid))
                log.info("events (han/recv): {0}/{1}".format(self.events_han,
                                                             self.events_rec))
                log.info("threads (cre/joi):{0}/{1}".format(self.threads_cre,
                                                            self.threads_join))


            statf.write("\n")
            statf.close()
            sys.stdout.flush()
        except IOError as ioerr:
            log.critical("Failed to write state to {0}".format(self.state_file))
            log.exception(ioerr)
        except OSError as oserr:
            log.critical("Failed to write state to {0}".format(self.state_file))
            log.exception(oserr)

    def _init_events(self, events={}):
        '''
        Creates a dict of precompiled regexes for all defined events
        from config for maximum performance.
        '''
        self.event_map = events
        # we precompile all regexes
        log.info("Initialising events...")

        for key in events.keys():
            # we compile the regex configured in the config
            self.event_map[key]['tag'] = compile(events[key]['tag'])
            log.info("Added event '{0}'".format(key))

            # if subevents are configured, also update them with
            # regex-matching object
            if 'subs' in events[key]:
                for sub_ev in events[key]['subs'].keys():
                    try:
                        self.event_map[key]['subs'][sub_ev]['fun'] = compile(events[key]['subs'][sub_ev]['fun'])
                    except KeyError:
                        pass

                    try:
                        self.event_map[key]['subs'][sub_ev]['tag'] = compile(events[key]['subs'][sub_ev]['tag'])
                    except KeyError:
                        pass

                    log.info("Added sub-event '{0}->{1}'".format(key, sub_ev))

    def _init_events(self, events={}):
        '''
        this is used to tell the class about the events it should handle.
        it has to be a dictionary with appropriate mappings in it. see the
        config file for examples on how to compose the dict. each entry is
        converted to a precompiled regex for maximum flexibility
        '''
        self.event_map = events
        # we precompile all regexes
        log.info("initialising events...")
        for key in events.keys():
            # we compile the regex configured in the config
            self.event_map[key]['tag'] = compile( events[key]['tag'] )
            log.info("Added event '{0}'".format(key))

            # if subevents are configured, also update them with 
            # regex-macthing object
            if( events[key].has_key('subs') ):
                for sub_ev in events[key]['subs'].keys():
                    try:
                        self.event_map[key]['subs'][sub_ev]['fun'] = compile(events[key]['subs'][sub_ev]['fun'])
                    except KeyError:
                        pass

                    try:
                        self.event_map[key]['subs'][sub_ev]['tag'] = compile(events[key]['subs'][sub_ev]['tag'])
                    except KeyError:
                        pass

                    log.info("Added sub-event '{0}->{1}'".format(key,
                                                                 sub_ev))

def ec2_tags():

    boto_version = StrictVersion(boto.__version__)
    required_boto_version = StrictVersion('2.8.0')
    if boto_version < required_boto_version:
        log.error("%s: installed boto version %s < %s, can't find ec2_tags",
                __name__, boto_version, required_boto_version)
        return None

    if not _on_ec2():
        log.info("%s: not an EC2 instance, skipping", __name__)
        return None

    (instance_id, region) = _get_instance_info()
    credentials = _get_credentials()

    # Connect to EC2 and parse the Roles tags for this instance
    try:
        conn = boto.ec2.connect_to_region(region,
                aws_access_key_id=credentials['access_key'],
                aws_secret_access_key=credentials['secret_key'])
    except:
        if not (credentials['access_key'] and credentials['secret_key']):
            log.error("%s: no AWS credentials found, see documentation for how to provide them.", __name__)
            return None
        else:
            log.error("%s: invalid AWS credentials found, see documentation for how to provide them.", __name__)
            return None

    tags = {}
    try:
        _tags = conn.get_all_tags(filters={'resource-type': 'instance',
                'resource-id': instance_id})
        for tag in _tags:
            tags[tag.name] = tag.value
    except IndexError, e:
        log.error("Couldn't retrieve instance information: %s", e)
        return None

 def authenticate(self):
     '''
     Authenticate with the master, this method breaks the functional
     paradigm, it will update the master information from a fresh sign
     in, signing in can occur as often as needed to keep up with the
     revolving master aes key.
     '''
     log.debug(
         'Attempting to authenticate with the Salt Master at {0}'.format(
             self.opts['master_ip']
         )
     )
     auth = salt.crypt.Auth(self.opts)
     while True:
         creds = auth.sign_in()
         if creds != 'retry':
             log.info('Authentication with master successful!')
             break
         log.info('Waiting for minion key to be accepted by the master.')
         time.sleep(self.opts['acceptance_wait_time'])
     self.aes = creds['aes']
     self.publish_port = creds['publish_port']
     self.crypticle = salt.crypt.Crypticle(self.opts, self.aes)

 def send(self, enc, load, tries=1, timeout=60):
     '''
     Takes two arguments, the encryption type and the base payload
     '''
     payload = {'enc': enc}
     payload['load'] = load
     pkg = self.serial.dumps(payload)
     self.socket.send(pkg)
     self.poller.register(self.socket, zmq.POLLIN)
     tried = 0
     while True:
         polled = self.poller.poll(timeout * 1000)
         tried += 1
         if polled:
             break
         if tries > 1:
             log.info('SaltReqTimeoutError: after {0} seconds. (Try {1} of {2})'.format(
               timeout, tried, tries))
         if tried >= tries:
             self.clear_socket()
             raise SaltReqTimeoutError(
                 'SaltReqTimeoutError: after {0} seconds, ran {1} tries'.format(timeout * tried, tried)
             )
     return self.serial.loads(self.socket.recv())

def _linux_gpu_data():
    '''
    num_gpus: int
    gpus:
      - vendor: nvidia|amd|ati|...
        model: string
    '''
    lspci = salt.utils.which('lspci')
    if not lspci:
        log.info(
            'The `lspci` binary is not available on the system. GPU grains '
            'will not be available.'
        )
        return {}

    elif __opts__.get('enable_gpu_grains', None) is False:
        log.info(
            'Skipping lspci call because enable_gpu_grains was set to False '
            'in the config. GPU grains will not be available.'
        )
        return {}

    # dominant gpu vendors to search for (MUST be lowercase for matching below)
    known_vendors = ['nvidia', 'amd', 'ati', 'intel']

    devs = []
    try:
        lspci_out = __salt__['cmd.run']('lspci -vmm')

        cur_dev = {}
        error = False
        # Add a blank element to the lspci_out.splitlines() list,
        # otherwise the last device is not evaluated as a cur_dev and ignored.
        lspci_list = lspci_out.splitlines()
        lspci_list.append('')
        for line in lspci_list:
            # check for record-separating empty lines
            if line == '':
                if cur_dev.get('Class', '') == 'VGA compatible controller':
                    devs.append(cur_dev)
                # XXX; may also need to search for "3D controller"
                cur_dev = {}
                continue
            if re.match(r'^\w+:\s+.*', line):
                key, val = line.split(':', 1)
                cur_dev[key.strip()] = val.strip()
            else:
                error = True
                log.debug('Unexpected lspci output: \'{0}\''.format(line))

        if error:
            log.warn(
                'Error loading grains, unexpected linux_gpu_data output, '
                'check that you have a valid shell configured and '
                'permissions to run lspci command'
            )
    except OSError:
        pass

    gpus = []
    for gpu in devs:
        vendor_strings = gpu['Vendor'].lower().split()
        # default vendor to 'unknown', overwrite if we match a known one
        vendor = 'unknown'
        for name in known_vendors:
            # search for an 'expected' vendor name in the list of strings
            if name in vendor_strings:
                vendor = name
                break
        gpus.append({'vendor': vendor, 'model': gpu['Device']})

    grains = {}
    grains['num_gpus'] = len(gpus)
    grains['gpus'] = gpus
    return grains

    def run(self):
        '''
        Main loop of the ConCache, starts updates in intervals and
        answers requests from the MWorkers
        '''
        context = zmq.Context()
        # the socket for incoming cache requests
        creq_in = context.socket(zmq.REP)
        creq_in.setsockopt(zmq.LINGER, 100)
        creq_in.bind('ipc://' + self.cache_sock)

        # the socket for incoming cache-updates from workers
        cupd_in = context.socket(zmq.SUB)
        cupd_in.setsockopt(zmq.SUBSCRIBE, '')
        cupd_in.setsockopt(zmq.LINGER, 100)
        cupd_in.bind('ipc://' + self.update_sock)

        # the socket for the timer-event
        timer_in = context.socket(zmq.SUB)
        timer_in.setsockopt(zmq.SUBSCRIBE, '')
        timer_in.setsockopt(zmq.LINGER, 100)
        timer_in.connect('ipc://' + self.upd_t_sock)

        poller = zmq.Poller()
        poller.register(creq_in, zmq.POLLIN)
        poller.register(cupd_in, zmq.POLLIN)
        poller.register(timer_in, zmq.POLLIN)

        # our serializer
        serial = salt.payload.Serial(self.opts.get('serial', ''))

        # register a signal handler
        signal.signal(signal.SIGINT, self.signal_handler)

        # secure the sockets from the world
        self.secure()

        log.info('ConCache started')

        while self.running:

            # we check for new events with the poller
            try:
                socks = dict(poller.poll(1))
            except KeyboardInterrupt:
                self.stop()
            except zmq.ZMQError as zmq_err:
                log.error('ConCache ZeroMQ-Error occurred')
                log.exception(zmq_err)
                self.stop()

            # check for next cache-request
            if socks.get(creq_in) == zmq.POLLIN:
                msg = serial.loads(creq_in.recv())
                log.debug('ConCache Received request: {0}'.format(msg))

                # requests to the minion list are send as str's
                if isinstance(msg, str):
                    if msg == 'minions':
                        # Send reply back to client
                        reply = serial.dumps(self.minions)
                        creq_in.send(reply)

            # check for next cache-update from workers
            if socks.get(cupd_in) == zmq.POLLIN:
                new_c_data = serial.loads(cupd_in.recv())
                # tell the worker to exit
                #cupd_in.send(serial.dumps('ACK'))

                # check if the returned data is usable
                if not isinstance(new_c_data, list):
                    log.error('ConCache Worker returned unusable result')
                    del new_c_data
                    continue

                # the cache will receive lists of minions
                # 1. if the list only has 1 item, its from an MWorker, we append it
                # 2. if the list contains another list, its from a CacheWorker and
                #    the currently cached minions are replaced with that list
                # 3. anything else is considered malformed

                try:

                    if len(new_c_data) == 0:
                        log.debug('ConCache Got empty update from worker')
                        continue

                    data = new_c_data[0]

                    if isinstance(data, str):
                        if data not in self.minions:
                            log.debug('ConCache Adding minion {0} to cache'.format(new_c_data[0]))
                            self.minions.append(data)

                    elif isinstance(data, list):
                        log.debug('ConCache Replacing minion list from worker')
                        self.minions = data

                except IndexError:
                    log.debug('ConCache Got malformed result dict from worker')
#.........这里部分代码省略.........

def _linux_gpu_data():
    """
    num_gpus: int
    gpus:
      - vendor: nvidia|amd|ati|...
        model: string
    """
    lspci = salt.utils.which("lspci")
    if not lspci:
        log.info("The `lspci` binary is not available on the system. GPU grains " "will not be available.")
        return {}

    elif __opts__.get("enable_gpu_grains", None) is False:
        log.info(
            "Skipping lspci call because enable_gpu_grains was set to False "
            "in the config. GPU grains will not be available."
        )
        return {}

    # dominant gpu vendors to search for (MUST be lowercase for matching below)
    known_vendors = ["nvidia", "amd", "ati", "intel"]

    devs = []
    try:
        lspci_out = __salt__["cmd.run"]("lspci -vmm")

        cur_dev = {}
        error = False
        # Add a blank element to the lspci_out.splitlines() list,
        # otherwise the last device is not evaluated as a cur_dev and ignored.
        lspci_list = lspci_out.splitlines()
        lspci_list.append("")
        for line in lspci_list:
            # check for record-separating empty lines
            if line == "":
                if cur_dev.get("Class", "") == "VGA compatible controller":
                    devs.append(cur_dev)
                # XXX; may also need to search for "3D controller"
                cur_dev = {}
                continue
            if re.match(r"^\w+:\s+.*", line):
                key, val = line.split(":", 1)
                cur_dev[key.strip()] = val.strip()
            else:
                error = True
                log.debug("Unexpected lspci output: '{0}'".format(line))

        if error:
            log.warn(
                "Error loading grains, unexpected linux_gpu_data output, "
                "check that you have a valid shell configured and "
                "permissions to run lspci command"
            )
    except OSError:
        pass

    gpus = []
    for gpu in devs:
        vendor_strings = gpu["Vendor"].lower().split()
        # default vendor to 'unknown', overwrite if we match a known one
        vendor = "unknown"
        for name in known_vendors:
            # search for an 'expected' vendor name in the list of strings
            if name in vendor_strings:
                vendor = name
                break
        gpus.append({"vendor": vendor, "model": gpu["Device"]})

    grains = {}
    grains["num_gpus"] = len(gpus)
    grains["gpus"] = gpus
    return grains