from ceph.deployment.utils import verify_positive_int, verify_non_negative_number
from ceph.deployment.utils import verify_boolean, verify_enum
from ceph.utils import is_hex
+from ceph.smb import constants as smbconst
ServiceSpecT = TypeVar('ServiceSpecT', bound='ServiceSpec')
FuncT = TypeVar('FuncT', bound=Callable)
class SMBSpec(ServiceSpec):
service_type = 'smb'
- _valid_features = {'domain', 'clustered', 'cephfs-proxy', 'remote-control'}
- _valid_service_names = {'smb', 'smbmetrics', 'ctdb', 'remote-control'}
+ _valid_features = smbconst.FEATURES
+ _valid_service_names = smbconst.SERVICES
_default_cluster_meta_obj = 'cluster.meta.json'
_default_cluster_lock_obj = 'cluster.meta.lock'
raise ValueError(
f'invalid feature flags: {", ".join(invalid)}'
)
- if 'clustered' in self.features and not self.cluster_meta_uri:
+ _clustered = smbconst.CLUSTERED
+ if _clustered in self.features and not self.cluster_meta_uri:
# derive a cluster meta uri from config uri by default (if possible)
self.cluster_meta_uri = self._derive_cluster_uri(
self.config_uri,
self._default_cluster_meta_obj,
)
- if 'clustered' not in self.features and self.cluster_meta_uri:
+ if _clustered not in self.features and self.cluster_meta_uri:
raise ValueError(
'cluster meta uri unsupported when "clustered" feature not set'
)
- if 'clustered' in self.features and not self.cluster_lock_uri:
+ if _clustered in self.features and not self.cluster_lock_uri:
# derive a cluster meta uri from config uri by default (if possible)
self.cluster_lock_uri = self._derive_cluster_uri(
self.config_uri,
self._default_cluster_lock_obj,
)
- if 'clustered' not in self.features and self.cluster_lock_uri:
+ if _clustered not in self.features and self.cluster_lock_uri:
raise ValueError(
'cluster lock uri unsupported when "clustered" feature not set'
)
return uri
def _default_ports(self) -> Dict[str, int]:
- return {
- 'smb': 445,
- 'smbmetrics': 9922,
- 'ctdb': 4379,
- 'remote-control': 54445,
- }
+ return dict(smbconst.DEFAULT_PORTS)
def service_ports(self) -> Dict[str, int]:
ports = self._default_ports()
return ports
def metrics_exporter_port(self) -> int:
- return self.service_ports()['smbmetrics']
+ return self.service_ports()[smbconst.SMBMETRICS]
def get_port_start(self) -> List[int]:
_ports = self.service_ports()
- ports = [_ports['smb'], _ports['smbmetrics']]
- if 'clustered' in self.features:
- ports.append(_ports['ctdb'])
+ ports = [_ports[smbconst.SMB], _ports[smbconst.SMBMETRICS]]
+ if smbconst.CLUSTERED in self.features:
+ ports.append(_ports[smbconst.CTDB])
return ports
def strict_cluster_ip_specs(self) -> List[Dict[str, Any]]: