+++ /dev/null
-#!/usr/bin/python
-#
-# Copyright (c) 2019 Wind River Systems, Inc.
-#
-# SPDX-License-Identifier: Apache-2.0
-#
-
-
-### BEGIN INIT INFO
-# Provides: ceph/mgr RESTful API plugin
-# Required-Start: $ceph
-# Required-Stop: $ceph
-# Default-Start: 2 3 4 5
-# Default-Stop: 0 1 6
-# Short-Description: Ceph MGR RESTful API plugin
-# Description: Ceph MGR RESTful API plugin
-### END INIT INFO
-
-import argparse
-import contextlib
-import errno
-import fcntl
-import inspect
-import json
-import logging
-import multiprocessing
-import os
-import shutil
-import signal
-import socket
-import subprocess
-import sys
-import tempfile
-import time
-
-import daemon
-import psutil
-import requests
-
-# 'timeout' command returns exit status 124
-# if command times out (see man page)
-GNU_TIMEOUT_EXPIRED_RETCODE = 124
-
-
-def psutil_terminate_kill(target, timeout):
-
- """Extend psutil functionality to stop a process.
-
- SIGINT is sent to each target then after a grace period SIGKILL
- is sent to the ones that are still running.
- """
-
- if not isinstance(target, list):
- target = [target]
- _, target = psutil.wait_procs(target, timeout=0)
- for action in [lambda p: p.terminate(), lambda p: p.kill()]:
- for proc in target:
- action(proc)
- _, target = psutil.wait_procs(
- target, timeout=timeout)
-
-
-class Config(object):
-
- """ceph-mgr service wrapper configuration options.
-
- In the future we may want to load them from a configuration file
- (for example /etc/ceph/mgr-restful-plugin.conf )
- """
-
- def __init__(self):
- self.log_level = logging.INFO
- self.log_dir = '/var/log'
-
- self.ceph_mgr_service = '/usr/bin/ceph-mgr'
- self.ceph_mgr_config = '/etc/ceph/ceph.conf'
- self.ceph_mgr_cluster = 'ceph'
- self.ceph_mgr_rundir = '/var/run/ceph/mgr'
- self.ceph_mgr_confdir = '/var/lib/ceph/mgr'
- self.ceph_mgr_identity = socket.gethostname()
-
- self.service_name = 'mgr-restful-plugin'
- self.service_socket = os.path.join(
- self.ceph_mgr_rundir, '{}.socket'.format(self.service_name))
- self.service_lock = os.path.join(
- self.ceph_mgr_rundir, '{}.lock'.format(self.service_name))
- self.service_pid_file = os.path.join(
- '/var/run/ceph', '{}.pid'.format(self.service_name))
-
- self.restful_plugin_port = 5001
-
- # maximum size of a message received/sent via
- # service monitor control socket
- self.service_socket_bufsize = 1024
-
- # maximum time to wait for ceph cli to exit
- self.ceph_cli_timeout_sec = 30
-
- # how much time to wait after ceph cli commands fail with timeout
- # before running any other commands
- self.cluster_grace_period_sec = 30
-
- # after ceph-mgr is started it goes through an internal initialization
- # phase before; how much time to wait before querying ceph-mgr
- self.ceph_mgr_grace_period_sec = 15
-
- # after sending SIGTERM to ceph-mgr how much time to wait before
- # sending SIGKILL (maximum time allowed for ceph-mgr cleanup)
- self.ceph_mgr_kill_delay_sec = 5
-
- # if service monitor is running a recovery procedure it reports
- # status OK even if ceph-mgr is currently down. This sets the
- # maximum number of consecutive ceph-mgr failures before reporting
- # status error
- self.ceph_mgr_fail_count_report_error = 3
-
- # maximum number of consecutive ceph-mgr failures before
- # stopping mgr-restful-plugin service
- self.ceph_mgr_fail_count_exit = 5
-
- # maximum time allowed for ceph-mgr to respond to a REST API request
- self.rest_api_timeout_sec = 15
-
- # interval between consecutive REST API requests (ping's). A smaller
- # value here triggers more requests to ceph-mgr restful plugin. A
- # higher value makes recovery slower when services become unavailable
- self.restful_plugin_ping_delay_sec = 3
-
- # where to save the self-signed certificate generated by ceph-mgr
- self.restful_plugin_cert_path = os.path.join(
- self.ceph_mgr_rundir, 'restful.crt')
-
- # time to wait after enabling restful plugin
- self.restful_plugin_grace_period_sec = 3
-
- # after how many REST API ping failures to restart ceph-mgr
- self.ping_fail_count_restart_mgr = 3
-
- # after how many REST API ping failures to report status error.
- # Until then service monitor reports status OK just in case
- # restful plugin recovers
- self.ping_fail_count_report_error = 5
-
- @staticmethod
- def load():
- return Config()
-
-
-def setup_logging(name=None, cleanup_handlers=False):
- if not name:
- name = CONFIG.service_name
- log = logging.getLogger(name)
- log.setLevel(CONFIG.log_level)
- if cleanup_handlers:
- try:
- for handler in log.handlers:
- if isinstance(handler, logging.StreamHandler):
- handler.flush()
- if isinstance(handler, logging.FileHandler):
- handler.close()
- log.handlers = []
- except Exception:
- pass
- elif log.handlers:
- return log
- handler = logging.FileHandler(
- os.path.join(CONFIG.log_dir,
- '{}.log'.format(CONFIG.service_name)))
- handler.setFormatter(
- logging.Formatter('%(asctime)s %(process)s %(levelname)s %(name)s %(message)s'))
- log.addHandler(handler)
- return log
-
-
-CONFIG = Config.load()
-LOG = setup_logging(name='init-wrapper')
-
-
-class ServiceException(Exception):
-
- """Generic mgr-restful-plugin service exception.
-
- Build exception string based on static (per exception class)
- string plus args, keyword args passed to exception constructor.
- """
-
- message = ""
-
- def __init__(self, *args, **kwargs):
- if "message" not in kwargs:
- try:
- message = self.message.format(*args, **kwargs)
- except Exception: # noqa
- message = '{}, args:{}, kwargs: {}'.format(
- self.message, args, kwargs)
- else:
- message = kwargs["message"]
- super(ServiceException, self).__init__(message)
-
-
-class ServiceAlreadyStarted(ServiceException):
- message = ('Service monitor already started')
-
-
-class ServiceLockFailed(ServiceException):
- message = ('Unable to lock service monitor: '
- 'reason={reason}')
-
-
-class ServiceNoSocket(ServiceException):
- message = ('Unable to create service monitor socket: '
- 'reason={reason}')
-
-
-class ServiceSocketBindFailed(ServiceException):
- message = ('Failed to bind service monitor socket: '
- 'path={path}, reason={reason}')
-
-
-class ServiceNoPidFile(ServiceException):
- message = ('Failed to update pid file: '
- 'path={path}, reason={reason}')
-
-
-class CommandFailed(ServiceException):
- message = ('Command failed: command={command}, '
- 'reason={reason}, out={out}')
-
-
-class CommandTimeout(ServiceException):
- message = ('Command timeout: command={command}, '
- 'timeout={timeout}')
-
-
-class CephMgrStartFailed(ServiceException):
- message = ('Failed to start ceph_mgr: '
- 'reason={reason}')
-
-
-class CephRestfulPluginFailed(ServiceException):
- message = ('Failed to start restful plugin: '
- 'reason={reason}')
-
-
-class RestApiPingFailed(ServiceException):
- message = ('REST API ping failed: '
- 'reason={reason}')
-
-
-class ServiceMonitor(object):
-
- """Configure and monitor ceph-mgr and restful plugin (Ceph REST API)
-
- 1. process init script service requests: status, stop. Requests are
- received via a control socket. Stop has priority over whatever
- the monitor is doing currently. Any ceph command that may be running
- is terminated/killed. Note that while ceph-mgr and restful plugin
- configuration is in progress ServiceMonitor reports status OK to
- avoid being restarted by SM.
-
- 2. configure ceph-mgr and mgr restful plugin: authentication, REST API
- service port, self signed certificate. This runs as a separate
- process so it can be stopped when init script requests it.
-
- 3. periodically check (ping) REST API responds to HTTPS requests.
- Recovery actions are taken if REST API fails to respond: restart
- ceph-mgr, wait for cluster to become available again.
- """
-
- def __init__(self):
- # process running configuration & REST API ping loop
- self.monitor = None
-
- # command socket used by init script
- self.command = None
-
- # ceph-mgr process
- self.ceph_mgr = None
-
- # consecutive ceph-mgr/restful-plugin start failures. Service monitor
- # reports failure after CONFIG.ceph_mgr_max_failure_count
- self.ceph_mgr_failure_count = 0
-
- # consecutive REST API ping failures. ceph-mgr service is restarted
- # after CONFIG.ping_fail_count_restart_mgr threshold is exceeded
- self.ping_failure_count = 0
-
- # REST API url reported by ceph-mgr after enabling restful plugin
- self.restful_plugin_url = ''
-
- # REST API self signed certificate generated by restful plugin
- self.certificate = ''
-
- def run(self):
- self.disable_certificate_check()
- with self.service_lock(), self.service_socket(), \
- self.service_pid_file():
- self.start_monitor()
- self.server_loop()
-
- def disable_certificate_check(self):
- # ceph-mgr restful plugin is configured with a self-signed
- # certificate. Certificate host is hard-coded to "ceph-restful"
- # which causes HTTPS requests to fail because they don't
- # match current host name ("controller-..."). Disable HTTPS
- # certificates check in urllib3
- LOG.warning('Disable urllib3 certifcates check')
- requests.packages.urllib3.disable_warnings()
-
- def server_loop(self):
- self.command.listen(2)
- while True:
- try:
- client, _ = self.command.accept()
- request = client.recv(CONFIG.service_socket_bufsize)
- LOG.debug('Monitor command socket: request=%s', str(request))
- cmd = request.split(' ')
- cmd, args = cmd[0], cmd[1:]
- if cmd == 'status':
- self.send_response(client, request, self.status())
- elif cmd == 'stop':
- self.stop()
- self.send_response(client, request, 'OK')
- break
- elif cmd == 'restful-url':
- try:
- self.restful_plugin_url = args[0]
- self.send_response(client, request, 'OK')
- except IndexError:
- LOG.warning('Failed to update restful plugin url: '
- 'args=%s', str(args))
- self.send_response(client, request, 'ERR')
- elif cmd == 'certificate':
- try:
- self.certificate = args[0] if args else ''
- self.send_response(client, request, 'OK')
- except IndexError:
- LOG.warning('Failed to update certificate path: '
- 'args=%s', str(args))
- self.send_response(client, request, 'ERR')
- elif cmd == 'ceph-mgr-failures':
- try:
- self.ceph_mgr_failure_count = int(args[0])
- self.send_response(client, request, 'OK')
- if self.ceph_mgr_failure_count >= CONFIG.ceph_mgr_fail_count_exit:
- self.stop()
- break
- except (IndexError, ValueError):
- LOG.warning('Failed to update ceph-mgr failures: '
- 'args=%s', str(args))
- self.send_response(client, request, 'ERR')
- elif cmd == 'ping-failures':
- try:
- self.ping_failure_count = int(args[0])
- self.send_response(client, request, 'OK')
- except (IndexError, ValueError):
- LOG.warning('Failed to update ping failures: '
- 'args=%s', str(args))
- self.send_response(client, request, 'ERR')
- except Exception as err:
- LOG.exception(err)
-
- @staticmethod
- def send_response(client, request, response):
- try:
- client.send(response)
- except socket.error as err:
- LOG.warning('Failed to send response back. '
- 'request=%s, response=%s, reason=%s',
- request, response, err)
-
- def status(self):
- if not self.restful_plugin_url:
- if self.ceph_mgr_failure_count < CONFIG.ceph_mgr_fail_count_report_error \
- and self.ping_failure_count < CONFIG.ping_fail_count_report_error:
- LOG.debug('Monitor is starting services. Report status OK')
- return 'OK'
- LOG.debug('Too many failures: '
- 'ceph_mgr=%d < %d, ping=%d < %d. '
- 'Report status ERR',
- self.ceph_mgr_failure_count,
- CONFIG.ceph_mgr_fail_count_report_error,
- self.ping_failure_count,
- CONFIG.ping_fail_count_report_error)
- return 'ERR.down'
- try:
- self.restful_plugin_ping()
- LOG.debug('Restful plugin ping successful. Report status OK')
- return 'OK'
- except (CommandFailed, RestApiPingFailed):
- if self.ceph_mgr_failure_count < CONFIG.ceph_mgr_fail_count_report_error \
- and self.ping_failure_count < CONFIG.ping_fail_count_report_error:
- LOG.info('Restful plugin does not respond but failure '
- 'count is within acceptable limits: '
- ' ceph_mgr=%d < %d, ping=%d < %d. '
- 'Report status OK',
- self.ceph_mgr_failure_count,
- CONFIG.ceph_mgr_fail_count_report_error,
- self.ping_failure_count,
- CONFIG.ping_fail_count_report_error)
- return 'OK'
- LOG.debug('Restful does not respond (ping failure count %d). '
- 'Report status ERR', self.ping_failure_count)
- return 'ERR.ping_failed'
-
- def stop(self):
- if not self.monitor:
- return
- LOG.info('Stop monitor with SIGTERM to process group %d',
- self.monitor.pid)
- try:
- os.killpg(self.monitor.pid, signal.SIGTERM)
- except OSError as err:
- LOG.info('Stop monitor failed: reason=%s', str(err))
- return
- time.sleep(CONFIG.ceph_mgr_kill_delay_sec)
- LOG.info('Stop monitor with SIGKILL to process group %d',
- self.monitor.pid)
- try:
- os.killpg(self.monitor.pid, signal.SIGKILL)
- os.waitpid(self.monitor.pid, 0)
- except OSError as err:
- LOG.info('Stop monitor failed: reason=%s', str(err))
- return
- LOG.info('Monitor stopped: pid=%d', self.monitor.pid)
-
- @contextlib.contextmanager
- def service_lock(self):
- LOG.info('Take service lock: path=%s', CONFIG.service_lock)
- try:
- os.makedirs(os.path.dirname(CONFIG.service_lock))
- except OSError:
- pass
- lock_file = open(CONFIG.service_lock, 'w')
- try:
- fcntl.flock(lock_file.fileno(),
- fcntl.LOCK_EX | fcntl.LOCK_NB)
- except (IOError, OSError) as err:
- if err.errno == errno.EAGAIN:
- raise ServiceAlreadyStarted()
- else:
- raise ServiceLockFailed(reason=str(err))
- # even if we have the lock here there might be another service manager
- # running whose CONFIG.ceph_mgr_rundir was removed before starting
- # this instance. Make sure there is only one service manager running
- self.stop_other_service_managers()
- try:
- yield
- finally:
- os.unlink(CONFIG.service_lock)
- lock_file.close()
- LOG.info('Release service lock: path=%s', CONFIG.service_lock)
-
- def stop_other_service_managers(self):
- service = os.path.join('/etc/init.d', CONFIG.service_name)
- for p in psutil.process_iter():
- if p.cmdline()[:2] not in [[service], ['/usr/bin/python', service]]:
- continue
- if p.pid == os.getpid():
- continue
- p.kill()
-
- @contextlib.contextmanager
- def service_socket(self):
- LOG.info('Create service socket')
- try:
- self.command = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
- except socket.error as err:
- raise ServiceNoSocket(reason=str(err))
- LOG.info('Remove existing socket files')
- try:
- os.unlink(CONFIG.service_socket)
- except OSError:
- pass
- LOG.info('Bind service socket: path=%s', CONFIG.service_socket)
- try:
- self.command.bind(CONFIG.service_socket)
- except socket.error as err:
- raise ServiceSocketBindFailed(
- path=CONFIG.service_socket, reason=str(err))
- try:
- yield
- finally:
- LOG.info('Close service socket and remove file: path=%s',
- CONFIG.service_socket)
- self.command.close()
- os.unlink(CONFIG.service_socket)
-
- @contextlib.contextmanager
- def service_pid_file(self):
- LOG.info('Update service pid file: path=%s', CONFIG.service_pid_file)
- try:
- pid_file = open(CONFIG.service_pid_file, 'w')
- pid_file.write(str(os.getpid()))
- pid_file.flush()
- except OSError as err:
- raise ServiceNoPidFile(
- path=CONFIG.service_pid_file, reason=str(err))
- try:
- yield
- finally:
- LOG.info('Remove service pid file: path=%s',
- CONFIG.service_pid_file)
- try:
- os.unlink(CONFIG.service_pid_file)
- except OSError:
- pass
-
- def start_monitor(self):
- LOG.info('Start monitor loop')
- self.monitor = multiprocessing.Process(target=self.monitor_loop)
- self.monitor.start()
-
- def stop_unmanaged_ceph_mgr(self):
- LOG.info('Stop unmanaged running ceph-mgr processes')
- service_name = os.path.basename(CONFIG.ceph_mgr_service)
- if self.ceph_mgr:
- psutil_terminate_kill(
- [proc for proc in psutil.process_iter()
- if (proc.name() == service_name
- and proc.pid != self.ceph_mgr.pid)],
- CONFIG.ceph_mgr_kill_delay_sec)
- else:
- psutil_terminate_kill(
- [proc for proc in psutil.process_iter()
- if proc.name() == service_name],
- CONFIG.ceph_mgr_kill_delay_sec)
-
- def monitor_loop(self):
-
- """Bring up and monitor ceph-mgr restful plugin.
-
- Steps:
- - wait for Ceph cluster to become available
- - configure and start ceph-mgr
- - configure and enable restful plugin
- - send periodic requests to REST API
- - recover from failures
-
- Note: because this runs as a separate process it
- must send status updates to service monitor
- via control socket for: ping_failure_count,
- restful_plugin_url and certificate.
- """
-
- # Promote to process group leader so parent (service monitor)
- # can kill the monitor plus processes spawned by it. Otherwise
- # children of monitor_loop() will keep running in background and
- # will be reaped by init when they finish but by then they might
- # interfere with any new service instance.
- os.setpgrp()
-
- # Ignoring SIGTERM here ensures process group is not reused by
- # the time parent (service monitor) issues the final SIGKILL.
- signal.signal(signal.SIGTERM, signal.SIG_IGN)
-
- while True:
- try:
- # steps to configure/start ceph-mgr and restful plugin
- self.ceph_fsid_get()
- self.ceph_mgr_auth_create()
- self.restful_plugin_set_server_port()
- self.restful_plugin_create_certificate()
- self.ceph_mgr_start()
- self.restful_plugin_enable()
- self.restful_plugin_create_admin_key()
- self.restful_plugin_get_url()
- self.restful_plugin_get_certificate()
-
- # REST API should be available now
- # start making periodic requests (ping)
- while True:
- try:
- self.restful_plugin_ping()
- self.ping_failure_count = 0
- self.request_update_ping_failures(
- self.ping_failure_count)
- self.ceph_mgr_failure_count = 0
- self.request_update_ceph_mgr_failures(
- self.ceph_mgr_failure_count)
- time.sleep(CONFIG.restful_plugin_ping_delay_sec)
- continue
- except RestApiPingFailed as err:
- LOG.warning(str(err))
-
- LOG.info('REST API ping failure count=%d',
- self.ping_failure_count)
- self.ping_failure_count += 1
- self.request_update_ping_failures(
- self.ping_failure_count)
-
- # maybe request failed because ceph-mgr is not running
- if not self.ceph_mgr_is_running():
- self.ceph_mgr_failure_count += 1
- self.request_update_ceph_mgr_failures(
- self.ceph_mgr_failure_count)
- self.ceph_mgr_start()
- time.sleep(CONFIG.ceph_mgr_grace_period_sec)
- continue
-
- # maybe request failed because cluster health is not ok
- if not self.ceph_fsid_get():
- LOG.info('Unable to get cluster fsid. '
- 'Sleep for a while')
- time.sleep(CONFIG.cluster_grace_period_sec)
- break
-
- # too many failures? Restart ceph-mgr and go again
- # through configuration steps
- if (self.ping_failure_count
- % CONFIG.ping_fail_count_restart_mgr == 0):
- LOG.info('Too many consecutive REST API failures. '
- 'Restart ceph-mgr. Update service '
- 'url and certificate')
- self.ceph_mgr_stop()
- self.restful_plugin_url = ''
- self.request_update_plugin_url(self.restful_plugin_url)
- self.certificate = ''
- self.request_update_certificate(self.certificate)
- break
-
- time.sleep(CONFIG.restful_plugin_ping_delay_sec)
-
- except CommandFailed as err:
- LOG.warning(str(err))
- time.sleep(CONFIG.cluster_grace_period_sec)
- except CommandTimeout as err:
- LOG.warning(str(err))
- except (CephMgrStartFailed, CephRestfulPluginFailed) as err:
- LOG.warning(str(err))
- self.ceph_mgr_failure_count += 1
- self.request_update_ceph_mgr_failures(
- self.ceph_mgr_failure_count)
- time.sleep(CONFIG.ceph_mgr_grace_period_sec)
- except Exception as err:
- LOG.exception(err)
- time.sleep(CONFIG.cluster_grace_period_sec)
-
- @staticmethod
- def run_with_timeout(command, timeout, stderr=subprocess.STDOUT):
- try:
- LOG.info('Run command: %s', ' '.join(command))
- return subprocess.check_output(
- ['/usr/bin/timeout', str(timeout)] + command,
- stderr=stderr, shell=False).strip()
- except subprocess.CalledProcessError as err:
- if err.returncode == GNU_TIMEOUT_EXPIRED_RETCODE:
- raise CommandTimeout(command=err.cmd, timeout=timeout)
- raise CommandFailed(command=err.cmd, reason=str(err),
- out=err.output)
-
- def ceph_fsid_get(self):
- return self.run_with_timeout(['/usr/bin/ceph', 'fsid'],
- CONFIG.ceph_cli_timeout_sec)
-
- def ceph_mgr_has_auth(self):
- path = '{}/ceph-{}'.format(
- CONFIG.ceph_mgr_confdir, CONFIG.ceph_mgr_identity)
- try:
- os.makedirs(path)
- except OSError as err:
- pass
- try:
- self.run_with_timeout(
- ['/usr/bin/ceph', 'auth', 'get',
- 'mgr.{}'.format(CONFIG.ceph_mgr_identity),
- '-o', '{}/keyring'.format(path)],
- CONFIG.ceph_cli_timeout_sec)
- return True
- except CommandFailed as err:
- if 'ENOENT' in str(err):
- return False
- raise
-
- def ceph_mgr_auth_create(self):
- if self.ceph_mgr_has_auth():
- return
- LOG.info('Create ceph-mgr authentication')
- self.run_with_timeout(
- ['/usr/bin/ceph', 'auth', 'get-or-create',
- 'mgr.{}'.format(CONFIG.ceph_mgr_identity),
- 'mon', 'allow *', 'osd', 'allow *'],
- CONFIG.ceph_cli_timeout_sec)
-
- def ceph_mgr_is_running(self):
- if not self.ceph_mgr:
- return None
- try:
- self.ceph_mgr.wait(timeout=0)
- except psutil.TimeoutExpired:
- return True
- return False
-
- def ceph_mgr_start(self):
- if self.ceph_mgr_is_running():
- return
- self.stop_unmanaged_ceph_mgr()
- LOG.info('Start ceph-mgr daemon')
- try:
- with open(os.devnull, 'wb') as null:
- self.ceph_mgr = psutil.Popen(
- [CONFIG.ceph_mgr_service,
- '--cluster', CONFIG.ceph_mgr_cluster,
- '--conf', CONFIG.ceph_mgr_config,
- '--id', CONFIG.ceph_mgr_identity,
- '-f'],
- close_fds=True,
- stdout=null,
- stderr=null,
- shell=False)
- except (OSError, ValueError) as err:
- raise CephMgrStartFailed(reason=str(err))
- time.sleep(CONFIG.ceph_mgr_grace_period_sec)
-
- def ceph_mgr_stop(self):
- if not self.ceph_mgr:
- return
- LOG.info('Stop ceph-mgr')
- psutil_terminate_kill(self.ceph_mgr, CONFIG.ceph_mgr_kill_delay_sec)
-
- def restful_plugin_has_server_port(self):
- try:
- with open(os.devnull, 'wb') as null:
- out = self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'get',
- 'mgr/restful/server_port'],
- CONFIG.ceph_cli_timeout_sec, stderr=null)
- if out == str(CONFIG.restful_plugin_port):
- return True
- LOG.warning('Restful plugin port mismatch: '
- 'current=%d, expected=%d', out,
- CONFIG.restful_plugin_port)
- except CommandFailed as err:
- LOG.warning('Failed to get restful plugin port: '
- 'reason=%s', str(err))
- return False
-
- def restful_plugin_set_server_port(self):
- if self.restful_plugin_has_server_port():
- return
- LOG.info('Set restful plugin port=%d', CONFIG.restful_plugin_port)
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'set',
- 'mgr/restful/server_port', str(CONFIG.restful_plugin_port)],
- CONFIG.ceph_cli_timeout_sec)
-
- def restful_plugin_has_admin_key(self):
- try:
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'get',
- 'mgr/restful/keys/admin'],
- CONFIG.ceph_cli_timeout_sec)
- return True
- except CommandFailed:
- pass
- return False
-
- def restful_plugin_create_admin_key(self):
- if self.restful_plugin_has_admin_key():
- return
- LOG.info('Create restful plugin admin key')
- self.run_with_timeout(
- ['/usr/bin/ceph', 'restful',
- 'create-key', 'admin'],
- CONFIG.ceph_cli_timeout_sec)
-
- def restful_plugin_has_certificate(self):
- try:
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'get',
- 'config/mgr/restful/{}/crt'.format(CONFIG.ceph_mgr_identity)],
- CONFIG.ceph_cli_timeout_sec)
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'get',
- 'mgr/restful/{}/crt'.format(CONFIG.ceph_mgr_identity)],
- CONFIG.ceph_cli_timeout_sec)
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'get',
- 'config/mgr/restful/{}/key'.format(CONFIG.ceph_mgr_identity)],
- CONFIG.ceph_cli_timeout_sec)
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'get',
- '/mgr/restful/{}/key'.format(CONFIG.ceph_mgr_identity)],
- CONFIG.ceph_cli_timeout_sec)
- return True
- except CommandFailed:
- pass
- return False
-
- def restful_plugin_create_certificate(self):
- if self.restful_plugin_has_certificate():
- return
- LOG.info('Create restful plugin self signed certificate')
- path = tempfile.mkdtemp()
- try:
- try:
- with tempfile.NamedTemporaryFile() as restful_cnf:
- restful_cnf.write((
- '[req]\n'
- 'req_extensions = v3_ca\n'
- 'distinguished_name = req_distinguished_name\n'
- '[v3_ca]\n'
- 'subjectAltName=DNS:{}\n'
- 'basicConstraints = CA:true\n'
- '[ req_distinguished_name ]\n'
- '0.organizationName = IT\n'
- 'commonName = ceph-restful\n').format(
- CONFIG.ceph_mgr_identity))
- restful_cnf.flush()
- subprocess.check_call([
- '/usr/bin/openssl', 'req', '-new', '-nodes', '-x509',
- '-subj', '/O=IT/CN=' + CONFIG.ceph_mgr_identity,
- '-days', '3650',
- '-config', restful_cnf.name,
- '-out', os.path.join(path, 'crt'),
- '-keyout', os.path.join(path, 'key'),
- '-extensions', 'v3_ca'])
- except subprocess.CalledProcessError as err:
- raise CommandFailed(
- command=' '.join(err.cmd),
- reason='failed to generate self-signed certificate: {}'.format(str(err)),
- out=err.output)
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'set',
- 'config/mgr/restful/{}/crt'.format(CONFIG.ceph_mgr_identity),
- '-i', os.path.join(path, 'crt')],
- CONFIG.ceph_cli_timeout_sec)
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'set',
- 'mgr/restful/{}/crt'.format(CONFIG.ceph_mgr_identity),
- '-i', os.path.join(path, 'crt')],
- CONFIG.ceph_cli_timeout_sec)
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'set',
- 'config/mgr/restful/{}/key'.format(CONFIG.ceph_mgr_identity),
- '-i', os.path.join(path, 'key')],
- CONFIG.ceph_cli_timeout_sec)
- self.run_with_timeout(
- ['/usr/bin/ceph', 'config-key', 'set',
- 'mgr/restful/{}/key'.format(CONFIG.ceph_mgr_identity),
- '-i', os.path.join(path, 'key')],
- CONFIG.ceph_cli_timeout_sec)
- finally:
- shutil.rmtree(path)
-
- def restful_plugin_is_enabled(self):
- command = ['/usr/bin/ceph', 'mgr', 'module', 'ls',
- '--format', 'json']
- with open(os.devnull, 'wb') as null:
- out = self.run_with_timeout(
- command, CONFIG.ceph_cli_timeout_sec, stderr=null)
- try:
- if 'restful' in json.loads(out)['enabled_modules']:
- return True
- except ValueError as err:
- raise CommandFailed(
- command=' '.join(command),
- reason='unable to decode json: {}'.format(err), out=out)
- except KeyError as err:
- raise CommandFailed(
- command=' '.join(command),
- reason='missing expected key: {}'.format(err), out=out)
- return False
-
- def restful_plugin_enable(self):
- if not self.restful_plugin_is_enabled():
- LOG.info('Enable restful plugin')
- self.run_with_timeout(
- ['/usr/bin/ceph', 'mgr',
- 'module', 'enable', 'restful'],
- CONFIG.ceph_cli_timeout_sec)
- time.sleep(CONFIG.restful_plugin_grace_period_sec)
-
- def restful_plugin_get_url(self):
- command = ['/usr/bin/ceph', 'mgr', 'services',
- '--format', 'json']
- with open(os.devnull, 'wb') as null:
- out = self.run_with_timeout(
- command, CONFIG.ceph_cli_timeout_sec, stderr=null)
- try:
- self.restful_plugin_url = json.loads(out)['restful']
- except ValueError as err:
- raise CephRestfulPluginFailed(
- reason='unable to decode json: {} output={}'.format(err, out))
- except KeyError as err:
- raise CephRestfulPluginFailed(
- reason='missing expected key: {} in ouput={}'.format(err, out))
- self.request_update_plugin_url(self.restful_plugin_url)
-
- def restful_plugin_get_certificate(self):
- command = ['/usr/bin/ceph', 'config-key', 'get',
- 'config/mgr/restful/{}/crt'.format(CONFIG.ceph_mgr_identity)]
- with open(os.devnull, 'wb') as null:
- certificate = self.run_with_timeout(
- command, CONFIG.ceph_cli_timeout_sec, stderr=null)
- with open(CONFIG.restful_plugin_cert_path, 'wb') as cert_file:
- cert_file.write(certificate)
- self.certificate = CONFIG.restful_plugin_cert_path
- self.request_update_certificate(
- self.certificate)
-
- def restful_plugin_ping(self):
- if not self.restful_plugin_url:
- raise RestApiPingFailed(reason='missing service url')
- if not self.certificate:
- raise RestApiPingFailed(reason='missing certificate')
- LOG.debug('Ping restful plugin: url=%d', self.restful_plugin_url)
- try:
- response = requests.request(
- 'GET', self.restful_plugin_url, verify=False,
- timeout=CONFIG.rest_api_timeout_sec)
- if not response.ok:
- raise RestApiPingFailed(
- reason='response not ok ({})'.format(response))
- LOG.debug('Ping restful plugin OK')
- except (requests.ConnectionError,
- requests.Timeout,
- requests.HTTPError) as err:
- raise RestApiPingFailed(reason=str(err))
-
- @staticmethod
- def _make_client_socket():
- sock = socket.socket(
- socket.AF_UNIX, socket.SOCK_SEQPACKET)
- sock.settimeout(2 * CONFIG.rest_api_timeout_sec)
- sock.connect(CONFIG.service_socket)
- return sock
-
- @staticmethod
- def request_status():
- try:
- with contextlib.closing(
- ServiceMonitor._make_client_socket()) as sock:
- sock.send('status')
- status = sock.recv(CONFIG.service_socket_bufsize)
- LOG.debug('Status %s', status)
- return status.startswith('OK')
- except socket.error as err:
- LOG.error('Status error: reason=%s', err)
- return False
-
- @staticmethod
- def request_stop():
- try:
- with contextlib.closing(
- ServiceMonitor._make_client_socket()) as sock:
- sock.send('stop')
- response = sock.recv(CONFIG.service_socket_bufsize)
- LOG.debug('Stop response: %s', response)
- return True
- except socket.error as err:
- LOG.error('Stop error: reason=%s', err)
- return False
-
- @staticmethod
- def request_update_ceph_mgr_failures(count):
- try:
- with contextlib.closing(
- ServiceMonitor._make_client_socket()) as sock:
- sock.send('ceph-mgr-failures {}'.format(count))
- sock.recv(CONFIG.service_socket_bufsize)
- return True
- except socket.error as err:
- LOG.error('Stop error: reason=%s', err)
- return False
-
- @staticmethod
- def request_update_ping_failures(count):
- try:
- with contextlib.closing(
- ServiceMonitor._make_client_socket()) as sock:
- sock.send('ping-failures {}'.format(count))
- sock.recv(CONFIG.service_socket_bufsize)
- return True
- except socket.error as err:
- LOG.error('Stop error: reason=%s', err)
- return False
-
- @staticmethod
- def request_update_plugin_url(url):
- try:
- with contextlib.closing(
- ServiceMonitor._make_client_socket()) as sock:
- sock.send('restful-url {}'.format(url))
- sock.recv(CONFIG.service_socket_bufsize)
- return True
- except socket.error as err:
- LOG.error('Stop error: reason=%s', err)
- return False
-
- @staticmethod
- def request_update_certificate(path):
- try:
- with contextlib.closing(
- ServiceMonitor._make_client_socket()) as sock:
- sock.send('certificate {}'.format(path))
- sock.recv(CONFIG.service_socket_bufsize)
- return True
- except socket.error as err:
- LOG.error('Stop error: reason=%s', err)
- return False
-
-
-class InitWrapper(object):
-
- """Handle System V init script actions: start, stop, restart, etc. """
-
- def __init__(self):
-
- """Dispatch command line action to the corresponding function.
-
- Candidate action functions are all class methods except ones
- that start with an underscore.
- """
-
- parser = argparse.ArgumentParser()
- actions = [m[0]
- for m in inspect.getmembers(self)
- if (inspect.ismethod(m[1])
- and not m[0].startswith('_'))]
- parser.add_argument(
- 'action',
- choices=actions)
- self.args = parser.parse_args()
- getattr(self, self.args.action)()
-
- def start(self):
-
- """Start ServiceMonitor as a daemon unless one is already running.
-
- Use a pipe to report monitor status back to this process.
- """
-
- pipe = os.pipe()
- child = os.fork()
- if child == 0:
- os.close(pipe[0])
- with daemon.DaemonContext(files_preserve=[pipe[1]]):
- # prevent duplication of messages in log
- global LOG
- LOG = setup_logging(cleanup_handlers=True)
- try:
- monitor = ServiceMonitor()
- status = 'OK'
- except ServiceAlreadyStarted:
- os.write(pipe[1], 'OK')
- os.close(pipe[1])
- return
- except Exception as err:
- status = str(err)
- os.write(pipe[1], status)
- os.close(pipe[1])
- if status == 'OK':
- try:
- monitor.run()
- except ServiceException as err:
- LOG.warning(str(err))
- except Exception as err:
- LOG.exception('Service monitor error: reason=%s', err)
- else:
- os.close(pipe[1])
- try:
- status = os.read(pipe[0], CONFIG.service_socket_bufsize)
- if status == 'OK':
- sys.exit(0)
- else:
- LOG.warning('Service monitor failed to start: '
- 'status=%s', status)
- except IOError as err:
- LOG.warning('Failed to read monitor status: reason=%s', err)
- os.close(pipe[0])
- os.waitpid(child, 0)
- sys.exit(1)
-
- def stop(self):
-
- """Tell ServiceMonitor daemon to stop running.
-
- In case request fails stop ServiceMonitor and ceph_mgr proecsses
- using SIGTERM followed by SIGKILL.
- """
-
- result = ServiceMonitor.request_stop()
- if not result:
- ceph_mgr = os.path.basename(CONFIG.ceph_mgr_service)
- procs = []
- for proc in psutil.process_iter():
- name = proc.name()
- if name == CONFIG.service_name:
- procs.append(proc)
- if name == ceph_mgr:
- procs.append(proc)
- psutil_terminate_kill(procs, CONFIG.ceph_mgr_kill_delay_sec)
-
- def restart(self):
- self.stop()
- self.start()
-
- def force_reload(self):
- self.stop()
- self.start()
-
- def reload(self):
- self.stop()
- self.start()
-
- def status(self):
-
- """Report status from ServiceMonitor.
-
- We don't just try to access REST API here because ServiceMonitor may
- be in the process of starting/configuring ceph-mgr and restful
- plugin in which case we report OK to avoid being restarted by SM.
- """
-
- status = ServiceMonitor.request_status()
- sys.exit(0 if status is True else 1)
-
-
-if __name__ == '__main__':
- InitWrapper()
Code Review / pti / rtp.git / blobdiff