-
-""".format(api_version=CephadmDaemonHandler.api_version)
-
- def _fetch_root(self):
- self.send_response(200)
- self.send_header('Content-type', 'text/html; charset=utf-8')
- self.end_headers()
- self.wfile.write(self._help_page().encode('utf-8'))
-
- @Decorators.authorize
- def do_GET(self):
- """Handle *all* GET requests"""
-
- if self.path == '/':
- # provide a html response if someone hits the root url, to document the
- # available api endpoints
- return self._fetch_root()
- elif self.path in CephadmDaemonHandler.valid_routes:
- u = self.path.split('/')[-1]
- data = json.dumps({})
- status_code = 200
-
- tasks = self.server.cephadm_cache.health.get('tasks', {})
- assert tasks
-
- # We're using the http status code to help indicate thread health
- # - 200 (OK): request successful
- # - 204 (No Content): access to a cache relating to a dead thread
- # - 206 (Partial content): one or more theads are inactive
- # - 500 (Server Error): all threads inactive
- if u == 'metadata':
- data = json.dumps(self.server.cephadm_cache.to_json())
- if all([tasks[task_name] == 'inactive' for task_name in tasks if task_name != 'http_server']):
- # All the subtasks are dead!
- status_code = 500
- elif any([tasks[task_name] == 'inactive' for task_name in tasks if task_name != 'http_server']):
- status_code = 206
-
- # Individual GETs against the a tasks endpoint will also return a 503 if the corresponding thread is inactive
- elif u == 'daemons':
- data = json.dumps(self.server.cephadm_cache.daemons)
- if tasks['daemons'] == 'inactive':
- status_code = 204
- elif u == 'disks':
- data = json.dumps(self.server.cephadm_cache.disks)
- if tasks['disks'] == 'inactive':
- status_code = 204
- elif u == 'host':
- data = json.dumps(self.server.cephadm_cache.host)
- if tasks['host'] == 'inactive':
- status_code = 204
-
- # a GET against health will always return a 200, since the op is always successful
- elif u == 'health':
- data = json.dumps(self.server.cephadm_cache.health)
-
- self.send_response(status_code)
- self.send_header('Content-type','application/json')
- self.end_headers()
- self.wfile.write(data.encode('utf-8'))
- else:
- # Invalid GET URL
- bad_request_msg = "Valid URLs are: {}".format(', '.join(CephadmDaemonHandler.valid_routes))
- self.send_response(404, message=bad_request_msg) # reason
- self.send_header('Content-type','application/json')
- self.end_headers()
- self.wfile.write(json.dumps({"message": bad_request_msg}).encode('utf-8'))
-
- def log_message(self, format, *args):
- rqst = " ".join(str(a) for a in args)
- logger.info(f"client:{self.address_string()} [{self.log_date_time_string()}] {rqst}")
-
-
-class CephadmDaemon():
-
- daemon_type = "cephadm-exporter"
- default_port = 9443
- bin_name = 'cephadm'
- key_name = "key"
- crt_name = "crt"
- token_name = "token"
- config_requirements = [
- key_name,
- crt_name,
- token_name,
- ]
- loop_delay = 1
- thread_check_interval = 5
-
- def __init__(self, ctx: CephadmContext, fsid, daemon_id=None, port=None):
- self.ctx = ctx
- self.fsid = fsid
- self.daemon_id = daemon_id
- if not port:
- self.port = CephadmDaemon.default_port
- else:
- self.port = port
- self.workers: List[Thread] = []
- self.http_server: CephadmHTTPServer
- self.stop = False
- self.cephadm_cache = CephadmCache()
- self.errors: List[str] = []
- self.token = read_file([os.path.join(self.daemon_path, CephadmDaemon.token_name)])
-
- @classmethod
- def validate_config(cls, config):
- reqs = ", ".join(CephadmDaemon.config_requirements)
- errors = []
-
- if not config or not all([k_name in config for k_name in CephadmDaemon.config_requirements]):
- raise Error(f"config must contain the following fields : {reqs}")
-
- if not all([isinstance(config[k_name], str) for k_name in CephadmDaemon.config_requirements]):
- errors.append(f"the following fields must be strings: {reqs}")
-
- crt = config[CephadmDaemon.crt_name]
- key = config[CephadmDaemon.key_name]
- token = config[CephadmDaemon.token_name]
-
- if not crt.startswith('-----BEGIN CERTIFICATE-----') or not crt.endswith('-----END CERTIFICATE-----\n'):
- errors.append("crt field is not a valid SSL certificate")
- if not key.startswith('-----BEGIN PRIVATE KEY-----') or not key.endswith('-----END PRIVATE KEY-----\n'):
- errors.append("key is not a valid SSL private key")
- if len(token) < 8:
- errors.append("'token' must be more than 8 characters long")
-
- if 'port' in config:
- try:
- p = int(config['port'])
- if p <= 1024:
- raise ValueError
- except (TypeError, ValueError):
- errors.append("port must be an integer > 1024")
-
- if errors:
- raise Error("Parameter errors : {}".format(", ".join(errors)))
-
- @property
- def port_active(self):
- return port_in_use(self.ctx, self.port)
-
- @property
- def can_run(self):
- # if port is in use
- if self.port_active:
- self.errors.append(f"TCP port {self.port} already in use, unable to bind")
- if not os.path.exists(os.path.join(self.daemon_path, CephadmDaemon.key_name)):
- self.errors.append(f"Key file '{CephadmDaemon.key_name}' is missing from {self.daemon_path}")
- if not os.path.exists(os.path.join(self.daemon_path, CephadmDaemon.crt_name)):
- self.errors.append(f"Certificate file '{CephadmDaemon.crt_name}' is missing from {self.daemon_path}")
- if self.token == "Unknown":
- self.errors.append(f"Authentication token '{CephadmDaemon.token_name}' is missing from {self.daemon_path}")
- return len(self.errors) == 0
-
- @staticmethod
- def _unit_name(fsid, daemon_id):
- return "{}.service".format(get_unit_name(fsid, CephadmDaemon.daemon_type, daemon_id))
-
- @property
- def unit_name(self):
- return CephadmDaemon._unit_name(self.fsid, self.daemon_id)
-
- @property
- def daemon_path(self):
- return os.path.join(
- self.ctx.data_dir,
- self.fsid,
- f'{self.daemon_type}.{self.daemon_id}'
- )
-
- @property
- def binary_path(self):
- return os.path.join(
- self.ctx.data_dir,
- self.fsid,
- CephadmDaemon.bin_name
- )
-
- def _handle_thread_exception(self, exc, thread_type):
- e_msg = f"{exc.__class__.__name__} exception: {str(exc)}"
- thread_info = getattr(self.cephadm_cache, thread_type)
- errors = thread_info.get('scrape_errors', [])
- errors.append(e_msg)
- logger.error(e_msg)
- logger.exception(exc)
- self.cephadm_cache.update_task(
- thread_type,
- {
- "scrape_errors": errors,
- "data": None,
- }
- )
-
- def _scrape_host_facts(self, refresh_interval=10):
- ctr = 0
- exception_encountered = False
-
- while True:
-
- if self.stop or exception_encountered:
- break
-
- if ctr >= refresh_interval:
- ctr = 0
- logger.debug("executing host-facts scrape")
- errors = []
- s_time = time.time()
-
- try:
- facts = HostFacts(self.ctx)
- except Exception as e:
- self._handle_thread_exception(e, 'host')
- exception_encountered = True
- else:
- elapsed = time.time() - s_time
- try:
- data = json.loads(facts.dump())
- except json.decoder.JSONDecodeError:
- errors.append("host-facts provided invalid JSON")
- logger.warning(errors[-1])
- data = {}
- self.cephadm_cache.update_task(
- 'host',
- {
- "scrape_timestamp": s_time,
- "scrape_duration_secs": elapsed,
- "scrape_errors": errors,
- "data": data,
- }
- )
- logger.debug(f"completed host-facts scrape - {elapsed}s")
-
- time.sleep(CephadmDaemon.loop_delay)
- ctr += CephadmDaemon.loop_delay
- logger.info("host-facts thread stopped")
-
- def _scrape_ceph_volume(self, refresh_interval=15):
- # we're invoking the ceph_volume command, so we need to set the args that it
- # expects to use
- self.ctx.command = "inventory --format=json".split()
- self.ctx.fsid = self.fsid
- self.ctx.log_output = False
-
- ctr = 0
- exception_encountered = False
-
- while True:
- if self.stop or exception_encountered:
- break
-
- if ctr >= refresh_interval:
- ctr = 0
- logger.debug("executing ceph-volume scrape")
- errors = []
- s_time = time.time()
- stream = io.StringIO()
- try:
- with redirect_stdout(stream):
- command_ceph_volume(self.ctx)
- except Exception as e:
- self._handle_thread_exception(e, 'disks')
- exception_encountered = True
- else:
- elapsed = time.time() - s_time
-
- # if the call to ceph-volume returns junk with the
- # json, it won't parse
- stdout = stream.getvalue()
-
- data = []
- if stdout:
- try:
- data = json.loads(stdout)
- except json.decoder.JSONDecodeError:
- errors.append("ceph-volume thread provided bad json data")
- logger.warning(errors[-1])
- else:
- errors.append("ceph-volume didn't return any data")
- logger.warning(errors[-1])
-
- self.cephadm_cache.update_task(
- 'disks',
- {
- "scrape_timestamp": s_time,
- "scrape_duration_secs": elapsed,
- "scrape_errors": errors,
- "data": data,
- }
- )
-
- logger.debug(f"completed ceph-volume scrape - {elapsed}s")
- time.sleep(CephadmDaemon.loop_delay)
- ctr += CephadmDaemon.loop_delay
-
- logger.info("ceph-volume thread stopped")
-
- def _scrape_list_daemons(self, refresh_interval=20):
- ctr = 0
- exception_encountered = False
- while True:
- if self.stop or exception_encountered:
- break
-
- if ctr >= refresh_interval:
- ctr = 0
- logger.debug("executing list-daemons scrape")
- errors = []
- s_time = time.time()
-
- try:
- # list daemons should ideally be invoked with a fsid
- data = list_daemons(self.ctx)
- except Exception as e:
- self._handle_thread_exception(e, 'daemons')
- exception_encountered = True
- else:
- if not isinstance(data, list):
- errors.append("list-daemons didn't supply a list?")
- logger.warning(errors[-1])
- data = []
- elapsed = time.time() - s_time
- self.cephadm_cache.update_task(
- 'daemons',
- {
- "scrape_timestamp": s_time,
- "scrape_duration_secs": elapsed,
- "scrape_errors": errors,
- "data": data,
- }
- )
- logger.debug(f"completed list-daemons scrape - {elapsed}s")
-
- time.sleep(CephadmDaemon.loop_delay)
- ctr += CephadmDaemon.loop_delay
- logger.info("list-daemons thread stopped")
-
- def _create_thread(self, target, name, refresh_interval=None):
- if refresh_interval:
- t = Thread(target=target, args=(refresh_interval,))
- else:
- t = Thread(target=target)
- t.daemon = True
- t.name = name
- self.cephadm_cache.update_health(name, "active")
- t.start()
-
- start_msg = f"Started {name} thread"
- if refresh_interval:
- logger.info(f"{start_msg}, with a refresh interval of {refresh_interval}s")
- else:
- logger.info(f"{start_msg}")
- return t
-
- def reload(self, *args):
- """reload -HUP received
-
- This is a placeholder function only, and serves to provide the hook that could
- be exploited later if the exporter evolves to incorporate a config file
- """
- logger.info("Reload request received - ignoring, no action needed")
-
- def shutdown(self, *args):
- logger.info("Shutdown request received")
- self.stop = True
- self.http_server.shutdown()
-
- def run(self):
- logger.info(f"cephadm exporter starting for FSID '{self.fsid}'")
- if not self.can_run:
- logger.error("Unable to start the exporter daemon")
- for e in self.errors:
- logger.error(e)
- return
-
- # register signal handlers for running under systemd control
- signal.signal(signal.SIGTERM, self.shutdown)
- signal.signal(signal.SIGINT, self.shutdown)
- signal.signal(signal.SIGHUP, self.reload)
- logger.debug("Signal handlers attached")
-
- host_facts = self._create_thread(self._scrape_host_facts, 'host', 5)
- self.workers.append(host_facts)
-
- daemons = self._create_thread(self._scrape_list_daemons, 'daemons', 20)
- self.workers.append(daemons)
-
- disks = self._create_thread(self._scrape_ceph_volume, 'disks', 20)
- self.workers.append(disks)
-
- self.http_server = CephadmHTTPServer(('0.0.0.0', self.port), CephadmDaemonHandler) # IPv4 only
- self.http_server.socket = ssl.wrap_socket(self.http_server.socket,
- keyfile=os.path.join(self.daemon_path, CephadmDaemon.key_name),
- certfile=os.path.join(self.daemon_path, CephadmDaemon.crt_name),
- server_side=True)
-
- self.http_server.cephadm_cache = self.cephadm_cache
- self.http_server.token = self.token
- server_thread = self._create_thread(self.http_server.serve_forever, 'http_server')
- logger.info(f"https server listening on {self.http_server.server_address[0]}:{self.http_server.server_port}")
-
- ctr = 0
- while server_thread.is_alive():
- if self.stop:
- break
-
- if ctr >= CephadmDaemon.thread_check_interval:
- ctr = 0
- for worker in self.workers:
- if self.cephadm_cache.tasks[worker.name] == 'inactive':
- continue
- if not worker.is_alive():
- logger.warning(f"{worker.name} thread not running")
- stop_time = datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S")
- self.cephadm_cache.update_health(worker.name, "inactive", f"{worker.name} stopped at {stop_time}")
-
- time.sleep(CephadmDaemon.loop_delay)
- ctr += CephadmDaemon.loop_delay
-
- logger.info("Main http server thread stopped")
-
- @property
- def unit_run(self):
-
- return """set -e
-{py3} {bin_path} exporter --fsid {fsid} --id {daemon_id} --port {port} &""".format(
- py3 = shutil.which('python3'),
- bin_path=self.binary_path,
- fsid=self.fsid,
- daemon_id=self.daemon_id,
- port=self.port
- )
-
- @property
- def unit_file(self):
- return """#generated by cephadm
-[Unit]
-Description=cephadm exporter service for cluster {fsid}
-After=network-online.target
-Wants=network-online.target
-
-PartOf=ceph-{fsid}.target
-Before=ceph-{fsid}.target
-
-[Service]
-Type=forking
-ExecStart=/bin/bash {daemon_path}/unit.run
-ExecReload=/bin/kill -HUP $MAINPID
-Restart=on-failure
-RestartSec=10s
-
-[Install]
-WantedBy=ceph-{fsid}.target
-""".format(
- fsid=self.fsid,
- daemon_path=self.daemon_path
-)
-
- def deploy_daemon_unit(self, config=None):
- """deploy a specific unit file for cephadm
-
- The normal deploy_daemon_units doesn't apply for this
- daemon since it's not a container, so we just create a
- simple service definition and add it to the fsid's target
- """
- if not config:
- raise Error("Attempting to deploy cephadm daemon without a config")
- assert isinstance(config, dict)
-
- # Create the required config files in the daemons dir, with restricted permissions
- for filename in config:
- with open(os.open(os.path.join(self.daemon_path, filename), os.O_CREAT | os.O_WRONLY, mode=0o600), "w") as f:
- f.write(config[filename])
-
- # When __file__ is