RETIRED, Tricircle is to provide networking automation across Neutron.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

258 lines
8.2 KiB

# Copyright 2015 Huawei Technologies Co., Ltd.
#
# 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.
import os
import random
import sys
from oslo_config import cfg
from oslo_log import log as logging
import oslo_messaging as messaging
from oslo_service import service as srv
from tricircle.common.i18n import _, _LE, _LI
from tricircle.common import baserpc
from tricircle.common import context
from tricircle.common import rpc
from tricircle.common import version
from tricircle.common.serializer import TricircleSerializer as Serializer
from tricircle.common import topics
from tricircle.xjob.xmanager import XManager
_TIMER_INTERVAL = 30
_TIMER_INTERVAL_MAX = 60
common_opts = [
cfg.StrOpt('host', default='tricircle.xhost',
help=_("The host name for RPC server")),
cfg.IntOpt('workers', default=1,
help=_("Number of workers")),
cfg.IntOpt('worker_handle_timeout', default=1800,
help=_("Timeout for worker's one turn of processing, in"
" seconds")),
cfg.IntOpt('job_run_expire', default=180,
help=_("Running job is considered expires after this time, in"
" seconds")),
cfg.FloatOpt('worker_sleep_time', default=0.1,
help=_("Seconds a worker sleeps after one run in a loop")),
cfg.BoolOpt('enable_api_gateway',
default=False,
help=_('Whether the Nova API gateway is enabled'))
]
service_opts = [
cfg.IntOpt('report_interval',
default=10,
help='Seconds between nodes reporting state to datastore'),
cfg.BoolOpt('periodic_enable',
default=True,
help='Enable periodic tasks'),
cfg.IntOpt('periodic_fuzzy_delay',
default=60,
help='Range of seconds to randomly delay when starting the'
' periodic task scheduler to reduce stampeding.'
' (Disable by setting to 0)'),
]
CONF = cfg.CONF
CONF.register_opts(service_opts)
LOG = logging.getLogger(__name__)
class XService(srv.Service):
"""class Service
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
its state to the database services table.
"""
def __init__(self, host, binary, topic, manager, report_interval=None,
periodic_enable=None, periodic_fuzzy_delay=None,
periodic_interval_max=None, serializer=None,
*args, **kwargs):
super(XService, self).__init__()
self.host = host
self.binary = binary
self.topic = topic
self.manager = manager
self.rpc_server = None
self.report_interval = report_interval
self.periodic_enable = periodic_enable
self.periodic_fuzzy_delay = periodic_fuzzy_delay
self.interval_max = periodic_interval_max
self.serializer = serializer
self.saved_args, self.saved_kwargs = args, kwargs
def start(self):
ver_str = version.version_info
LOG.info(_LI('Starting %(topic)s node (version %(version)s)'),
{'topic': self.topic, 'version': ver_str})
self.basic_config_check()
self.manager.init_host()
self.manager.pre_start_hook()
LOG.debug(_("Creating RPC server for service %s"), self.topic)
target = messaging.Target(topic=self.topic, server=self.host)
endpoints = [
self.manager,
baserpc.BaseServerRPCAPI(self.manager.service_name)
]
endpoints.extend(self.manager.additional_endpoints)
self.rpc_server = rpc.get_server(target, endpoints, self.serializer)
self.rpc_server.start()
self.manager.post_start_hook()
if self.periodic_enable:
if self.periodic_fuzzy_delay:
initial_delay = random.randint(0, self.periodic_fuzzy_delay)
else:
initial_delay = None
self.tg.add_dynamic_timer(self.periodic_tasks,
initial_delay=initial_delay,
periodic_interval_max=self.interval_max)
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,
report_interval=None, periodic_enable=None,
periodic_fuzzy_delay=None, periodic_interval_max=None,
serializer=None,):
"""Instantiates class and passes back application object.
:param host: defaults to CONF.host
:param binary: defaults to basename of executable
:param topic: defaults to bin_name - 'nova-' part
:param manager: defaults to CONF.<topic>_manager
:param report_interval: defaults to CONF.report_interval
:param periodic_enable: defaults to CONF.periodic_enable
:param periodic_fuzzy_delay: defaults to CONF.periodic_fuzzy_delay
:param periodic_interval_max: if set, the max time to wait between runs
"""
if not host:
host = CONF.host
if not binary:
binary = os.path.basename(sys.argv[0])
if not topic:
topic = binary.rpartition('tricircle-')[2]
if not manager:
manager_cls = ('%s_manager' %
binary.rpartition('tricircle-')[2])
manager = CONF.get(manager_cls, None)
if report_interval is None:
report_interval = CONF.report_interval
if periodic_enable is None:
periodic_enable = CONF.periodic_enable
if periodic_fuzzy_delay is None:
periodic_fuzzy_delay = CONF.periodic_fuzzy_delay
service_obj = cls(host, binary, topic, manager,
report_interval=report_interval,
periodic_enable=periodic_enable,
periodic_fuzzy_delay=periodic_fuzzy_delay,
periodic_interval_max=periodic_interval_max,
serializer=serializer)
return service_obj
def kill(self):
self.stop()
def stop(self):
try:
self.rpc_server.stop()
except Exception:
pass
try:
self.manager.cleanup_host()
except Exception:
LOG.exception(_LE('Service error occurred during cleanup_host'))
pass
super(XService, self).stop()
def periodic_tasks(self, raise_on_error=False):
"""Tasks to be run at a periodic interval."""
ctxt = context.get_admin_context()
return self.manager.periodic_tasks(ctxt, raise_on_error=raise_on_error)
def basic_config_check(self):
"""Perform basic config checks before starting processing."""
# Make sure the tempdir exists and is writable
# try:
# with utils.tempdir():
# pass
# except Exception as e:
# LOG.error(_LE('Temporary directory is invalid: %s'), e)
# sys.exit(1)
def create_service():
LOG.debug(_('create xjob server'))
xmanager = XManager()
xservice = XService(
host=CONF.host,
binary="xjob",
topic=topics.TOPIC_XJOB,
manager=xmanager,
periodic_enable=True,
report_interval=_TIMER_INTERVAL,
periodic_interval_max=_TIMER_INTERVAL_MAX,
serializer=Serializer()
)
xservice.start()
return xservice
_launcher = None
def serve(xservice, workers=1):
global _launcher
if _launcher:
raise RuntimeError(_('serve() can only be called once'))
_launcher = srv.launch(CONF, xservice, workers=workers)
def wait():
_launcher.wait()