ceph/src/test/rgw/rgw_multi/zone_es.py

   1 import json
   2 import requests.compat
   3 import logging
   4
   5 import boto
   6 import boto.s3.connection
   7
   8 import dateutil.parser
   9
  10 from nose.tools import eq_ as eq
  11 try:
  12     from itertools import izip_longest as zip_longest  # type: ignore
  13 except ImportError:
  14     from itertools import zip_longest
  15
  16 from .multisite import *
  17 from .tools import *
  18
  19 log = logging.getLogger(__name__)
  20
  21 def get_key_ver(k):
  22     if not k.version_id:
  23         return 'null'
  24     return k.version_id
  25
  26 def check_object_eq(k1, k2, check_extra = True):
  27     assert k1
  28     assert k2
  29     log.debug('comparing key name=%s', k1.name)
  30     eq(k1.name, k2.name)
  31     eq(k1.metadata, k2.metadata)
  32     # eq(k1.cache_control, k2.cache_control)
  33     eq(k1.content_type, k2.content_type)
  34     # eq(k1.content_encoding, k2.content_encoding)
  35     # eq(k1.content_disposition, k2.content_disposition)
  36     # eq(k1.content_language, k2.content_language)
  37     eq(k1.etag, k2.etag)
  38     mtime1 = dateutil.parser.parse(k1.last_modified)
  39     mtime2 = dateutil.parser.parse(k2.last_modified)
  40     assert abs((mtime1 - mtime2).total_seconds()) < 1 # handle different time resolution
  41     if check_extra:
  42         eq(k1.owner.id, k2.owner.id)
  43         eq(k1.owner.display_name, k2.owner.display_name)
  44     # eq(k1.storage_class, k2.storage_class)
  45     eq(k1.size, k2.size)
  46     eq(get_key_ver(k1), get_key_ver(k2))
  47     # eq(k1.encrypted, k2.encrypted)
  48
  49 def make_request(conn, method, bucket, key, query_args, headers):
  50     result = conn.make_request(method, bucket=bucket, key=key, query_args=query_args, headers=headers)
  51     if result.status / 100 != 2:
  52         raise boto.exception.S3ResponseError(result.status, result.reason, result.read())
  53     return result
  54
  55
  56 class MDSearch:
  57     def __init__(self, conn, bucket_name, query, query_args = None, marker = None):
  58         self.conn = conn
  59         self.bucket_name = bucket_name or ''
  60         if bucket_name:
  61             self.bucket = boto.s3.bucket.Bucket(name=bucket_name)
  62         else:
  63             self.bucket = None
  64         self.query = query
  65         self.query_args = query_args
  66         self.max_keys = None
  67         self.marker = marker
  68
  69     def raw_search(self):
  70         q = self.query or ''
  71         query_args = append_query_arg(self.query_args, 'query', requests.compat.quote_plus(q))
  72         if self.max_keys is not None:
  73             query_args = append_query_arg(query_args, 'max-keys', self.max_keys)
  74         if self.marker:
  75             query_args = append_query_arg(query_args, 'marker', self.marker)
  76
  77         query_args = append_query_arg(query_args, 'format', 'json')
  78
  79         headers = {}
  80
  81         result = make_request(self.conn, "GET", bucket=self.bucket_name, key='', query_args=query_args, headers=headers)
  82
  83         l = []
  84
  85         result_dict = json.loads(result.read())
  86
  87         for entry in result_dict['Objects']:
  88             bucket = self.conn.get_bucket(entry['Bucket'], validate = False)
  89             k = boto.s3.key.Key(bucket, entry['Key'])
  90
  91             k.version_id = entry['Instance']
  92             k.etag = entry['ETag']
  93             k.owner = boto.s3.user.User(id=entry['Owner']['ID'], display_name=entry['Owner']['DisplayName'])
  94             k.last_modified = entry['LastModified']
  95             k.size = entry['Size']
  96             k.content_type = entry['ContentType']
  97             k.versioned_epoch = entry['VersionedEpoch']
  98
  99             k.metadata = {}
 100             for e in entry['CustomMetadata']:
 101                 k.metadata[e['Name']] = str(e['Value']) # int values will return as int, cast to string for compatibility with object meta response
 102
 103             l.append(k)
 104
 105         return result_dict, l
 106
 107     def search(self, drain = True, sort = True, sort_key = None):
 108         l = []
 109
 110         is_done = False
 111
 112         while not is_done:
 113             result, result_keys = self.raw_search()
 114
 115             l = l + result_keys
 116
 117             is_done = not (drain and (result['IsTruncated'] == "true"))
 118             marker = result['Marker']
 119
 120         if sort:
 121             if not sort_key:
 122                 sort_key = lambda k: (k.name, -k.versioned_epoch)
 123             l.sort(key = sort_key)
 124
 125         return l
 126
 127
 128 class MDSearchConfig:
 129     def __init__(self, conn, bucket_name):
 130         self.conn = conn
 131         self.bucket_name = bucket_name or ''
 132         if bucket_name:
 133             self.bucket = boto.s3.bucket.Bucket(name=bucket_name)
 134         else:
 135             self.bucket = None
 136
 137     def send_request(self, conf, method):
 138         query_args = 'mdsearch'
 139         headers = None
 140         if conf:
 141             headers = { 'X-Amz-Meta-Search': conf }
 142
 143         query_args = append_query_arg(query_args, 'format', 'json')
 144
 145         return make_request(self.conn, method, bucket=self.bucket_name, key='', query_args=query_args, headers=headers)
 146
 147     def get_config(self):
 148         result = self.send_request(None, 'GET')
 149         return json.loads(result.read())
 150
 151     def set_config(self, conf):
 152         self.send_request(conf, 'POST')
 153
 154     def del_config(self):
 155         self.send_request(None, 'DELETE')
 156
 157
 158 class ESZoneBucket:
 159     def __init__(self, zone_conn, name, conn):
 160         self.zone_conn = zone_conn
 161         self.name = name
 162         self.conn = conn
 163
 164         self.bucket = boto.s3.bucket.Bucket(name=name)
 165
 166     def get_all_versions(self):
 167
 168         marker = None
 169         is_done = False
 170
 171         req = MDSearch(self.conn, self.name, 'bucket == ' + self.name, marker=marker)
 172
 173         for k in req.search():
 174             yield k
 175
 176
 177
 178
 179 class ESZone(Zone):
 180     def __init__(self, name, es_endpoint, zonegroup = None, cluster = None, data = None, zone_id = None, gateways = None):
 181         self.es_endpoint = es_endpoint
 182         super(ESZone, self).__init__(name, zonegroup, cluster, data, zone_id, gateways)
 183
 184     def is_read_only(self):
 185         return True
 186
 187     def tier_type(self):
 188         return "elasticsearch"
 189
 190     def create(self, cluster, args = None, check_retcode = True):
 191         """ create the object with the given arguments """
 192
 193         if args is None:
 194             args = ''
 195
 196         tier_config = ','.join([ 'endpoint=' + self.es_endpoint, 'explicit_custom_meta=false' ])
 197
 198         args += [ '--tier-type', self.tier_type(), '--tier-config', tier_config ]
 199
 200         return self.json_command(cluster, 'create', args, check_retcode=check_retcode)
 201
 202     def has_buckets(self):
 203         return False
 204
 205     class Conn(ZoneConn):
 206         def __init__(self, zone, credentials):
 207             super(ESZone.Conn, self).__init__(zone, credentials)
 208
 209         def get_bucket(self, bucket_name):
 210             return ESZoneBucket(self, bucket_name, self.conn)
 211
 212         def create_bucket(self, name):
 213             # should not be here, a bug in the test suite
 214             log.critical('Conn.create_bucket() should not be called in ES zone')
 215             assert False
 216
 217         def check_bucket_eq(self, zone_conn, bucket_name):
 218             assert(zone_conn.zone.tier_type() == "rados")
 219
 220             log.info('comparing bucket=%s zones={%s, %s}', bucket_name, self.name, self.name)
 221             b1 = self.get_bucket(bucket_name)
 222             b2 = zone_conn.get_bucket(bucket_name)
 223
 224             log.debug('bucket1 objects:')
 225             for o in b1.get_all_versions():
 226                 log.debug('o=%s', o.name)
 227             log.debug('bucket2 objects:')
 228             for o in b2.get_all_versions():
 229                 log.debug('o=%s', o.name)
 230
 231             for k1, k2 in zip_longest(b1.get_all_versions(), b2.get_all_versions()):
 232                 if k1 is None:
 233                     log.critical('key=%s is missing from zone=%s', k2.name, self.name)
 234                     assert False
 235                 if k2 is None:
 236                     log.critical('key=%s is missing from zone=%s', k1.name, zone_conn.name)
 237                     assert False
 238
 239                 check_object_eq(k1, k2)
 240
 241
 242             log.info('success, bucket identical: bucket=%s zones={%s, %s}', bucket_name, self.name, zone_conn.name)
 243
 244             return True
 245
 246     def get_conn(self, credentials):
 247         return self.Conn(self, credentials)
 248
 249
 250 class ESZoneConfig:
 251     def __init__(self, cfg, section):
 252         self.endpoint = cfg.get(section, 'endpoint')
 253