+++ /dev/null
-"""
-A RESTful API for Ceph
-"""
-
-import os
-import json
-import time
-import errno
-import inspect
-import tempfile
-import threading
-import traceback
-import socket
-
-import common
-
-from uuid import uuid4
-from pecan import jsonify, make_app
-from OpenSSL import crypto
-from pecan.rest import RestController
-from werkzeug.serving import make_server, make_ssl_devcert
-
-from hooks import ErrorHook
-from mgr_module import MgrModule, CommandResult
-
-# Global instance to share
-instance = None
-
-
-class CannotServe(Exception):
- pass
-
-
-class CommandsRequest(object):
- """
- This class handles parallel as well as sequential execution of
- commands. The class accept a list of iterables that should be
- executed sequentially. Each iterable can contain several commands
- that can be executed in parallel.
-
- Example:
- [[c1,c2],[c3,c4]]
- - run c1 and c2 in parallel
- - wait for them to finish
- - run c3 and c4 in parallel
- - wait for them to finish
- """
-
-
- def __init__(self, commands_arrays):
- self.id = str(id(self))
-
- # Filter out empty sub-requests
- commands_arrays = filter(
- lambda x: len(x) != 0,
- commands_arrays,
- )
-
- self.running = []
- self.waiting = commands_arrays[1:]
- self.finished = []
- self.failed = []
-
- self.lock = threading.RLock()
- if not len(commands_arrays):
- # Nothing to run
- return
-
- # Process first iteration of commands_arrays in parallel
- results = self.run(commands_arrays[0])
-
- self.running.extend(results)
-
-
- def run(self, commands):
- """
- A static method that will execute the given list of commands in
- parallel and will return the list of command results.
- """
-
- # Gather the results (in parallel)
- results = []
- for index in range(len(commands)):
- tag = '%s:%d' % (str(self.id), index)
-
- # Store the result
- result = CommandResult(tag)
- result.command = common.humanify_command(commands[index])
- results.append(result)
-
- # Run the command
- instance.send_command(result, 'mon', '', json.dumps(commands[index]), tag)
-
- return results
-
-
- def next(self):
- with self.lock:
- if not self.waiting:
- # Nothing to run
- return
-
- # Run a next iteration of commands
- commands = self.waiting[0]
- self.waiting = self.waiting[1:]
-
- self.running.extend(self.run(commands))
-
-
- def finish(self, tag):
- with self.lock:
- for index in range(len(self.running)):
- if self.running[index].tag == tag:
- if self.running[index].r == 0:
- self.finished.append(self.running.pop(index))
- else:
- self.failed.append(self.running.pop(index))
- return True
-
- # No such tag found
- return False
-
-
- def is_running(self, tag):
- for result in self.running:
- if result.tag == tag:
- return True
- return False
-
-
- def is_ready(self):
- with self.lock:
- return not self.running and self.waiting
-
-
- def is_waiting(self):
- return bool(self.waiting)
-
-
- def is_finished(self):
- with self.lock:
- return not self.running and not self.waiting
-
-
- def has_failed(self):
- return bool(self.failed)
-
-
- def get_state(self):
- with self.lock:
- if not self.is_finished():
- return "pending"
-
- if self.has_failed():
- return "failed"
-
- return "success"
-
-
- def __json__(self):
- return {
- 'id': self.id,
- 'running': map(
- lambda x: {
- 'command': x.command,
- 'outs': x.outs,
- 'outb': x.outb,
- },
- self.running
- ),
- 'finished': map(
- lambda x: {
- 'command': x.command,
- 'outs': x.outs,
- 'outb': x.outb,
- },
- self.finished
- ),
- 'waiting': map(
- lambda x: {
- 'command': x.command,
- 'outs': x.outs,
- 'outb': x.outb,
- },
- self.waiting
- ),
- 'failed': map(
- lambda x: {
- 'command': x.command,
- 'outs': x.outs,
- 'outb': x.outb,
- },
- self.failed
- ),
- 'is_waiting': self.is_waiting(),
- 'is_finished': self.is_finished(),
- 'has_failed': self.has_failed(),
- 'state': self.get_state(),
- }
-
-
-
-class Module(MgrModule):
- COMMANDS = [
- {
- "cmd": "restful create-key name=key_name,type=CephString",
- "desc": "Create an API key with this name",
- "perm": "rw"
- },
- {
- "cmd": "restful delete-key name=key_name,type=CephString",
- "desc": "Delete an API key with this name",
- "perm": "rw"
- },
- {
- "cmd": "restful list-keys",
- "desc": "List all API keys",
- "perm": "rw"
- },
- {
- "cmd": "restful create-self-signed-cert",
- "desc": "Create localized self signed certificate",
- "perm": "rw"
- },
- {
- "cmd": "restful restart",
- "desc": "Restart API server",
- "perm": "rw"
- },
- ]
-
- def __init__(self, *args, **kwargs):
- super(Module, self).__init__(*args, **kwargs)
- global instance
- instance = self
-
- self.requests = []
- self.requests_lock = threading.RLock()
-
- self.keys = {}
- self.disable_auth = False
-
- self.server = None
-
- self.stop_server = False
- self.serve_event = threading.Event()
-
-
- def serve(self):
- while not self.stop_server:
- try:
- self._serve()
- self.server.socket.close()
- except CannotServe as cs:
- self.log.warn("server not running: {0}".format(cs.message))
- except:
- self.log.error(str(traceback.format_exc()))
-
- # Wait and clear the threading event
- self.serve_event.wait()
- self.serve_event.clear()
-
- def refresh_keys(self):
- self.keys = {}
- rawkeys = self.get_config_prefix('keys/') or {}
- for k, v in rawkeys.iteritems():
- self.keys[k[5:]] = v # strip of keys/ prefix
-
- def _serve(self):
- # Load stored authentication keys
- self.refresh_keys()
-
- jsonify._instance = jsonify.GenericJSON(
- sort_keys=True,
- indent=4,
- separators=(',', ': '),
- )
-
- server_addr = self.get_localized_config('server_addr', '::')
- if server_addr is None:
- raise CannotServe('no server_addr configured; try "ceph config-key set mgr/restful/server_addr <ip>"')
-
- server_port = int(self.get_localized_config('server_port', '8003'))
- self.log.info('server_addr: %s server_port: %d',
- server_addr, server_port)
-
- cert = self.get_localized_config("crt")
- if cert is not None:
- cert_tmp = tempfile.NamedTemporaryFile()
- cert_tmp.write(cert)
- cert_tmp.flush()
- cert_fname = cert_tmp.name
- else:
- cert_fname = self.get_localized_config('crt_file')
-
- pkey = self.get_localized_config("key")
- if pkey is not None:
- pkey_tmp = tempfile.NamedTemporaryFile()
- pkey_tmp.write(pkey)
- pkey_tmp.flush()
- pkey_fname = pkey_tmp.name
- else:
- pkey_fname = self.get_localized_config('key_file')
-
- if not cert_fname or not pkey_fname:
- raise CannotServe('no certificate configured')
- if not os.path.isfile(cert_fname):
- raise CannotServe('certificate %s does not exist' % cert_fname)
- if not os.path.isfile(pkey_fname):
- raise CannotServe('private key %s does not exist' % pkey_fname)
-
- # Publish the URI that others may use to access the service we're
- # about to start serving
- self.set_uri("https://{0}:{1}/".format(
- socket.gethostname() if server_addr == "::" else server_addr,
- server_port
- ))
-
- # Create the HTTPS werkzeug server serving pecan app
- self.server = make_server(
- host=server_addr,
- port=server_port,
- app=make_app(
- root='restful.api.Root',
- hooks = [ErrorHook()], # use a callable if pecan >= 0.3.2
- ),
- ssl_context=(cert_fname, pkey_fname),
- )
-
- self.server.serve_forever()
-
-
- def shutdown(self):
- try:
- self.stop_server = True
- if self.server:
- self.server.shutdown()
- self.serve_event.set()
- except:
- self.log.error(str(traceback.format_exc()))
- raise
-
-
- def restart(self):
- try:
- if self.server:
- self.server.shutdown()
- self.serve_event.set()
- except:
- self.log.error(str(traceback.format_exc()))
-
-
- def notify(self, notify_type, tag):
- try:
- self._notify(notify_type, tag)
- except:
- self.log.error(str(traceback.format_exc()))
-
-
- def _notify(self, notify_type, tag):
- if notify_type == "command":
- # we can safely skip all the sequential commands
- if tag == 'seq':
- return
-
- request = filter(
- lambda x: x.is_running(tag),
- self.requests)
-
- if len(request) != 1:
- self.log.warn("Unknown request '%s'" % str(tag))
- return
-
- request = request[0]
- request.finish(tag)
- if request.is_ready():
- request.next()
- else:
- self.log.debug("Unhandled notification type '%s'" % notify_type)
-
-
- def create_self_signed_cert(self):
- # create a key pair
- pkey = crypto.PKey()
- pkey.generate_key(crypto.TYPE_RSA, 2048)
-
- # create a self-signed cert
- cert = crypto.X509()
- cert.get_subject().O = "IT"
- cert.get_subject().CN = "ceph-restful"
- cert.set_serial_number(int(uuid4()))
- cert.gmtime_adj_notBefore(0)
- cert.gmtime_adj_notAfter(10*365*24*60*60)
- cert.set_issuer(cert.get_subject())
- cert.set_pubkey(pkey)
- cert.sign(pkey, 'sha512')
-
- return (
- crypto.dump_certificate(crypto.FILETYPE_PEM, cert),
- crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey)
- )
-
-
- def handle_command(self, command):
- self.log.warn("Handling command: '%s'" % str(command))
- if command['prefix'] == "restful create-key":
- if command['key_name'] in self.keys:
- return 0, self.keys[command['key_name']], ""
-
- else:
- key = str(uuid4())
- self.keys[command['key_name']] = key
- self.set_config('keys/' + command['key_name'], key)
-
- return (
- 0,
- self.keys[command['key_name']],
- "",
- )
-
- elif command['prefix'] == "restful delete-key":
- if command['key_name'] in self.keys:
- del self.keys[command['key_name']]
- self.set_config('keys/' + command['key_name'], None)
-
- return (
- 0,
- "",
- "",
- )
-
- elif command['prefix'] == "restful list-keys":
- self.refresh_keys()
- return (
- 0,
- json.dumps(self.keys, indent=2),
- "",
- )
-
- elif command['prefix'] == "restful create-self-signed-cert":
- cert, pkey = self.create_self_signed_cert()
-
- self.set_config(self.get_mgr_id() + '/crt', cert)
- self.set_config(self.get_mgr_id() + '/key', pkey)
-
- self.restart()
- return (
- 0,
- "Restarting RESTful API server...",
- ""
- )
-
- elif command['prefix'] == 'restful restart':
- self.restart();
- return (
- 0,
- "Restarting RESTful API server...",
- ""
- )
-
- else:
- return (
- -errno.EINVAL,
- "",
- "Command not found '{0}'".format(command['prefix'])
- )
-
-
- def get_doc_api(self, root, prefix=''):
- doc = {}
- for _obj in dir(root):
- obj = getattr(root, _obj)
-
- if isinstance(obj, RestController):
- doc.update(self.get_doc_api(obj, prefix + '/' + _obj))
-
- if getattr(root, '_lookup', None) and isinstance(root._lookup('0')[0], RestController):
- doc.update(self.get_doc_api(root._lookup('0')[0], prefix + '/<arg>'))
-
- prefix = prefix or '/'
-
- doc[prefix] = {}
- for method in 'get', 'post', 'patch', 'delete':
- if getattr(root, method, None):
- doc[prefix][method.upper()] = inspect.getdoc(getattr(root, method)).split('\n')
-
- if len(doc[prefix]) == 0:
- del doc[prefix]
-
- return doc
-
-
- def get_mons(self):
- mon_map_mons = self.get('mon_map')['mons']
- mon_status = json.loads(self.get('mon_status')['json'])
-
- # Add more information
- for mon in mon_map_mons:
- mon['in_quorum'] = mon['rank'] in mon_status['quorum']
- mon['server'] = self.get_metadata("mon", mon['name'])['hostname']
- mon['leader'] = mon['rank'] == mon_status['quorum'][0]
-
- return mon_map_mons
-
-
- def get_osd_pools(self):
- osds = dict(map(lambda x: (x['osd'], []), self.get('osd_map')['osds']))
- pools = dict(map(lambda x: (x['pool'], x), self.get('osd_map')['pools']))
- crush_rules = self.get('osd_map_crush')['rules']
-
- osds_by_pool = {}
- for pool_id, pool in pools.items():
- pool_osds = None
- for rule in [r for r in crush_rules if r['rule_id'] == pool['crush_rule']]:
- if rule['min_size'] <= pool['size'] <= rule['max_size']:
- pool_osds = common.crush_rule_osds(self.get('osd_map_tree')['nodes'], rule)
-
- osds_by_pool[pool_id] = pool_osds
-
- for pool_id in pools.keys():
- for in_pool_id in osds_by_pool[pool_id]:
- osds[in_pool_id].append(pool_id)
-
- return osds
-
-
- def get_osds(self, pool_id=None, ids=None):
- # Get data
- osd_map = self.get('osd_map')
- osd_metadata = self.get('osd_metadata')
-
- # Update the data with the additional info from the osd map
- osds = osd_map['osds']
-
- # Filter by osd ids
- if ids is not None:
- osds = filter(
- lambda x: str(x['osd']) in ids,
- osds
- )
-
- # Get list of pools per osd node
- pools_map = self.get_osd_pools()
-
- # map osd IDs to reweight
- reweight_map = dict([
- (x.get('id'), x.get('reweight', None))
- for x in self.get('osd_map_tree')['nodes']
- ])
-
- # Build OSD data objects
- for osd in osds:
- osd['pools'] = pools_map[osd['osd']]
- osd['server'] = osd_metadata.get(str(osd['osd']), {}).get('hostname', None)
-
- osd['reweight'] = reweight_map.get(osd['osd'], 0.0)
-
- if osd['up']:
- osd['valid_commands'] = common.OSD_IMPLEMENTED_COMMANDS
- else:
- osd['valid_commands'] = []
-
- # Filter by pool
- if pool_id:
- pool_id = int(pool_id)
- osds = filter(
- lambda x: pool_id in x['pools'],
- osds
- )
-
- return osds
-
-
- def get_osd_by_id(self, osd_id):
- osd = filter(
- lambda x: x['osd'] == osd_id,
- self.get('osd_map')['osds']
- )
-
- if len(osd) != 1:
- return None
-
- return osd[0]
-
-
- def get_pool_by_id(self, pool_id):
- pool = filter(
- lambda x: x['pool'] == pool_id,
- self.get('osd_map')['pools'],
- )
-
- if len(pool) != 1:
- return None
-
- return pool[0]
-
-
- def submit_request(self, _request, **kwargs):
- request = CommandsRequest(_request)
- with self.requests_lock:
- self.requests.append(request)
- if kwargs.get('wait', 0):
- while not request.is_finished():
- time.sleep(0.001)
- return request
-
-
- def run_command(self, command):
- # tag with 'seq' so that we can ingore these in notify function
- result = CommandResult('seq')
-
- self.send_command(result, 'mon', '', json.dumps(command), 'seq')
- return result.wait()
Code Review / stor4nfv.git / blobdiff