ceph/src/pybind/mgr/cephadm/services/exporter.py

   1 import json
   2 import logging
   3 from typing import TYPE_CHECKING, List, Dict, Any, Tuple
   4
   5 from orchestrator import OrchestratorError
   6 from mgr_util import ServerConfigException, verify_tls
   7
   8 from .cephadmservice import CephadmService, CephadmDaemonDeploySpec
   9
  10 if TYPE_CHECKING:
  11     from cephadm.module import CephadmOrchestrator
  12
  13 logger = logging.getLogger(__name__)
  14
  15
  16 class CephadmExporterConfig:
  17     required_keys = ['crt', 'key', 'token', 'port']
  18     DEFAULT_PORT = '9443'
  19
  20     def __init__(self, mgr, crt="", key="", token="", port=""):
  21         # type: (CephadmOrchestrator, str, str, str, str) -> None
  22         self.mgr = mgr
  23         self.crt = crt
  24         self.key = key
  25         self.token = token
  26         self.port = port
  27
  28     @property
  29     def ready(self) -> bool:
  30         return all([self.crt, self.key, self.token, self.port])
  31
  32     def load_from_store(self) -> None:
  33         cfg = self.mgr._get_exporter_config()
  34
  35         assert isinstance(cfg, dict)
  36         self.crt = cfg.get('crt', "")
  37         self.key = cfg.get('key', "")
  38         self.token = cfg.get('token', "")
  39         self.port = cfg.get('port', "")
  40
  41     def load_from_json(self, json_str: str) -> Tuple[int, str]:
  42         try:
  43             cfg = json.loads(json_str)
  44         except ValueError:
  45             return 1, "Invalid JSON provided - unable to load"
  46
  47         if not all([k in cfg for k in CephadmExporterConfig.required_keys]):
  48             return 1, "JSON file must contain crt, key, token and port"
  49
  50         self.crt = cfg.get('crt')
  51         self.key = cfg.get('key')
  52         self.token = cfg.get('token')
  53         self.port = cfg.get('port')
  54
  55         return 0, ""
  56
  57     def validate_config(self) -> Tuple[int, str]:
  58         if not self.ready:
  59             return 1, "Incomplete configuration. cephadm-exporter needs crt, key, token and port to be set"
  60
  61         for check in [self._validate_tls, self._validate_token, self._validate_port]:
  62             rc, reason = check()
  63             if rc:
  64                 return 1, reason
  65
  66         return 0, ""
  67
  68     def _validate_tls(self) -> Tuple[int, str]:
  69
  70         try:
  71             verify_tls(self.crt, self.key)
  72         except ServerConfigException as e:
  73             return 1, str(e)
  74
  75         return 0, ""
  76
  77     def _validate_token(self) -> Tuple[int, str]:
  78         if not isinstance(self.token, str):
  79             return 1, "token must be a string"
  80         if len(self.token) < 8:
  81             return 1, "Token must be a string of at least 8 chars in length"
  82
  83         return 0, ""
  84
  85     def _validate_port(self) -> Tuple[int, str]:
  86         try:
  87             p = int(str(self.port))
  88             if p <= 1024:
  89                 raise ValueError
  90         except ValueError:
  91             return 1, "Port must be a integer (>1024)"
  92
  93         return 0, ""
  94
  95
  96 class CephadmExporter(CephadmService):
  97     TYPE = 'cephadm-exporter'
  98
  99     def prepare_create(self, daemon_spec: CephadmDaemonDeploySpec) -> CephadmDaemonDeploySpec:
 100         assert self.TYPE == daemon_spec.daemon_type
 101
 102         cfg = CephadmExporterConfig(self.mgr)
 103         cfg.load_from_store()
 104
 105         if cfg.ready:
 106             rc, reason = cfg.validate_config()
 107             if rc:
 108                 raise OrchestratorError(reason)
 109         else:
 110             logger.info(
 111                 "Incomplete/Missing configuration, applying defaults")
 112             self.mgr._set_exporter_defaults()
 113             cfg.load_from_store()
 114
 115         if not daemon_spec.ports:
 116             daemon_spec.ports = [int(cfg.port)]
 117
 118         daemon_spec.final_config, daemon_spec.deps = self.generate_config(daemon_spec)
 119
 120         return daemon_spec
 121
 122     def generate_config(self, daemon_spec: CephadmDaemonDeploySpec) -> Tuple[Dict[str, Any], List[str]]:
 123         assert self.TYPE == daemon_spec.daemon_type
 124         deps: List[str] = []
 125
 126         cfg = CephadmExporterConfig(self.mgr)
 127         cfg.load_from_store()
 128
 129         if cfg.ready:
 130             rc, reason = cfg.validate_config()
 131             if rc:
 132                 raise OrchestratorError(reason)
 133         else:
 134             logger.info("Using default configuration for cephadm-exporter")
 135             self.mgr._set_exporter_defaults()
 136             cfg.load_from_store()
 137
 138         config = {
 139             "crt": cfg.crt,
 140             "key": cfg.key,
 141             "token": cfg.token
 142         }
 143         return config, deps
 144
 145     def purge(self, service_name: str) -> None:
 146         logger.info("Purging cephadm-exporter settings from mon K/V store")
 147         self.mgr._clear_exporter_config_settings()