
oslo.service provides a wsgi functionality for defining new long-running services using by OpenStack applications. It might usefull for working with SSL for example. Ironic has started migrate to oslo.service, this change continue this work, and consumes wsgi from it. Change-Id: Ic7865709cd87c45e6b7d49f15ce73354aa15401e Closes-Bug: #1484044
199 lines
6.8 KiB
Python
199 lines
6.8 KiB
Python
# -*- encoding: utf-8 -*-
|
|
#
|
|
# Copyright © 2012 eNovance <licensing@enovance.com>
|
|
#
|
|
# 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 signal
|
|
import socket
|
|
|
|
from oslo_concurrency import processutils
|
|
from oslo_config import cfg
|
|
from oslo_context import context
|
|
from oslo_log import log
|
|
import oslo_messaging as messaging
|
|
from oslo_service import service
|
|
from oslo_service import wsgi
|
|
from oslo_utils import importutils
|
|
|
|
from ironic.api import app
|
|
from ironic.common import config
|
|
from ironic.common import exception
|
|
from ironic.common.i18n import _
|
|
from ironic.common.i18n import _LE
|
|
from ironic.common.i18n import _LI
|
|
from ironic.common import rpc
|
|
from ironic.objects import base as objects_base
|
|
|
|
|
|
service_opts = [
|
|
cfg.IntOpt('periodic_interval',
|
|
default=60,
|
|
help=_('Seconds between running periodic tasks.')),
|
|
cfg.StrOpt('host',
|
|
default=socket.getfqdn(),
|
|
help=_('Name of this node. This can be an opaque identifier. '
|
|
'It is not necessarily a hostname, FQDN, or IP address. '
|
|
'However, the node name must be valid within '
|
|
'an AMQP key, and if using ZeroMQ, a valid '
|
|
'hostname, FQDN, or IP address.')),
|
|
]
|
|
|
|
CONF = cfg.CONF
|
|
LOG = log.getLogger(__name__)
|
|
|
|
CONF.register_opts(service_opts)
|
|
|
|
|
|
class RPCService(service.Service):
|
|
|
|
def __init__(self, host, manager_module, manager_class):
|
|
super(RPCService, self).__init__()
|
|
self.host = host
|
|
manager_module = importutils.try_import(manager_module)
|
|
manager_class = getattr(manager_module, manager_class)
|
|
self.manager = manager_class(host, manager_module.MANAGER_TOPIC)
|
|
self.topic = self.manager.topic
|
|
self.rpcserver = None
|
|
self.deregister = True
|
|
|
|
def start(self):
|
|
super(RPCService, self).start()
|
|
admin_context = context.RequestContext('admin', 'admin', is_admin=True)
|
|
|
|
target = messaging.Target(topic=self.topic, server=self.host)
|
|
endpoints = [self.manager]
|
|
serializer = objects_base.IronicObjectSerializer()
|
|
self.rpcserver = rpc.get_server(target, endpoints, serializer)
|
|
self.rpcserver.start()
|
|
|
|
self.handle_signal()
|
|
self.manager.init_host()
|
|
self.tg.add_dynamic_timer(
|
|
self.manager.periodic_tasks,
|
|
periodic_interval_max=CONF.periodic_interval,
|
|
context=admin_context)
|
|
|
|
LOG.info(_LI('Created RPC server for service %(service)s on host '
|
|
'%(host)s.'),
|
|
{'service': self.topic, 'host': self.host})
|
|
|
|
def stop(self):
|
|
try:
|
|
self.rpcserver.stop()
|
|
self.rpcserver.wait()
|
|
except Exception as e:
|
|
LOG.exception(_LE('Service error occurred when stopping the '
|
|
'RPC server. Error: %s'), e)
|
|
try:
|
|
self.manager.del_host(deregister=self.deregister)
|
|
except Exception as e:
|
|
LOG.exception(_LE('Service error occurred when cleaning up '
|
|
'the RPC manager. Error: %s'), e)
|
|
|
|
super(RPCService, self).stop(graceful=True)
|
|
LOG.info(_LI('Stopped RPC server for service %(service)s on host '
|
|
'%(host)s.'),
|
|
{'service': self.topic, 'host': self.host})
|
|
|
|
def _handle_signal(self, signo, frame):
|
|
LOG.info(_LI('Got signal SIGUSR1. Not deregistering on next shutdown '
|
|
'of service %(service)s on host %(host)s.'),
|
|
{'service': self.topic, 'host': self.host})
|
|
self.deregister = False
|
|
|
|
def handle_signal(self):
|
|
"""Add a signal handler for SIGUSR1.
|
|
|
|
The handler ensures that the manager is not deregistered when it is
|
|
shutdown.
|
|
"""
|
|
signal.signal(signal.SIGUSR1, self._handle_signal)
|
|
|
|
|
|
def prepare_service(argv=[]):
|
|
log.register_options(CONF)
|
|
log.set_defaults(default_log_levels=['amqp=WARNING',
|
|
'amqplib=WARNING',
|
|
'qpid.messaging=INFO',
|
|
'oslo_messaging=INFO',
|
|
'sqlalchemy=WARNING',
|
|
'keystoneclient=INFO',
|
|
'stevedore=INFO',
|
|
'eventlet.wsgi.server=WARNING',
|
|
'iso8601=WARNING',
|
|
'paramiko=WARNING',
|
|
'requests=WARNING',
|
|
'neutronclient=WARNING',
|
|
'glanceclient=WARNING',
|
|
'ironic.openstack.common=WARNING',
|
|
'urllib3.connectionpool=WARNING',
|
|
])
|
|
config.parse_args(argv)
|
|
log.setup(CONF, 'ironic')
|
|
|
|
|
|
def process_launcher():
|
|
return service.ProcessLauncher(CONF)
|
|
|
|
|
|
class WSGIService(service.ServiceBase):
|
|
"""Provides ability to launch ironic API from wsgi app."""
|
|
|
|
def __init__(self, name):
|
|
"""Initialize, but do not start the WSGI server.
|
|
|
|
:param name: The name of the WSGI server given to the loader.
|
|
:returns: None
|
|
"""
|
|
self.name = name
|
|
self.app = app.VersionSelectorApplication()
|
|
self.workers = (CONF.api.api_workers or
|
|
processutils.get_worker_count())
|
|
if self.workers and self.workers < 1:
|
|
raise exception.ConfigInvalid(
|
|
_("api_workers value of %d is invalid, "
|
|
"must be greater than 0.") % self.workers)
|
|
|
|
self.server = wsgi.Server(CONF, name, self.app,
|
|
host=CONF.api.host_ip,
|
|
port=CONF.api.port)
|
|
|
|
def start(self):
|
|
"""Start serving this service using loaded configuration.
|
|
|
|
:returns: None
|
|
"""
|
|
self.server.start()
|
|
|
|
def stop(self):
|
|
"""Stop serving this API.
|
|
|
|
:returns: None
|
|
"""
|
|
self.server.stop()
|
|
|
|
def wait(self):
|
|
"""Wait for the service to stop serving this API.
|
|
|
|
:returns: None
|
|
"""
|
|
self.server.wait()
|
|
|
|
def reset(self):
|
|
"""Reset server greenpool size to default.
|
|
|
|
:returns: None
|
|
"""
|
|
self.server.reset()
|