Change-Id: Idbc145209c039e9362cf97c9926e050f809ef0fachanges/65/15265/2
Angus Salkeld
11 years ago
parent
9874dec669
commit
8135da2fb4
@ -1,182 +0,0 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2010 United States Government as represented by the
|
||||
# Administrator of the National Aeronautics and Space Administration.
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
"""Base Manager class.
|
||||
|
||||
Managers are responsible for a certain aspect of the system. It is a logical
|
||||
grouping of code relating to a portion of the system. In general other
|
||||
components should be using the manager to make changes to the components that
|
||||
it is responsible for.
|
||||
|
||||
For example, other components that need to deal with volumes in some way,
|
||||
should do so by calling methods on the VolumeManager instead of directly
|
||||
changing fields in the database. This allows us to keep all of the code
|
||||
relating to volumes in the same place.
|
||||
|
||||
We have adopted a basic strategy of Smart managers and dumb data, which means
|
||||
rather than attaching methods to data objects, components should call manager
|
||||
methods that act on the data.
|
||||
|
||||
Methods on managers that can be executed locally should be called directly. If
|
||||
a particular method must execute on a remote host, this should be done via rpc
|
||||
to the service that wraps the manager
|
||||
|
||||
Managers should be responsible for most of the db access, and
|
||||
non-implementation specific data. Anything implementation specific that can't
|
||||
be generalized should be done by the Driver.
|
||||
|
||||
In general, we prefer to have one manager with multiple drivers for different
|
||||
implementations, but sometimes it makes sense to have multiple managers. You
|
||||
can think of it this way: Abstract different overall strategies at the manager
|
||||
level(FlatNetwork vs VlanNetwork), and different implementations at the driver
|
||||
level(LinuxNetDriver vs CiscoNetDriver).
|
||||
|
||||
Managers will often provide methods for initial setup of a host or periodic
|
||||
tasks to a wrapping service.
|
||||
|
||||
This module provides Manager, a base class for managers.
|
||||
|
||||
"""
|
||||
|
||||
from heat import version
|
||||
|
||||
from heat.openstack.common.rpc import dispatcher as rpc_dispatcher
|
||||
from heat.openstack.common import log as logging
|
||||
from heat.openstack.common import cfg
|
||||
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def periodic_task(*args, **kwargs):
|
||||
"""Decorator to indicate that a method is a periodic task.
|
||||
|
||||
This decorator can be used in two ways:
|
||||
|
||||
1. Without arguments '@periodic_task', this will be run on every tick
|
||||
of the periodic scheduler.
|
||||
|
||||
2. With arguments, @periodic_task(ticks_between_runs=N), this will be
|
||||
run on every N ticks of the periodic scheduler.
|
||||
"""
|
||||
def decorator(f):
|
||||
f._periodic_task = True
|
||||
f._ticks_between_runs = kwargs.pop('ticks_between_runs', 0)
|
||||
return f
|
||||
|
||||
# NOTE(sirp): The `if` is necessary to allow the decorator to be used with
|
||||
# and without parens.
|
||||
#
|
||||
# In the 'with-parens' case (with kwargs present), this function needs to
|
||||
# return a decorator function since the interpreter will invoke it like:
|
||||
#
|
||||
# periodic_task(*args, **kwargs)(f)
|
||||
#
|
||||
# In the 'without-parens' case, the original function will be passed
|
||||
# in as the first argument, like:
|
||||
#
|
||||
# periodic_task(f)
|
||||
if kwargs:
|
||||
return decorator
|
||||
else:
|
||||
return decorator(args[0])
|
||||
|
||||
|
||||
class ManagerMeta(type):
|
||||
def __init__(cls, names, bases, dict_):
|
||||
"""Metaclass that allows us to collect decorated periodic tasks."""
|
||||
super(ManagerMeta, cls).__init__(names, bases, dict_)
|
||||
|
||||
# NOTE(sirp): if the attribute is not present then we must be the base
|
||||
# class, so, go ahead an initialize it. If the attribute is present,
|
||||
# then we're a subclass so make a copy of it so we don't step on our
|
||||
# parent's toes.
|
||||
try:
|
||||
cls._periodic_tasks = cls._periodic_tasks[:]
|
||||
except AttributeError:
|
||||
cls._periodic_tasks = []
|
||||
|
||||
try:
|
||||
cls._ticks_to_skip = cls._ticks_to_skip.copy()
|
||||
except AttributeError:
|
||||
cls._ticks_to_skip = {}
|
||||
|
||||
for value in cls.__dict__.values():
|
||||
if getattr(value, '_periodic_task', False):
|
||||
task = value
|
||||
name = task.__name__
|
||||
cls._periodic_tasks.append((name, task))
|
||||
cls._ticks_to_skip[name] = task._ticks_between_runs
|
||||
|
||||
|
||||
class Manager(object):
|
||||
__metaclass__ = ManagerMeta
|
||||
|
||||
def __init__(self, host=None, db_driver=None):
|
||||
if not host:
|
||||
host = cfg.CONF.host
|
||||
self.host = host
|
||||
super(Manager, self).__init__(db_driver)
|
||||
|
||||
def create_rpc_dispatcher(self):
|
||||
'''Get the rpc dispatcher for this manager.
|
||||
|
||||
If a manager would like to set an rpc API version, or support more than
|
||||
one class as the target of rpc messages, override this method.
|
||||
'''
|
||||
return rpc_dispatcher.RpcDispatcher([self])
|
||||
|
||||
def periodic_tasks(self, context, raise_on_error=False):
|
||||
"""Tasks to be run at a periodic interval."""
|
||||
for task_name, task in self._periodic_tasks:
|
||||
full_task_name = '.'.join([self.__class__.__name__, task_name])
|
||||
|
||||
ticks_to_skip = self._ticks_to_skip[task_name]
|
||||
if ticks_to_skip > 0:
|
||||
LOG.debug(_("Skipping %(full_task_name)s, %(ticks_to_skip)s"
|
||||
" ticks left until next run"), locals())
|
||||
self._ticks_to_skip[task_name] -= 1
|
||||
continue
|
||||
|
||||
self._ticks_to_skip[task_name] = task._ticks_between_runs
|
||||
LOG.debug(_("Running periodic task %(full_task_name)s"), locals())
|
||||
|
||||
try:
|
||||
task(self, context)
|
||||
except Exception as e:
|
||||
if raise_on_error:
|
||||
raise
|
||||
LOG.exception(_("Error during %(full_task_name)s: %(e)s"),
|
||||
locals())
|
||||
|
||||
def init_host(self):
|
||||
"""Handle initialization if this is a standalone service.
|
||||
|
||||
Child classes should override this method.
|
||||
|
||||
"""
|
||||
pass
|
||||
|
||||
def service_version(self, context):
|
||||
return version.version_string()
|
||||
|
||||
def service_config(self, context):
|
||||
config = {}
|
||||
for key in cfg.CONF:
|
||||
config[key] = cfg.CONF.get(key, None)
|
||||
return config
|
||||
@ -1,241 +0,0 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2010 United States Government as represented by the
|
||||
# Administrator of the National Aeronautics and Space Administration.
|
||||
# Copyright 2011 Justin Santa Barbara
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
"""Generic Node base class for all workers that run on hosts."""
|
||||
|
||||
import inspect
|
||||
import os
|
||||
|
||||
import eventlet
|
||||
import greenlet
|
||||
|
||||
from heat.openstack.common import rpc
|
||||
from heat.openstack.common import cfg
|
||||
from heat.openstack.common import importutils
|
||||
from heat.openstack.common import log as logging
|
||||
|
||||
from heat.common import utils as heat_utils
|
||||
from heat.common import exception
|
||||
from heat.common import context
|
||||
|
||||
from heat import version
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Launcher(object):
|
||||
"""
|
||||
Launch one or more services and wait for them to complete.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the service launcher.
|
||||
|
||||
:returns: None
|
||||
|
||||
"""
|
||||
self._services = []
|
||||
|
||||
@staticmethod
|
||||
def run_server(server):
|
||||
"""Start and wait for a server to finish.
|
||||
|
||||
:param service: Server to run and wait for.
|
||||
:returns: None
|
||||
|
||||
"""
|
||||
server.start()
|
||||
server.wait()
|
||||
|
||||
def launch_server(self, server):
|
||||
"""Load and start the given server.
|
||||
|
||||
:param server: The server you would like to start.
|
||||
:returns: None
|
||||
|
||||
"""
|
||||
gt = eventlet.spawn(self.run_server, server)
|
||||
self._services.append(gt)
|
||||
|
||||
def stop(self):
|
||||
"""Stop all services which are currently running.
|
||||
|
||||
:returns: None
|
||||
|
||||
"""
|
||||
for service in self._services:
|
||||
service.kill()
|
||||
|
||||
def wait(self):
|
||||
"""Waits until all services have been stopped, and then returns.
|
||||
|
||||
:returns: None
|
||||
|
||||
"""
|
||||
for service in self._services:
|
||||
try:
|
||||
service.wait()
|
||||
except greenlet.GreenletExit:
|
||||
pass
|
||||
|
||||
|
||||
class Service(object):
|
||||
"""Service object for binaries running on hosts.
|
||||
|
||||
A service takes a manager and enables rpc by listening to queues based
|
||||
on topic. It also periodically runs tasks on the manager and reports
|
||||
it state to the database services table."""
|
||||
|
||||
def __init__(self, host, binary, topic, manager,
|
||||
periodic_interval=None, *args, **kwargs):
|
||||
self.host = host
|
||||
self.binary = binary
|
||||
self.topic = topic
|
||||
self.manager_class_name = manager
|
||||
manager_class = importutils.import_class(self.manager_class_name)
|
||||
self.manager = manager_class(host=self.host, *args, **kwargs)
|
||||
self.periodic_interval = periodic_interval
|
||||
super(Service, self).__init__(*args, **kwargs)
|
||||
self.saved_args, self.saved_kwargs = args, kwargs
|
||||
self.timers = []
|
||||
|
||||
def start(self):
|
||||
vcs_string = version.version_string(type='long')
|
||||
LOG.info(_('Starting %(topic)s node (version %(vcs_string)s)'),
|
||||
{'topic': self.topic, 'vcs_string': vcs_string})
|
||||
# TODO do we need this ? -> utils.cleanup_file_locks()
|
||||
self.manager.init_host()
|
||||
self.model_disconnected = False
|
||||
ctxt = context.get_admin_context()
|
||||
# self._create_service_ref(ctxt)
|
||||
|
||||
self.conn = rpc.create_connection(new=True)
|
||||
LOG.debug(_("Creating Consumer connection for Service %s") %
|
||||
self.topic)
|
||||
|
||||
rpc_dispatcher = self.manager.create_rpc_dispatcher()
|
||||
|
||||
# Share this same connection for these Consumers
|
||||
self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=False)
|
||||
|
||||
node_topic = '%s.%s' % (self.topic, self.host)
|
||||
self.conn.create_consumer(node_topic, rpc_dispatcher, fanout=False)
|
||||
|
||||
self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=True)
|
||||
|
||||
# Consume from all consumers in a thread
|
||||
self.conn.consume_in_thread()
|
||||
|
||||
if self.periodic_interval:
|
||||
periodic = heat_utils.LoopingCall(self.periodic_tasks)
|
||||
periodic.start(interval=self.periodic_interval, now=False)
|
||||
self.timers.append(periodic)
|
||||
|
||||
def __getattr__(self, key):
|
||||
manager = self.__dict__.get('manager', None)
|
||||
return getattr(manager, key)
|
||||
|
||||
@classmethod
|
||||
def create(cls, host=None, binary=None, topic=None, manager=None,
|
||||
periodic_interval=None, config=None):
|
||||
"""Instantiates class and passes back application object.
|
||||
|
||||
:param host: defaults to cfg.CONF.host
|
||||
:param binary: defaults to basename of executable
|
||||
:param topic: defaults to bin_name - 'heat-' part
|
||||
:param manager: defaults to cfg.CONF.<topic>_manager
|
||||
:param periodic_interval: defaults to cfg.CONF.periodic_interval
|
||||
|
||||
"""
|
||||
if not host:
|
||||
host = cfg.CONF.host
|
||||
if not binary:
|
||||
binary = os.path.basename(inspect.stack()[-1][1])
|
||||
if not topic:
|
||||
topic = binary.rpartition('heat-')[2]
|
||||
if not manager:
|
||||
manager = cfg.CONF.get('%s_manager' % topic, None)
|
||||
if not periodic_interval:
|
||||
periodic_interval = cfg.CONF.periodic_interval
|
||||
service_obj = cls(host, binary, topic, manager,
|
||||
periodic_interval)
|
||||
|
||||
return service_obj
|
||||
|
||||
def kill(self):
|
||||
self.stop()
|
||||
|
||||
def stop(self):
|
||||
# Try to shut the connection down, but if we get any sort of
|
||||
# errors, go ahead and ignore them.. as we're shutting down anyway
|
||||
try:
|
||||
self.conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
for x in self.timers:
|
||||
try:
|
||||
x.stop()
|
||||
except Exception:
|
||||
pass
|
||||
self.timers = []
|
||||
|
||||
def wait(self):
|
||||
for x in self.timers:
|
||||
try:
|
||||
x.wait()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def periodic_tasks(self, raise_on_error=False):
|
||||
"""Tasks to be run at a periodic interval."""
|
||||
ctxt = context.get_admin_context()
|
||||
self.manager.periodic_tasks(ctxt, raise_on_error=raise_on_error)
|
||||
|
||||
|
||||
# NOTE(vish): the global launcher is to maintain the existing
|
||||
# functionality of calling service.serve +
|
||||
# service.wait
|
||||
_launcher = None
|
||||
|
||||
|
||||
def serve(*servers):
|
||||
global _launcher
|
||||
if not _launcher:
|
||||
_launcher = Launcher()
|
||||
for server in servers:
|
||||
_launcher.launch_server(server)
|
||||
|
||||
|
||||
def wait():
|
||||
LOG.debug(_('Full set of CONF:'))
|
||||
for flag in cfg.CONF:
|
||||
flag_get = cfg.CONF.get(flag, None)
|
||||
# hide flag contents from log if contains a password
|
||||
# should use secret flag when switch over to openstack-common
|
||||
if ("_password" in flag or "_key" in flag or
|
||||
(flag == "sql_connection" and "mysql:" in flag_get)):
|
||||
LOG.debug(_('%(flag)s : FLAG SET ') % locals())
|
||||
else:
|
||||
LOG.debug('%(flag)s : %(flag_get)s' % locals())
|
||||
try:
|
||||
_launcher.wait()
|
||||
rpc.cleanup()
|
||||
except KeyboardInterrupt:
|
||||
rpc.cleanup()
|
||||
_launcher.stop()
|
||||
Loading…
Reference in new issue