import re
from collections import OrderedDict
from functools import wraps
+from ipaddress import ip_network, ip_address
from typing import Optional, Dict, Any, List, Union, Callable, Iterable, Type, TypeVar, cast, \
NamedTuple
if not require_network:
return host_spec
- from ipaddress import ip_network, ip_address
networks = list() # type: List[str]
network = host_spec.network
# in case we have [v2:1.2.3.4:3000,v1:1.2.3.4:6478]
config: Optional[Dict[str, str]] = None,
unmanaged: bool = False,
preview_only: bool = False,
+ networks: Optional[List[str]] = None,
):
self.placement = PlacementSpec() if placement is None else placement # type: PlacementSpec
self.service_id = service_id
self.unmanaged = unmanaged
self.preview_only = preview_only
+ self.networks: List[str] = networks or []
self.config: Optional[Dict[str, str]] = None
if config:
service_id: foo
config:
some_option: the_value
+ networks: [10.10.0.0/16]
spec:
pool: mypool
namespace: myns
ret['placement'] = self.placement.to_json()
if self.unmanaged:
ret['unmanaged'] = self.unmanaged
+ if self.networks:
+ ret['networks'] = self.networks
c = {}
for key, val in sorted(self.__dict__.items(), key=lambda tpl: tpl[0]):
raise ServiceSpecValidationError(
f'Cannot set config option {k} in spec: it is managed by cephadm'
)
+ for network in self.networks or []:
+ try:
+ ip_network(network)
+ except ValueError as e:
+ raise ServiceSpecValidationError(
+ f'Cannot parse network {network}: {e}'
+ )
def __repr__(self) -> str:
return "{}({!r})".format(self.__class__.__name__, self.__dict__)
unmanaged: bool = False,
preview_only: bool = False,
config: Optional[Dict[str, str]] = None,
+ networks: Optional[List[str]] = None,
):
assert service_type == 'nfs'
super(NFSServiceSpec, self).__init__(
'nfs', service_id=service_id,
placement=placement, unmanaged=unmanaged, preview_only=preview_only,
- config=config)
+ config=config, networks=networks)
#: RADOS pool where NFS client recovery data is stored.
self.pool = pool
ssl: bool = False,
preview_only: bool = False,
config: Optional[Dict[str, str]] = None,
+ networks: Optional[List[str]] = None,
subcluster: Optional[str] = None, # legacy, only for from_json on upgrade
):
assert service_type == 'rgw', service_type
super(RGWSpec, self).__init__(
'rgw', service_id=service_id,
placement=placement, unmanaged=unmanaged,
- preview_only=preview_only, config=config)
+ preview_only=preview_only, config=config, networks=networks)
self.rgw_realm = rgw_realm
self.rgw_zone = rgw_zone
unmanaged: bool = False,
preview_only: bool = False,
config: Optional[Dict[str, str]] = None,
+ networks: Optional[List[str]] = None,
):
assert service_type == 'iscsi'
super(IscsiServiceSpec, self).__init__('iscsi', service_id=service_id,
placement=placement, unmanaged=unmanaged,
preview_only=preview_only,
- config=config)
+ config=config, networks=networks)
#: RADOS pool where ceph-iscsi config data is stored.
self.pool = pool
preview_only: bool = False,
user_data: Optional[Dict[str, Any]] = None,
config: Optional[Dict[str, str]] = None,
+ networks: Optional[List[str]] = None,
):
assert service_type == 'alertmanager'
super(AlertManagerSpec, self).__init__(
'alertmanager', service_id=service_id,
placement=placement, unmanaged=unmanaged,
- preview_only=preview_only, config=config)
+ preview_only=preview_only, config=config, networks=networks)
# Custom configuration.
#
service_type: str = 'ha-rgw',
service_id: Optional[str] = None,
config: Optional[Dict[str, str]] = None,
+ networks: Optional[List[str]] = None,
placement: Optional[PlacementSpec] = None,
virtual_ip_interface: Optional[str] = None,
virtual_ip_address: Optional[str] = None,
):
assert service_type == 'ha-rgw'
super(HA_RGWSpec, self).__init__('ha-rgw', service_id=service_id,
- placement=placement, config=config)
+ placement=placement, config=config,
+ networks=networks)
self.virtual_ip_interface = virtual_ip_interface
self.virtual_ip_address = virtual_ip_address
service_type: str = 'container',
service_id: Optional[str] = None,
config: Optional[Dict[str, str]] = None,
+ networks: Optional[List[str]] = None,
placement: Optional[PlacementSpec] = None,
unmanaged: bool = False,
preview_only: bool = False,
super(CustomContainerSpec, self).__init__(
service_type, service_id,
placement=placement, unmanaged=unmanaged,
- preview_only=preview_only, config=config)
+ preview_only=preview_only, config=config,
+ networks=networks)
self.image = image
self.entrypoint = entrypoint