"""Defines a custom container"""
daemon_type = 'container'
- def __init__(self, ctx: CephadmContext,
+ def __init__(self,
fsid: str, daemon_id: Union[int, str],
config_json: Dict, image: str) -> None:
- self.ctx = ctx
self.fsid = fsid
self.daemon_id = daemon_id
self.image = image
@classmethod
def init(cls, ctx: CephadmContext,
fsid: str, daemon_id: Union[int, str]) -> 'CustomContainer':
- return cls(ctx, fsid, daemon_id,
+ return cls(fsid, daemon_id,
get_parm(ctx.args.config_json), ctx.args.image)
def create_daemon_dirs(self, data_dir: str, uid: int, gid: int) -> None:
# type: (CephadmContext, str, int) -> None
if not ctx.args.skip_ping_check:
logger.info('Verifying IP %s port %d ...' % (ip, port))
- if is_ipv6(ctx, ip):
+ if is_ipv6(ip):
s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
ip = unwrap_ipv6(ip)
else:
'--filter', 'label=ceph=True',
'--filter', 'dangling=false',
'--format', '{{.Repository}}@{{.Digest}}'])
- return _filter_last_local_ceph_image(ctx, out)
+ return _filter_last_local_ceph_image(out)
-def _filter_last_local_ceph_image(ctx, out):
- # type: (CephadmContext, str) -> Optional[str]
+def _filter_last_local_ceph_image(out):
+ # type: (str) -> Optional[str]
for image in out.splitlines():
if image and not image.endswith('@'):
logger.info('Using recent ceph image %s' % image)
return address
-def is_ipv6(ctx, address):
- # type: (CephadmContext, str) -> bool
+def is_ipv6(address):
+ # type: (str) -> bool
address = unwrap_ipv6(address)
try:
return ipaddress.ip_address(unicode(address)).version == 6
ipv6 = False
if ctx.args.mon_ip:
- ipv6 = is_ipv6(ctx, ctx.args.mon_ip)
+ ipv6 = is_ipv6(ctx.args.mon_ip)
if ipv6:
ctx.args.mon_ip = wrap_ipv6(ctx.args.mon_ip)
hasport = r.findall(ctx.args.mon_ip)