ceph/src/test/rgw/test_multi.py

   1 import subprocess
   2 import os
   3 import random
   4 import string
   5 import argparse
   6 import sys
   7 import logging
   8 try:
   9     import configparser
  10 except ImportError:
  11     import ConfigParser as configparser
  12
  13 import nose.core
  14
  15 from rgw_multi import multisite
  16 from rgw_multi.zone_rados import RadosZone as RadosZone
  17 from rgw_multi.zone_es import ESZone as ESZone
  18 from rgw_multi.zone_es import ESZoneConfig as ESZoneConfig
  19 from rgw_multi.zone_cloud import CloudZone as CloudZone
  20 from rgw_multi.zone_cloud import CloudZoneConfig as CloudZoneConfig
  21 from rgw_multi.zone_az import AZone as AZone
  22 from rgw_multi.zone_az import AZoneConfig as AZoneConfig
  23
  24 # make tests from rgw_multi.tests available to nose
  25 from rgw_multi.tests import *
  26 from rgw_multi.tests_es import *
  27 from rgw_multi.tests_az import *
  28
  29 mstart_path = os.getenv('MSTART_PATH')
  30 if mstart_path is None:
  31     mstart_path = os.path.normpath(os.path.dirname(os.path.realpath(__file__)) + '/../..') + '/'
  32
  33 test_path = os.path.normpath(os.path.dirname(os.path.realpath(__file__))) + '/'
  34
  35 # configure logging for the tests module
  36 log = logging.getLogger('rgw_multi.tests')
  37
  38 def bash(cmd, **kwargs):
  39     log.debug('running cmd: %s', ' '.join(cmd))
  40     check_retcode = kwargs.pop('check_retcode', True)
  41     kwargs['stdout'] = subprocess.PIPE
  42     process = subprocess.Popen(cmd, **kwargs)
  43     s = process.communicate()[0].decode('utf-8')
  44     log.debug('command returned status=%d stdout=%s', process.returncode, s)
  45     if check_retcode:
  46         assert(process.returncode == 0)
  47     return (s, process.returncode)
  48
  49 class Cluster(multisite.Cluster):
  50     """ cluster implementation based on mstart/mrun scripts """
  51     def __init__(self, cluster_id):
  52         super(Cluster, self).__init__()
  53         self.cluster_id = cluster_id
  54         self.needs_reset = True
  55
  56     def admin(self, args = None, **kwargs):
  57         """ radosgw-admin command """
  58         cmd = [test_path + 'test-rgw-call.sh', 'call_rgw_admin', self.cluster_id]
  59         if args:
  60             cmd += args
  61         cmd += ['--debug-rgw=' + str(kwargs.pop('debug_rgw', 0))]
  62         cmd += ['--debug-ms=' + str(kwargs.pop('debug_ms', 0))]
  63         if kwargs.pop('read_only', False):
  64             cmd += ['--rgw-cache-enabled=false']
  65         return bash(cmd, **kwargs)
  66
  67     def start(self):
  68         cmd = [mstart_path + 'mstart.sh', self.cluster_id]
  69         env = None
  70         if self.needs_reset:
  71             env = os.environ.copy()
  72             env['CEPH_NUM_MDS'] = '0'
  73             cmd += ['-n']
  74             # cmd += ['-o']
  75             # cmd += ['rgw_cache_enabled=false']
  76         bash(cmd, env=env)
  77         self.needs_reset = False
  78
  79     def stop(self):
  80         cmd = [mstart_path + 'mstop.sh', self.cluster_id]
  81         bash(cmd)
  82
  83 class Gateway(multisite.Gateway):
  84     """ gateway implementation based on mrgw/mstop scripts """
  85     def __init__(self, client_id = None, *args, **kwargs):
  86         super(Gateway, self).__init__(*args, **kwargs)
  87         self.id = client_id
  88
  89     def start(self, args = None):
  90         """ start the gateway """
  91         assert(self.cluster)
  92         env = os.environ.copy()
  93         # to change frontend, set RGW_FRONTEND env variable
  94         # e.g. RGW_FRONTEND=civetweb
  95         # to run test under valgrind memcheck, set RGW_VALGRIND to 'yes'
  96         # e.g. RGW_VALGRIND=yes
  97         cmd = [mstart_path + 'mrgw.sh', self.cluster.cluster_id, str(self.port), str(self.ssl_port)]
  98         if self.id:
  99             cmd += ['-i', self.id]
 100         cmd += ['--debug-rgw=20', '--debug-ms=1']
 101         if args:
 102             cmd += args
 103         bash(cmd, env=env)
 104
 105     def stop(self):
 106         """ stop the gateway """
 107         assert(self.cluster)
 108         cmd = [mstart_path + 'mstop.sh', self.cluster.cluster_id, 'radosgw', str(self.port)]
 109         bash(cmd)
 110
 111 def gen_access_key():
 112     return ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(16))
 113
 114 def gen_secret():
 115     return ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for _ in range(32))
 116
 117 def gen_credentials():
 118     return multisite.Credentials(gen_access_key(), gen_secret())
 119
 120 def cluster_name(cluster_num):
 121     return 'c' + str(cluster_num)
 122
 123 def zonegroup_name(zonegroup_num):
 124     return string.ascii_lowercase[zonegroup_num]
 125
 126 def zone_name(zonegroup_num, zone_num):
 127     return zonegroup_name(zonegroup_num) + str(zone_num + 1)
 128
 129 def gateway_port(zonegroup_num, gateway_num):
 130     return 8000 + 100 * zonegroup_num + gateway_num
 131
 132 def gateway_name(zonegroup_num, zone_num, gateway_num):
 133     return zone_name(zonegroup_num, zone_num) + '-' + str(gateway_num + 1)
 134
 135 def zone_endpoints(zonegroup_num, zone_num, gateways_per_zone):
 136     endpoints = []
 137     base = gateway_port(zonegroup_num, zone_num * gateways_per_zone)
 138     for i in range(0, gateways_per_zone):
 139         endpoints.append('http://localhost:' + str(base + i))
 140     return endpoints
 141
 142 def get_log_level(log_level):
 143     if log_level >= 20:
 144         return logging.DEBUG
 145     if log_level >= 10:
 146         return logging.INFO
 147     if log_level >= 5:
 148         return logging.WARN
 149     if log_level >= 1:
 150         return logging.ERROR
 151     return logging.CRITICAL
 152
 153 def setup_logging(log_level_console, log_file, log_level_file):
 154     if log_file:
 155         formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
 156         fh = logging.FileHandler(log_file)
 157         fh.setFormatter(formatter)
 158         fh.setLevel(get_log_level(log_level_file))
 159         log.addHandler(fh)
 160
 161     formatter = logging.Formatter('%(levelname)s %(message)s')
 162     ch = logging.StreamHandler()
 163     ch.setFormatter(formatter)
 164     ch.setLevel(get_log_level(log_level_console))
 165     log.addHandler(ch)
 166     log.setLevel(get_log_level(log_level_console))
 167
 168 def init(parse_args):
 169     cfg = configparser.RawConfigParser({
 170                                          'num_zonegroups': 1,
 171                                          'num_zones': 3,
 172                                          'num_az_zones': 0,
 173                                          'gateways_per_zone': 2,
 174                                          'no_bootstrap': 'false',
 175                                          'log_level': 20,
 176                                          'log_file': None,
 177                                          'file_log_level': 20,
 178                                          'tenant': None,
 179                                          'checkpoint_retries': 60,
 180                                          'checkpoint_delay': 5,
 181                                          'reconfigure_delay': 5,
 182                                          'use_ssl': 'false',
 183                                          })
 184     try:
 185         path = os.environ['RGW_MULTI_TEST_CONF']
 186     except KeyError:
 187         path = test_path + 'test_multi.conf'
 188
 189     try:
 190         with open(path) as f:
 191             cfg.readfp(f)
 192     except:
 193         print('WARNING: error reading test config. Path can be set through the RGW_MULTI_TEST_CONF env variable')
 194         pass
 195
 196     parser = argparse.ArgumentParser(
 197             description='Run rgw multi-site tests',
 198             usage='test_multi [--num-zonegroups <num>] [--num-zones <num>] [--no-bootstrap]')
 199
 200     section = 'DEFAULT'
 201     parser.add_argument('--num-zonegroups', type=int, default=cfg.getint(section, 'num_zonegroups'))
 202     parser.add_argument('--num-zones', type=int, default=cfg.getint(section, 'num_zones'))
 203     parser.add_argument('--gateways-per-zone', type=int, default=cfg.getint(section, 'gateways_per_zone'))
 204     parser.add_argument('--no-bootstrap', action='store_true', default=cfg.getboolean(section, 'no_bootstrap'))
 205     parser.add_argument('--log-level', type=int, default=cfg.getint(section, 'log_level'))
 206     parser.add_argument('--log-file', type=str, default=cfg.get(section, 'log_file'))
 207     parser.add_argument('--file-log-level', type=int, default=cfg.getint(section, 'file_log_level'))
 208     parser.add_argument('--tenant', type=str, default=cfg.get(section, 'tenant'))
 209     parser.add_argument('--checkpoint-retries', type=int, default=cfg.getint(section, 'checkpoint_retries'))
 210     parser.add_argument('--checkpoint-delay', type=int, default=cfg.getint(section, 'checkpoint_delay'))
 211     parser.add_argument('--reconfigure-delay', type=int, default=cfg.getint(section, 'reconfigure_delay'))
 212     parser.add_argument('--use-ssl', type=bool, default=cfg.getboolean(section, 'use_ssl'))
 213
 214
 215     es_cfg = []
 216     cloud_cfg = []
 217     az_cfg = []
 218
 219     for s in cfg.sections():
 220         if s.startswith('elasticsearch'):
 221             es_cfg.append(ESZoneConfig(cfg, s))
 222         elif s.startswith('cloud'):
 223             cloud_cfg.append(CloudZoneConfig(cfg, s))
 224         elif s.startswith('archive'):
 225             az_cfg.append(AZoneConfig(cfg, s))
 226
 227
 228     argv = []
 229
 230     if parse_args:
 231         argv = sys.argv[1:]
 232
 233     args = parser.parse_args(argv)
 234     bootstrap = not args.no_bootstrap
 235
 236     setup_logging(args.log_level, args.log_file, args.file_log_level)
 237
 238     # start first cluster
 239     c1 = Cluster(cluster_name(1))
 240     if bootstrap:
 241         c1.start()
 242     clusters = []
 243     clusters.append(c1)
 244
 245     admin_creds = gen_credentials()
 246     admin_user = multisite.User('zone.user')
 247
 248     user_creds = gen_credentials()
 249     user = multisite.User('tester', tenant=args.tenant)
 250
 251     realm = multisite.Realm('r')
 252     if bootstrap:
 253         # create the realm on c1
 254         realm.create(c1)
 255     else:
 256         realm.get(c1)
 257     period = multisite.Period(realm=realm)
 258     realm.current_period = period
 259
 260     num_es_zones = len(es_cfg)
 261     num_cloud_zones = len(cloud_cfg)
 262     num_az_zones = cfg.getint(section, 'num_az_zones')
 263
 264     num_zones = args.num_zones + num_es_zones + num_cloud_zones + num_az_zones
 265
 266     use_ssl = cfg.getboolean(section, 'use_ssl')
 267
 268     if use_ssl and bootstrap:
 269         cmd = ['openssl', 'req',
 270                 '-x509',
 271                 '-newkey', 'rsa:4096',
 272                 '-sha256',
 273                 '-nodes',
 274                 '-keyout', 'key.pem',
 275                 '-out', 'cert.pem',
 276                 '-subj', '/CN=localhost',
 277                 '-days', '3650']
 278         bash(cmd)
 279         # append key to cert
 280         fkey = open('./key.pem', 'r')
 281         if fkey.mode == 'r':
 282             fcert = open('./cert.pem', 'a')
 283             fcert.write(fkey.read())
 284             fcert.close()
 285         fkey.close()
 286
 287     for zg in range(0, args.num_zonegroups):
 288         zonegroup = multisite.ZoneGroup(zonegroup_name(zg), period)
 289         period.zonegroups.append(zonegroup)
 290
 291         is_master_zg = zg == 0
 292         if is_master_zg:
 293             period.master_zonegroup = zonegroup
 294
 295         for z in range(0, num_zones):
 296             is_master = z == 0
 297             # start a cluster, or use c1 for first zone
 298             cluster = None
 299             if is_master_zg and is_master:
 300                 cluster = c1
 301             else:
 302                 cluster = Cluster(cluster_name(len(clusters) + 1))
 303                 clusters.append(cluster)
 304                 if bootstrap:
 305                     cluster.start()
 306                     # pull realm configuration from the master's gateway
 307                     gateway = realm.meta_master_zone().gateways[0]
 308                     realm.pull(cluster, gateway, admin_creds)
 309
 310             endpoints = zone_endpoints(zg, z, args.gateways_per_zone)
 311             if is_master:
 312                 if bootstrap:
 313                     # create the zonegroup on its first zone's cluster
 314                     arg = []
 315                     if is_master_zg:
 316                         arg += ['--master']
 317                     if len(endpoints): # use master zone's endpoints
 318                         arg += ['--endpoints', ','.join(endpoints)]
 319                     zonegroup.create(cluster, arg)
 320                 else:
 321                     zonegroup.get(cluster)
 322
 323             es_zone = (z >= args.num_zones and z < args.num_zones + num_es_zones)
 324             cloud_zone = (z >= args.num_zones + num_es_zones and z < args.num_zones + num_es_zones + num_cloud_zones)
 325             az_zone = (z >= args.num_zones + num_es_zones + num_cloud_zones)
 326
 327             # create the zone in its zonegroup
 328             zone = multisite.Zone(zone_name(zg, z), zonegroup, cluster)
 329             if es_zone:
 330                 zone_index = z - args.num_zones
 331                 zone = ESZone(zone_name(zg, z), es_cfg[zone_index].endpoint, zonegroup, cluster)
 332             elif cloud_zone:
 333                 zone_index = z - args.num_zones - num_es_zones
 334                 ccfg = cloud_cfg[zone_index]
 335                 zone = CloudZone(zone_name(zg, z), ccfg.endpoint, ccfg.credentials, ccfg.source_bucket,
 336                                  ccfg.target_path, zonegroup, cluster)
 337             elif az_zone:
 338                 zone_index = z - args.num_zones - num_es_zones - num_cloud_zones
 339                 zone = AZone(zone_name(zg, z), zonegroup, cluster)
 340             else:
 341                 zone = RadosZone(zone_name(zg, z), zonegroup, cluster)
 342
 343             if bootstrap:
 344                 arg = admin_creds.credential_args()
 345                 if is_master:
 346                     arg += ['--master']
 347                 if len(endpoints):
 348                     arg += ['--endpoints', ','.join(endpoints)]
 349                 zone.create(cluster, arg)
 350             else:
 351                 zone.get(cluster)
 352             zonegroup.zones.append(zone)
 353             if is_master:
 354                 zonegroup.master_zone = zone
 355
 356             zonegroup.zones_by_type.setdefault(zone.tier_type(), []).append(zone)
 357
 358             if zone.is_read_only():
 359                 zonegroup.ro_zones.append(zone)
 360             else:
 361                 zonegroup.rw_zones.append(zone)
 362
 363             # update/commit the period
 364             if bootstrap:
 365                 period.update(zone, commit=True)
 366
 367             ssl_port_offset = 1000
 368             # start the gateways
 369             for g in range(0, args.gateways_per_zone):
 370                 port = gateway_port(zg, g + z * args.gateways_per_zone)
 371                 client_id = gateway_name(zg, z, g)
 372                 gateway = Gateway(client_id, 'localhost', port, cluster, zone,
 373                         ssl_port = port+ssl_port_offset if use_ssl else 0)
 374                 if bootstrap:
 375                     gateway.start()
 376                 zone.gateways.append(gateway)
 377
 378             if is_master_zg and is_master:
 379                 if bootstrap:
 380                     # create admin user
 381                     arg = ['--display-name', '"Zone User"', '--system']
 382                     arg += admin_creds.credential_args()
 383                     admin_user.create(zone, arg)
 384                     # create test user
 385                     arg = ['--display-name', '"Test User"', '--caps', 'roles=*']
 386                     arg += user_creds.credential_args()
 387                     user.create(zone, arg)
 388                 else:
 389                     # read users and update keys
 390                     admin_user.info(zone)
 391                     admin_creds = admin_user.credentials[0]
 392                     arg = []
 393                     user.info(zone, arg)
 394                     user_creds = user.credentials[0]
 395
 396     if not bootstrap:
 397         period.get(c1)
 398
 399     config = Config(checkpoint_retries=args.checkpoint_retries,
 400                     checkpoint_delay=args.checkpoint_delay,
 401                     reconfigure_delay=args.reconfigure_delay,
 402                     tenant=args.tenant)
 403     init_multi(realm, user, config)
 404
 405 def setup_module():
 406     init(False)
 407
 408 if __name__ == "__main__":
 409     init(True)
 410