OpenStack Nova Compute Node资源使用情况

@manager.periodic_task
def _report_driver_status(self, context):
	curr_time = time.time()
	if curr_time - self._last_host_check > FLAGS.host_state_interval:
		self._last_host_check = curr_time
		LOG.info(_("Updating host status"))
		# This will grab info about the host and queue it
		# to be sent to the Schedulers.
		self.update_service_capabilities(
			self.driver.get_host_stats(refresh=True))

cfg.IntOpt('host_state_interval',
             default=120,
             help='Interval in seconds for querying the host status'),

class HostState(object):
    """Manages information about the compute node through libvirt"""
    def __init__(self, read_only):
        super(HostState, self).__init__()
        self.read_only = read_only
        self._stats = {}
        self.connection = None
        self.update_status()

    def get_host_stats(self, refresh=False):
        """Return the current state of the host.

        If 'refresh' is True, run update the stats first."""
        if refresh:
            self.update_status()
        return self._stats

    def update_status(self):
        """Retrieve status info from libvirt."""
        LOG.debug(_("Updating host stats"))
        if self.connection is None:
            self.connection = get_connection(self.read_only)
        data = {}
        data["vcpus"] = self.connection.get_vcpu_total()
        data["vcpus_used"] = self.connection.get_vcpu_used()
        data["cpu_info"] = utils.loads(self.connection.get_cpu_info())
        data["disk_total"] = self.connection.get_local_gb_total()
        data["disk_used"] = self.connection.get_local_gb_used()
        data["disk_available"] = data["disk_total"] - data["disk_used"]
        data["host_memory_total"] = self.connection.get_memory_mb_total()
        data["host_memory_free"] = (data["host_memory_total"] -
                                    self.connection.get_memory_mb_used())
        data["hypervisor_type"] = self.connection.get_hypervisor_type()
        data["hypervisor_version"] = self.connection.get_hypervisor_version()

        self._stats = data

        return data

def _schedule(self, context, topic, request_spec, *args, **kwargs):
	"""Returns a list of hosts that meet the required specs,
	ordered by their fitness.
	"""
	elevated = context.elevated()
	if topic != "compute":
		msg = _("Scheduler only understands Compute nodes (for now)")
		raise NotImplementedError(msg)

	instance_properties = request_spec['instance_properties']
	instance_type = request_spec.get("instance_type", None)

	cost_functions = self.get_cost_functions()
	config_options = self._get_configuration_options()

	filter_properties = kwargs.get('filter_properties', {})
	filter_properties.update({'context': context,
							'request_spec': request_spec,
							'config_options': config_options,
							'instance_type': instance_type})

	self.populate_filter_properties(request_spec,
									filter_properties)

	# Find our local list of acceptable hosts by repeatedly
	# filtering and weighing our options. Each time we choose a
	# host, we virtually consume resources on it so subsequent
	# selections can adjust accordingly.

	# unfiltered_hosts_dict is {host : ZoneManager.HostInfo()}
	unfiltered_hosts_dict = self.host_manager.get_all_host_states(
			elevated, topic)

	# Note: remember, we are using an iterator here. So only
	# traverse this list once. This can bite you if the hosts
	# are being scanned in a filter or weighing function.
	hosts = unfiltered_hosts_dict.itervalues()

	num_instances = request_spec.get('num_instances', 1)
	selected_hosts = []

	for num in xrange(num_instances):
		# Filter local hosts based on requirements ...
		hosts = self.host_manager.filter_hosts(hosts,
				filter_properties)
		if not hosts:
			# Can't get any more locally.
			break

		LOG.debug(_("Filtered %(hosts)s") % locals())

		# weighted_host = WeightedHost() ... the best
		# host for the job.
		# TODO(comstud): filter_properties will also be used for
		# weighing and I plan fold weighing into the host manager
		# in a future patch.  I'll address the naming of this
		# variable at that time.
		weighted_host = least_cost.weighted_sum(cost_functions,
				hosts, filter_properties)
		LOG.debug(_("Weighted %(weighted_host)s") % locals())
		selected_hosts.append(weighted_host)

		# Now consume the resources so the filter/weights
		# will change for the next instance.
		weighted_host.host_state.consume_from_instance(
				instance_properties)

	selected_hosts.sort(key=operator.attrgetter('weight'))
	return selected_hosts[:num_instances]

def notify(message):
    """Look for specific compute manager events and interprete them
    so as to keep the Capacity table up to date.

    NOTE: the True/False return codes are only for testing.
    """

	db.api.compute_node_utilization_update(context.get_admin_context(), host,
        free_ram_mb_delta=free_ram_mb, free_disk_gb_delta=free_disk_gb,
        work_delta=work, vm_delta=vms)

		cfg.IntOpt('host_state_interval',
	               default=120,
	               help='Interval in seconds for querying the host status'),

class SchedulerDependentManager(Manager):
    """Periodically send capability updates to the Scheduler services.

    Services that need to update the Scheduler of their capabilities
    should derive from this class. Otherwise they can derive from
    manager.Manager directly. Updates are only sent after
    update_service_capabilities is called with non-None values.

    """

    def __init__(self, host=None, db_driver=None, service_name='undefined'):
        self.last_capabilities = None
        self.service_name = service_name
        super(SchedulerDependentManager, self).__init__(host, db_driver)

    def update_service_capabilities(self, capabilities):
        """Remember these capabilities to send on next periodic update."""
        self.last_capabilities = capabilities

    @periodic_task
    def _publish_service_capabilities(self, context):
        """Pass data back to the scheduler at a periodic interval."""
        if self.last_capabilities:
            LOG.debug(_('Notifying Schedulers of capabilities ...'))
            api.update_service_capabilities(context, self.service_name,
                                self.host, self.last_capabilities)

def update_service_capabilities(context, service_name, host, capabilities):
    """Send an update to all the scheduler services informing them
       of the capabilities of this service."""
    kwargs = dict(method='update_service_capabilities',
                  args=dict(service_name=service_name, host=host,
                            capabilities=capabilities))
    return rpc.fanout_cast(context, 'scheduler', kwargs)

def update_service_capabilities(self, context, service_name=None,
		host=None, capabilities=None, **kwargs):
	"""Process a capability update from a service node."""
	if capabilities is None:
		capabilities = {}
	self.driver.update_service_capabilities(service_name, host,
			capabilities)

def update_service_capabilities(self, service_name, host, capabilities):
	"""Update the per-service capabilities based on this notification."""
	LOG.debug(_("Received %(service_name)s service update from "
				"%(host)s.") % locals())
	service_caps = self.service_states.get(host, {})
	# Copy the capabilities, so we don't modify the original dict
	capab_copy = dict(capabilities)
	capab_copy["timestamp"] = utils.utcnow()  # Reported time
	service_caps[service_name] = capab_copy
	self.service_states[host] = service_caps