def osd_fsid(self) -> Optional[str]:
return self._osd_fsid
+ def default_entrypoint(self) -> str:
+ config_json = fetch_configs(self.ctx)
+ if (
+ config_json is not None
+ and config_json.get('osd_type') == 'crimson'
+ ):
+ return '/usr/bin/ceph-osd-crimson'
+ return '/usr/bin/ceph-osd'
+
+ @register_daemon_form
+ class Crash(Ceph):
+ @classmethod
+ def for_daemon_type(cls, daemon_type: str) -> bool:
+ return daemon_type == 'crash'
+
+ def __init__(
+ self,
+ ctx: CephadmContext,
+ ident: DaemonIdentity,
+ ) -> None:
+ super().__init__(ctx, ident)
+
+ @classmethod
+ def create(cls, ctx: CephadmContext, ident: DaemonIdentity) -> 'Ceph':
+ (uid, gid) = extract_uid_gid(ctx)
+ data_dir_base = f'{ctx.data_dir}/{ident.fsid}'
+ makedirs(
+ os.path.join(data_dir_base, 'crash'), uid, gid, DATA_DIR_MODE
+ )
+ makedirs(
+ os.path.join(data_dir_base, 'crash', 'posted'),
+ uid,
+ gid,
+ DATA_DIR_MODE,
+ )
+ return cls(ctx, ident)
+
+
@register_daemon_form
class CephExporter(ContainerDaemonForm):
"""Defines a Ceph exporter container"""