src/ceph/src/ceph-volume/ceph_volume/util/arg_validators.py

   1 import argparse
   2 import os
   3 from ceph_volume import terminal
   4 from ceph_volume import decorators
   5 from ceph_volume.util import disk
   6
   7
   8 class LVPath(object):
   9     """
  10     A simple validator to ensure that a logical volume is specified like::
  11
  12         <vg name>/<lv name>
  13
  14     Or a full path to a device, like ``/dev/sda``
  15
  16     Because for LVM it is better to be specific on what group does an lv
  17     belongs to.
  18     """
  19
  20     def __call__(self, string):
  21         error = None
  22         if string.startswith('/'):
  23             if not os.path.exists(string):
  24                 error = "Argument (device) does not exist: %s" % string
  25                 raise argparse.ArgumentError(None, error)
  26             else:
  27                 return string
  28         try:
  29             vg, lv = string.split('/')
  30         except ValueError:
  31             error = "Logical volume must be specified as 'volume_group/logical_volume' but got: %s" % string
  32             raise argparse.ArgumentError(None, error)
  33
  34         if not vg:
  35             error = "Didn't specify a volume group like 'volume_group/logical_volume', got: %s" % string
  36         if not lv:
  37             error = "Didn't specify a logical volume like 'volume_group/logical_volume', got: %s" % string
  38
  39         if error:
  40             raise argparse.ArgumentError(None, error)
  41         return string
  42
  43
  44 class OSDPath(object):
  45     """
  46     Validate path exists and it looks like an OSD directory.
  47     """
  48
  49     @decorators.needs_root
  50     def __call__(self, string):
  51         if not os.path.exists(string):
  52             error = "Path does not exist: %s" % string
  53             raise argparse.ArgumentError(None, error)
  54
  55         arg_is_partition = disk.is_partition(string)
  56         if arg_is_partition:
  57             return os.path.abspath(string)
  58         absolute_path = os.path.abspath(string)
  59         if not os.path.isdir(absolute_path):
  60             error = "Argument is not a directory or device which is required to scan"
  61             raise argparse.ArgumentError(None, error)
  62         key_files = ['ceph_fsid', 'fsid', 'keyring', 'ready', 'type', 'whoami']
  63         dir_files = os.listdir(absolute_path)
  64         for key_file in key_files:
  65             if key_file not in dir_files:
  66                 terminal.error('All following files must exist in path: %s' % ' '.join(key_files))
  67                 error = "Required file (%s) was not found in OSD dir path: %s" % (
  68                     key_file,
  69                     absolute_path
  70                 )
  71                 raise argparse.ArgumentError(None, error)
  72
  73         return os.path.abspath(string)