ceph/src/cephadm/box/util.py

   1 import json
   2 import os
   3 import subprocess
   4 import sys
   5 from typing import Any, Callable, Dict
   6
   7
   8 class Config:
   9     args = {
  10         'fsid': '00000000-0000-0000-0000-0000deadbeef',
  11         'config_folder': '/etc/ceph/',
  12         'config': '/etc/ceph/ceph.conf',
  13         'keyring': '/etc/ceph/ceph.keyring',
  14         'loop_img': 'loop-images/loop.img',
  15     }
  16
  17     @staticmethod
  18     def set(key, value):
  19         Config.args[key] = value
  20
  21     @staticmethod
  22     def get(key):
  23         if key in Config.args:
  24             return Config.args[key]
  25         return None
  26
  27     @staticmethod
  28     def add_args(args: Dict[str, str]) -> None:
  29         Config.args.update(args)
  30
  31
  32 class Target:
  33     def __init__(self, argv, subparsers):
  34         self.argv = argv
  35         self.parser = subparsers.add_parser(
  36             self.__class__.__name__.lower(), help=self.__class__._help
  37         )
  38
  39     def set_args(self):
  40         """
  41         adding the required arguments of the target should go here, example:
  42         self.parser.add_argument(..)
  43         """
  44         raise NotImplementedError()
  45
  46     def main(self):
  47         """
  48         A target will be setup by first calling this main function
  49         where the parser is initialized.
  50         """
  51         args = self.parser.parse_args(self.argv)
  52         Config.add_args(vars(args))
  53         function = getattr(self, args.action)
  54         function()
  55
  56
  57 def ensure_outside_container(func) -> Callable:
  58     def wrapper(*args, **kwargs):
  59         if not inside_container():
  60             return func(*args, **kwargs)
  61         else:
  62             raise RuntimeError('This command should be ran outside a container')
  63
  64     return wrapper
  65
  66
  67 def ensure_inside_container(func) -> bool:
  68     def wrapper(*args, **kwargs):
  69         if inside_container():
  70             return func(*args, **kwargs)
  71         else:
  72             raise RuntimeError('This command should be ran inside a container')
  73
  74     return wrapper
  75
  76
  77 def run_shell_command(command: str, expect_error=False) -> str:
  78     if Config.get('verbose'):
  79         print(f'Running command: {command}')
  80     process = subprocess.Popen(
  81         command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
  82     )
  83
  84     out = ''
  85     # let's read when output comes so it is in real time
  86     while True:
  87         # TODO: improve performance of this part, I think this part is a problem
  88         pout = process.stdout.read(1).decode('latin1')
  89         if pout == '' and process.poll() is not None:
  90             break
  91         if pout:
  92             if Config.get('verbose'):
  93                 sys.stdout.write(pout)
  94                 sys.stdout.flush()
  95             out += pout
  96     process.wait()
  97
  98     # no last break line
  99     err = (
 100         process.stderr.read().decode().rstrip()
 101     )  # remove trailing whitespaces and new lines
 102     out = out.strip()
 103
 104     if process.returncode != 0 and not expect_error:
 105         raise RuntimeError(f'Failed command: {command}\n{err}')
 106         sys.exit(1)
 107     return out
 108
 109
 110 @ensure_inside_container
 111 def run_cephadm_shell_command(command: str, expect_error=False) -> str:
 112     config = Config.get('config')
 113     keyring = Config.get('keyring')
 114
 115     with_cephadm_image = 'CEPHADM_IMAGE=quay.ceph.io/ceph-ci/ceph:master'
 116     out = run_shell_command(
 117         f'{with_cephadm_image} cephadm --verbose shell --config {config} --keyring {keyring} -- {command}',
 118         expect_error,
 119     )
 120     return out
 121
 122
 123 def run_dc_shell_command(
 124     command: str, index: int, box_type: str, expect_error=False
 125 ) -> str:
 126     out = run_shell_command(
 127         f'docker-compose exec --index={index} {box_type} {command}', expect_error
 128     )
 129     return out
 130
 131
 132 def inside_container() -> bool:
 133     return os.path.exists('/.dockerenv')
 134
 135
 136 @ensure_outside_container
 137 def get_boxes_container_info(with_seed: bool = False) -> Dict[str, Any]:
 138     # NOTE: this could be cached
 139     IP = 0
 140     CONTAINER_NAME = 1
 141     HOSTNAME = 2
 142     ips_query = "docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}} %tab% {{.Name}} %tab% {{.Config.Hostname}}' $(docker ps -aq) | sed 's#%tab%#\t#g' | sed 's#/##g' | sort -t . -k 1,1n -k 2,2n -k 3,3n -k 4,4n"
 143     out = run_shell_command(ips_query)
 144     # FIXME: if things get more complex a class representing a container info might be useful,
 145     # for now representing data this way is faster.
 146     info = {'size': 0, 'ips': [], 'container_names': [], 'hostnames': []}
 147     for line in out.split('\n'):
 148         container = line.split()
 149         # Most commands use hosts only
 150         name_filter = 'box_' if with_seed else 'box_hosts'
 151         if container[1].strip()[: len(name_filter)] == name_filter:
 152             info['size'] += 1
 153             info['ips'].append(container[IP])
 154             info['container_names'].append(container[CONTAINER_NAME])
 155             info['hostnames'].append(container[HOSTNAME])
 156     return info
 157
 158
 159 def get_orch_hosts():
 160     orch_host_ls_out = run_cephadm_shell_command('ceph orch host ls --format json')
 161     hosts = json.loads(orch_host_ls_out)
 162     return hosts