Merge pull request ceph#36890 from sebastian-philipp/cephadm-extend-c…

…eph.conf

mgr/cephadm: Add extra-ceph-conf

src/pybind/mgr/cephadm/inventory.py

@@ -7,6 +7,7 @@
 import orchestrator
 from ceph.deployment import inventory
 from ceph.deployment.service_spec import ServiceSpec
+from cephadm.utils import str_to_datetime, datetime_to_str
 from orchestrator import OrchestratorError, HostSpec, OrchestratorEvent
 
 if TYPE_CHECKING:
@@ -17,7 +18,6 @@
 
 HOST_CACHE_PREFIX = "host."
 SPEC_STORE_PREFIX = "spec."
-DATEFMT = '%Y-%m-%dT%H:%M:%S.%f'
 
 
 class Inventory:
@@ -116,7 +116,7 @@ def load(self):
             try:
                 v = json.loads(v)
                 spec = ServiceSpec.from_json(v['spec'])
-                created = datetime.datetime.strptime(v['created'], DATEFMT)
+                created = str_to_datetime(v['created'])
                 self.specs[service_name] = spec
                 self.spec_created[service_name] = created
                 self.mgr.log.debug('SpecStore: loaded spec for %s' % (
@@ -137,7 +137,7 @@ def save(self, spec):
             SPEC_STORE_PREFIX + spec.service_name(),
             json.dumps({
                 'spec': spec.to_json(),
-                'created': self.spec_created[spec.service_name()].strftime(DATEFMT),
+                'created': datetime_to_str(self.spec_created[spec.service_name()]),
             }, sort_keys=True),
         )
         self.mgr.events.for_service(spec, OrchestratorEvent.INFO, 'service was created')
@@ -197,8 +197,7 @@ def load(self):
             try:
                 j = json.loads(v)
                 if 'last_device_update' in j:
-                    self.last_device_update[host] = datetime.datetime.strptime(
-                        j['last_device_update'], DATEFMT)
+                    self.last_device_update[host] = str_to_datetime(j['last_device_update'])
                 else:
                     self.device_refresh_queue.append(host)
                 # for services, we ignore the persisted last_*_update
@@ -220,15 +219,13 @@ def load(self):
                 for name, d in j.get('daemon_config_deps', {}).items():
                     self.daemon_config_deps[host][name] = {
                         'deps': d.get('deps', []),
-                        'last_config': datetime.datetime.strptime(
-                            d['last_config'], DATEFMT),
+                        'last_config': str_to_datetime(d['last_config']),
                     }
                 if 'last_host_check' in j:
-                    self.last_host_check[host] = datetime.datetime.strptime(
-                        j['last_host_check'], DATEFMT)
+                    self.last_host_check[host] = str_to_datetime(j['last_host_check'])
                 if 'last_etc_ceph_ceph_conf' in j:
-                    self.last_etc_ceph_ceph_conf[host] = datetime.datetime.strptime(
-                        j['last_etc_ceph_ceph_conf'], DATEFMT)
+                    self.last_etc_ceph_ceph_conf[host] = str_to_datetime(
+                        j['last_etc_ceph_ceph_conf'])
                 self.registry_login_queue.add(host)
                 self.scheduled_daemon_actions[host] = j.get('scheduled_daemon_actions', {})
 
@@ -303,11 +300,9 @@ def save_host(self, host: str) -> None:
             'daemon_config_deps': {},
         }
         if host in self.last_daemon_update:
-            j['last_daemon_update'] = self.last_daemon_update[host].strftime(
-                DATEFMT)
+            j['last_daemon_update'] = datetime_to_str(self.last_daemon_update[host])
         if host in self.last_device_update:
-            j['last_device_update'] = self.last_device_update[host].strftime(
-                DATEFMT)
+            j['last_device_update'] = datetime_to_str(self.last_device_update[host])
         for name, dd in self.daemons[host].items():
             j['daemons'][name] = dd.to_json()
         for d in self.devices[host]:
@@ -316,16 +311,16 @@ def save_host(self, host: str) -> None:
         for name, depi in self.daemon_config_deps[host].items():
             j['daemon_config_deps'][name] = {
                 'deps': depi.get('deps', []),
-                'last_config': depi['last_config'].strftime(DATEFMT),
+                'last_config': datetime_to_str(depi['last_config']),
             }
         if self.osdspec_previews[host]:
             j['osdspec_previews'] = self.osdspec_previews[host]
 
         if host in self.last_host_check:
-            j['last_host_check'] = self.last_host_check[host].strftime(DATEFMT)
+            j['last_host_check'] = datetime_to_str(self.last_host_check[host])
 
         if host in self.last_etc_ceph_ceph_conf:
-            j['last_etc_ceph_ceph_conf'] = self.last_etc_ceph_ceph_conf[host].strftime(DATEFMT)
+            j['last_etc_ceph_ceph_conf'] = datetime_to_str(self.last_etc_ceph_ceph_conf[host])
         if self.scheduled_daemon_actions.get(host, {}):
             j['scheduled_daemon_actions'] = self.scheduled_daemon_actions[host]
 
@@ -413,7 +408,7 @@ def get_daemon_names(self):
                 r.append(name)
         return r
 
-    def get_daemon_last_config_deps(self, host, name):
+    def get_daemon_last_config_deps(self, host, name) -> Tuple[Optional[List[str]], Optional[datetime.datetime]]:
         if host in self.daemon_config_deps:
             if name in self.daemon_config_deps[host]:
                 return self.daemon_config_deps[host][name].get('deps', []), \
@@ -488,13 +483,15 @@ def host_needs_new_etc_ceph_ceph_conf(self, host: str):
             return True
         if self.mgr.last_monmap > self.last_etc_ceph_ceph_conf[host]:
             return True
+        if self.mgr.extra_ceph_conf_is_newer(self.last_etc_ceph_ceph_conf[host]):
+            return True
         # already up to date:
         return False
 
     def update_last_etc_ceph_ceph_conf(self, host: str):
         if not self.mgr.last_monmap:
             return
-        self.last_etc_ceph_ceph_conf[host] = self.mgr.last_monmap
+        self.last_etc_ceph_ceph_conf[host] = datetime.datetime.utcnow()
 
     def host_needs_registry_login(self, host: str) -> bool:
         if host in self.mgr.offline_hosts:

src/pybind/mgr/cephadm/module.py

@@ -3,6 +3,7 @@
 import logging
 import shlex
 from collections import defaultdict
+from configparser import ConfigParser
 from contextlib import contextmanager
 from functools import wraps
 from tempfile import TemporaryDirectory
@@ -48,7 +49,8 @@
 from .inventory import Inventory, SpecStore, HostCache, EventStore
 from .upgrade import CEPH_UPGRADE_ORDER, CephadmUpgrade
 from .template import TemplateMgr
-from .utils import forall_hosts, CephadmNoImage, cephadmNoImage, is_repo_digest
+from .utils import forall_hosts, CephadmNoImage, cephadmNoImage, \
+    str_to_datetime, datetime_to_str, is_repo_digest
 
 try:
     import remoto
@@ -84,7 +86,6 @@ def remoto_has_connection(self):
   ConnectTimeout=30
 """
 
-DATEFMT = '%Y-%m-%dT%H:%M:%S.%f'
 CEPH_DATEFMT = '%Y-%m-%dT%H:%M:%S.%fZ'
 
 CEPH_TYPES = set(CEPH_UPGRADE_ORDER)
@@ -1028,6 +1029,53 @@ def _prepare_host(self, host, addr=None):
                     self.event.set()
         return 0, '%s (%s) ok' % (host, addr), err
 
+    @orchestrator._cli_write_command(
+        prefix='cephadm set-extra-ceph-conf',
+        desc="Text that is appended to all daemon's ceph.conf.\n"
+             "Mainly a workaround, till `config generate-minimal-conf` generates\n"
+             "a complete ceph.conf.\n\n"
+             "Warning: this is a dangerous operation.")
+    def _set_extra_ceph_conf(self, inbuf=None) -> HandleCommandResult:
+        if inbuf:
+            # sanity check.
+            cp = ConfigParser()
+            cp.read_string(inbuf, source='<infile>')
+
+        self.set_store("extra_ceph_conf", json.dumps({
+            'conf': inbuf,
+            'last_modified': datetime_to_str(datetime.datetime.utcnow())
+        }))
+        self.log.info('Set extra_ceph_conf')
+        self._kick_serve_loop()
+        return HandleCommandResult()
+
+    @orchestrator._cli_read_command(
+        'cephadm get-extra-ceph-conf',
+        desc='Get extra ceph conf that is appended')
+    def _get_extra_ceph_conf(self) -> HandleCommandResult:
+        return HandleCommandResult(stdout=self.extra_ceph_conf().conf)
+
+    class ExtraCephConf(NamedTuple):
+        conf: str
+        last_modified: Optional[datetime.datetime]
+
+    def extra_ceph_conf(self) -> 'CephadmOrchestrator.ExtraCephConf':
+        data = self.get_store('extra_ceph_conf')
+        if not data:
+            return CephadmOrchestrator.ExtraCephConf('', None)
+        try:
+            j = json.loads(data)
+        except ValueError:
+            self.log.exception('unable to laod extra_ceph_conf')
+            return CephadmOrchestrator.ExtraCephConf('', None)
+        return CephadmOrchestrator.ExtraCephConf(j['conf'], str_to_datetime(j['last_modified']))
+
+    def extra_ceph_conf_is_newer(self, dt: datetime.datetime) -> bool:
+        conf = self.extra_ceph_conf()
+        if not conf.last_modified:
+            return False
+        return conf.last_modified > dt
+
     def _get_connection(self, host: str):
         """
         Setup a connection for running commands on remote host.
@@ -1458,7 +1506,7 @@ def _refresh_host_daemons(self, host) -> Optional[str]:
             for k in ['created', 'started', 'last_configured', 'last_deployed']:
                 v = d.get(k, None)
                 if v:
-                    setattr(sd, k, datetime.datetime.strptime(d[k], DATEFMT))
+                    setattr(sd, k, str_to_datetime(d[k]))
             sd.daemon_type = d['name'].split('.')[0]
             sd.daemon_id = '.'.join(d['name'].split('.')[1:])
             sd.hostname = host
@@ -1521,9 +1569,7 @@ def _refresh_host_devices(self, host) -> Optional[str]:
         return None
 
     def _deploy_etc_ceph_ceph_conf(self, host: str) -> Optional[str]:
-        ret, config, err = self.check_mon_command({
-            "prefix": "config generate-minimal-conf",
-        })
+        config = self.get_minimal_ceph_conf()
 
         try:
             with self._remote_connection(host) as tpl:
@@ -1546,6 +1592,15 @@ def _deploy_etc_ceph_ceph_conf(self, host: str) -> Optional[str]:
             return f'failed to create /etc/ceph/ceph.conf on {host}: {str(e)}'
         return None
 
+    def get_minimal_ceph_conf(self) -> str:
+        _, config, _ = self.check_mon_command({
+            "prefix": "config generate-minimal-conf",
+        })
+        extra = self.extra_ceph_conf().conf
+        if extra:
+            config += '\n\n' + extra.strip() + '\n'
+        return config
+
     def _invalidate_daemons_and_kick_serve(self, filter_host=None):
         if filter_host:
             self.cache.invalidate_host_daemons(filter_host)
@@ -2301,6 +2356,10 @@ def _check_daemons(self):
                 dd.daemon_type in CEPH_TYPES:
                 self.log.info('Reconfiguring %s (monmap changed)...' % dd.name())
                 action = 'reconfig'
+            elif self.extra_ceph_conf_is_newer(last_config) and \
+                    dd.daemon_type in CEPH_TYPES:
+                self.log.info('Reconfiguring %s (extra config changed)...' % dd.name())
+                action = 'reconfig'
             if action:
                 if self.cache.get_scheduled_daemon_action(dd.hostname, dd.name()) == 'redeploy' \
                         and action == 'reconfig':

src/pybind/mgr/cephadm/schedule.py

@@ -144,7 +144,7 @@ def place(self):
             if self.filter_new_host:
                 old = others
                 others = [h for h in others if self.filter_new_host(h.hostname)]
-                logger.debug('filtered %s down to %s' % (old, candidates))
+                logger.debug('filtered %s down to %s' % (old, others))
 
             # ask the scheduler to return a set of hosts with a up to the value of <count>
             others = self.scheduler.place(others, need)

src/pybind/mgr/cephadm/services/cephadmservice.py

@@ -29,7 +29,8 @@ def __init__(self, host: str, daemon_id,
                  network: Optional[str] = None,
                  keyring: Optional[str] = None,
                  extra_args: Optional[List[str]] = None,
-                 extra_config: Optional[Dict[str, Any]] = None,
+                 ceph_conf: str = '',
+                 extra_files: Optional[Dict[str, Any]] = None,
                  daemon_type: Optional[str] = None,
                  ports: Optional[List[int]] = None,):
         """
@@ -56,14 +57,23 @@ def __init__(self, host: str, daemon_id,
 
         # For run_cephadm. Would be great to have more expressive names.
         self.extra_args: List[str] = extra_args or []
-        self.extra_config: Dict[str, Any] = extra_config or {}
+
+        self.ceph_conf = ceph_conf
+        self.extra_files = extra_files or {}
 
         # TCP ports used by the daemon
         self.ports:  List[int] = ports or []
 
     def name(self) -> str:
         return '%s.%s' % (self.daemon_type, self.daemon_id)
 
+    def config_get_files(self):
+        files = self.extra_files
+        if self.ceph_conf:
+            files['config'] = self.ceph_conf
+
+        return files
+
 
 class CephadmService(metaclass=ABCMeta):
     """
@@ -235,10 +245,10 @@ def generate_config(self, daemon_spec: CephadmDaemonSpec) -> Tuple[Dict[str, Any
             daemon_spec.daemon_id,
             host=daemon_spec.host,
             keyring=daemon_spec.keyring,
-            extra_ceph_config=daemon_spec.extra_config.pop('config', ''))
+            extra_ceph_config=daemon_spec.ceph_conf)
 
-        if daemon_spec.extra_config:
-            cephadm_config.update({'files': daemon_spec.extra_config})
+        if daemon_spec.config_get_files():
+            cephadm_config.update({'files': daemon_spec.config_get_files()})
 
         return cephadm_config, []
 
@@ -278,10 +288,8 @@ def get_config_and_keyring(self,
                 'entity': entity,
             })
 
-        # generate config
-        ret, config, err = self.mgr.check_mon_command({
-            "prefix": "config generate-minimal-conf",
-        })
+        config = self.mgr.get_minimal_ceph_conf()
+
         if extra_ceph_config:
             config += extra_ceph_config
 
@@ -353,7 +361,7 @@ def prepare_create(self, daemon_spec: CephadmDaemonSpec) -> CephadmDaemonSpec:
                     'public_network is set but does not look like a CIDR network: \'%s\'' % network)
             extra_config += 'public network = %s\n' % network
 
-        daemon_spec.extra_config = {'config': extra_config}
+        daemon_spec.ceph_conf = extra_config
         daemon_spec.keyring = keyring
 
         return daemon_spec

src/pybind/mgr/cephadm/services/iscsi.py

@@ -68,7 +68,7 @@ def prepare_create(self, daemon_spec: CephadmDaemonSpec[IscsiServiceSpec]) -> Ce
         igw_conf = self.mgr.template.render('services/iscsi/iscsi-gateway.cfg.j2', context)
 
         daemon_spec.keyring = keyring
-        daemon_spec.extra_config = {'iscsi-gateway.cfg': igw_conf}
+        daemon_spec.extra_files = {'iscsi-gateway.cfg': igw_conf}
 
         return daemon_spec
 

src/pybind/mgr/cephadm/services/osd.py

@@ -8,14 +8,13 @@
 
 from datetime import datetime
 import orchestrator
-from cephadm.utils import forall_hosts
+from cephadm.utils import forall_hosts, datetime_to_str, str_to_datetime
 from orchestrator import OrchestratorError
 from mgr_module import MonCommandFailed
 
 from cephadm.services.cephadmservice import CephadmDaemonSpec, CephService
 
 logger = logging.getLogger(__name__)
-DATEFMT = '%Y-%m-%dT%H:%M:%S.%f'
 
 
 class OSDService(CephService):
@@ -237,13 +236,8 @@ def _run_ceph_volume_command(self, host: str,
             'entity': 'client.bootstrap-osd',
         })
 
-        # generate config
-        ret, config, err = self.mgr.check_mon_command({
-            "prefix": "config generate-minimal-conf",
-        })
-
         j = json.dumps({
-            'config': config,
+            'config': self.mgr.get_minimal_ceph_conf(),
             'keyring': keyring,
         })
 
@@ -615,7 +609,7 @@ def pg_count_str(self):
         return 'n/a' if self.get_pg_count() < 0 else str(self.get_pg_count())
 
     def to_json(self) -> dict:
-        out = dict()
+        out: Dict[str, Any] = dict()
         out['osd_id'] = self.osd_id
         out['started'] = self.started
         out['draining'] = self.draining
@@ -626,7 +620,7 @@ def to_json(self) -> dict:
 
         for k in ['drain_started_at', 'drain_stopped_at', 'drain_done_at', 'process_started_at']:
             if getattr(self, k):
-                out[k] = getattr(self, k).strftime(DATEFMT)
+                out[k] = datetime_to_str(getattr(self, k))
             else:
                 out[k] = getattr(self, k)
         return out
@@ -637,7 +631,7 @@ def from_json(cls, inp: Optional[Dict[str, Any]], ctx: Optional[RemoveUtil] = No
             return None
         for date_field in ['drain_started_at', 'drain_stopped_at', 'drain_done_at', 'process_started_at']:
             if inp.get(date_field):
-                inp.update({date_field: datetime.strptime(inp.get(date_field, ''), DATEFMT)})
+                inp.update({date_field: str_to_datetime(inp.get(date_field, ''))})
         inp.update({'remove_util': ctx})
         if 'nodename' in inp:
             hostname = inp.pop('nodename')