code.grnet.gr Git - archipelago/blob - tools/archipelago/archipelago/vlmc.py

   1 #!/usr/bin/env python
   2
   3 # Copyright 2012 GRNET S.A. All rights reserved.
   4 #
   5 # Redistribution and use in source and binary forms, with or
   6 # without modification, are permitted provided that the following
   7 # conditions are met:
   8 #
   9 #   1. Redistributions of source code must retain the above
  10 #      copyright notice, this list of conditions and the following
  11 #      disclaimer.
  12 #
  13 #   2. Redistributions in binary form must reproduce the above
  14 #      copyright notice, this list of conditions and the following
  15 #      disclaimer in the documentation and/or other materials
  16 #      provided with the distribution.
  17 #
  18 # THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  19 # OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  20 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  21 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  22 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  23 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  24 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  25 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  26 # AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  27 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  28 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  29 # POSSIBILITY OF SUCH DAMAGE.
  30 #
  31 # The views and conclusions contained in the software and
  32 # documentation are those of the authors and should not be
  33 # interpreted as representing official policies, either expressed
  34 # or implied, of GRNET S.A.
  35 #
  36
  37
  38 import os
  39 import sys
  40 from struct import unpack
  41 from binascii import hexlify
  42 from ctypes import c_uint32, c_uint64
  43
  44 from .common import *
  45
  46
  47 @exclusive()
  48 def get_mapped():
  49     try:
  50         devices = os.listdir(os.path.join(XSEGBD_SYSFS, "devices/"))
  51     except:
  52         if loaded_module(xsegbd):
  53             raise Error("Cannot list %s/devices/" % XSEGBD_SYSFS)
  54         else:
  55             return None
  56     try:
  57         mapped = []
  58         for f in devices:
  59             d_id = open(XSEGBD_SYSFS + "devices/" + f + "/id")
  60             d_id = d_id.read().strip()
  61             target = open(XSEGBD_SYSFS + "devices/" + f + "/target")
  62             target = target.read().strip()
  63             mapped.append((d_id, target))
  64
  65     except Exception, reason:
  66         raise Error(reason)
  67
  68     return mapped
  69
  70
  71 def showmapped():
  72     mapped = get_mapped()
  73     if not mapped:
  74         print "No volumes mapped"
  75         print ""
  76         return 0
  77
  78     print "id\timage\t\tdevice"
  79     for m in mapped:
  80         print "%s\t%s\t%s" % (m[0], m[1], DEVICE_PREFIX + m[0])
  81
  82     return len(mapped)
  83
  84
  85 def showmapped_wrapper(**kwargs):
  86     showmapped()
  87
  88
  89 def is_mapped(volume):
  90     mapped = get_mapped()
  91     if not mapped:
  92         return None
  93
  94     for m in mapped:
  95         d_id = m[0]
  96         target = m[1]
  97         if target == volume:
  98             return d_id
  99     return None
 100
 101
 102 def create(name, size=None, snap=None, cont_addr=False, **kwargs):
 103     if len(name) < 6:
 104         raise Error("Name should have at least len 6")
 105     if size is None and snap is None:
 106         raise Error("At least one of the size/snap args must be provided")
 107
 108     if not snap:
 109         snap = ""
 110     if not size:
 111         size = 0
 112     else:
 113         size = size << 20
 114
 115     ret = False
 116     xseg_ctx = Xseg_ctx(get_segment())
 117     mport = peers['mapperd'].portno_start
 118     req = Request.get_clone_request(xseg_ctx, mport, snap, clone=name,
 119             clone_size=size, cont_addr=cont_addr)
 120     req.submit()
 121     req.wait()
 122     ret = req.success()
 123     req.put()
 124     xseg_ctx.shutdown()
 125     if not ret:
 126         raise Error("vlmc creation failed")
 127
 128
 129 def snapshot(name, snap_name=None, cli=False, **kwargs):
 130     if len(name) < 6:
 131         raise Error("Name should have at least len 6")
 132
 133     xseg_ctx = Xseg_ctx(get_segment())
 134     vport = peers['vlmcd'].portno_start
 135     req = Request.get_snapshot_request(xseg_ctx, vport, name, snap=snap_name)
 136     req.submit()
 137     req.wait()
 138     ret = req.success()
 139     req.put()
 140     xseg_ctx.shutdown()
 141
 142     if not ret:
 143         raise Error("vlmc snapshot failed")
 144     if cli:
 145         sys.stdout.write("Snapshot name: %s\n" % snap_name)
 146
 147 def hash(name, cli=False, **kwargs):
 148     if len(name) < 6:
 149         raise Error("Name should have at least len 6")
 150
 151     xseg_ctx = Xseg_ctx(get_segment())
 152     mport = peers['mapperd'].portno_start
 153     req = Request.get_hash_request(xseg_ctx, mport, name)
 154     req.submit()
 155     req.wait()
 156     ret = req.success()
 157     if ret:
 158         xhash = req.get_data(xseg_reply_hash).contents
 159         hash_name = ctypes.string_at(xhash.target, xhash.targetlen)
 160     req.put()
 161     xseg_ctx.shutdown()
 162
 163     if not ret:
 164         raise Error("vlmc hash failed")
 165     if cli:
 166         sys.stdout.write("Hash name: %s\n" % hash_name)
 167         return hash_name
 168
 169 def list_volumes(**kwargs):
 170     if isinstance(peers['blockerm'], Sosd):
 171         import rados
 172         cluster = rados.Rados(conffile=config['CEPH_CONF_FILE'])
 173         cluster.connect()
 174         ioctx = cluster.open_ioctx(peers['blockerm'].pool)
 175         oi = rados.ObjectIterator(ioctx)
 176         for o in oi:
 177             name = o.key
 178             if name.startswith(ARCHIP_PREFIX) and not name.endswith('_lock'):
 179                 print name[len(ARCHIP_PREFIX):]
 180     elif config['STORAGE'] == "files":
 181         raise Error("Vlmc list not supported for files yet")
 182     else:
 183         raise Error("Invalid storage")
 184
 185
 186 def remove(name, **kwargs):
 187     device = is_mapped(name)
 188     if device is not None:
 189         raise Error("Volume %s mapped on device %s%s" % (name, DEVICE_PREFIX,
 190                     device))
 191
 192     ret = False
 193     xseg_ctx = Xseg_ctx(get_segment())
 194     mport = peers['mapperd'].portno_start
 195     req = Request.get_delete_request(xseg_ctx, mport, name)
 196     req.submit()
 197     req.wait()
 198     ret = req.success()
 199     req.put()
 200     xseg_ctx.shutdown()
 201     if not ret:
 202         raise Error("vlmc removal failed")
 203
 204
 205 @exclusive()
 206 def map_volume(name, **kwargs):
 207     vport = peers['vlmcd'].portno_start
 208     if not loaded_module(xsegbd):
 209         raise Error("Xsegbd module not loaded")
 210
 211     device = is_mapped(name)
 212     if device is not None:
 213         raise Error("Volume %s already mapped on device %s%s" % (name,
 214             DEVICE_PREFIX, device))
 215
 216     prev = config['XSEGBD_START']
 217     try:
 218         result = [int(open(XSEGBD_SYSFS + "devices/" + f + "/srcport").read().
 219                   strip()) for f in os.listdir(XSEGBD_SYSFS + "devices/")]
 220         result.sort()
 221
 222         for p in result:
 223             if p - prev > 1:
 224                 break
 225             else:
 226                 prev = p
 227
 228         port = prev + 1
 229         if port > config['XSEGBD_END']:
 230             raise Error("Max xsegbd devices reached")
 231         fd = os.open(XSEGBD_SYSFS + "add", os.O_WRONLY)
 232         print >> sys.stderr, "write to %s : %s %d:%d:%d" % (XSEGBD_SYSFS +
 233                              "add", name, port, port - config['XSEGBD_START'] +
 234                              vport, REQS)
 235         os.write(fd, "%s %d:%d:%d" % (name, port, port - config['XSEGBD_START']
 236                                       + vport, REQS))
 237         os.close(fd)
 238         return port
 239     except Exception, reason:
 240         raise Error(name + ': ' + str(reason))
 241
 242
 243 @exclusive()
 244 def unmap_volume(name, **kwargs):
 245     if not loaded_module(xsegbd):
 246         raise Error("Xsegbd module not loaded")
 247     device = name
 248     try:
 249         for f in os.listdir(XSEGBD_SYSFS + "devices/"):
 250             d_id = open(XSEGBD_SYSFS + "devices/" + f + "/id")
 251             d_id = d_id.read().strip()
 252             target = open(XSEGBD_SYSFS + "devices/" + f + "/target")
 253             target = target.read().strip()
 254             if device == DEVICE_PREFIX + d_id:
 255                 fd = os.open(XSEGBD_SYSFS + "remove", os.O_WRONLY)
 256                 os.write(fd, d_id)
 257                 os.close(fd)
 258                 return
 259         raise Error("Device %s doesn't exist" % device)
 260     except Exception, reason:
 261         raise Error(device + ': ' + str(reason))
 262
 263
 264 # FIXME:
 265 def resize(name, size, **kwargs):
 266     if not loaded_module(xsegbd):
 267         raise Error("Xsegbd module not loaded")
 268
 269     try:
 270
 271         for f in os.listdir(XSEGBD_SYSFS + "devices/"):
 272             d_id = open(XSEGBD_SYSFS + "devices/" + f + "/id")
 273             d_id = d_id.read().strip()
 274             target = open(XSEGBD_SYSFS + "devices/" + f + "/target")
 275             target = target.read().strip()
 276             if name == target:
 277                 fd = os.open(XSEGBD_SYSFS + "devices/" + d_id + "/refresh",
 278                              os.O_WRONLY)
 279                 os.write(fd, "1")
 280                 os.close(fd)
 281
 282     except Exception, reason:
 283         raise Error(name + ': ' + str(reason))
 284
 285
 286 def lock(name, cli=False, **kwargs):
 287     if len(name) < 6:
 288         raise Error("Name should have at least len 6")
 289
 290     name = ARCHIP_PREFIX + name
 291
 292     xseg_ctx = Xseg_ctx(get_segment())
 293     mbport = peers['blockerm'].portno_start
 294     req = Request.get_acquire_request(xseg_ctx, mbport, name)
 295     req.submit()
 296     req.wait()
 297     ret = req.success()
 298     xseg_ctx.shutdown()
 299     if not ret:
 300         raise Error("vlmc lock failed")
 301     if cli:
 302         sys.stdout.write("Volume locked\n")
 303
 304
 305 def unlock(name, force=False, cli=False, **kwargs):
 306     if len(name) < 6:
 307         raise Error("Name should have at least len 6")
 308
 309     name = ARCHIP_PREFIX + name
 310
 311     xseg_ctx = Xseg_ctx(get_segment())
 312     mbport = peers['blockerm'].portno_start
 313     req = Request.get_release_request(xseg_ctx, mbport, name, force=force)
 314     req.submit()
 315     req.wait()
 316     ret = req.success()
 317     xseg_ctx.shutdown()
 318     if not ret:
 319         raise Error("vlmc unlock failed")
 320     if cli:
 321         sys.stdout.write("Volume unlocked\n")
 322
 323
 324 def open_volume(name, cli=False, **kwargs):
 325     if len(name) < 6:
 326         raise Error("Name should have at least len 6")
 327
 328     ret = False
 329     xseg_ctx = Xseg_ctx(get_segment())
 330     vport = peers['vlmcd'].portno_start
 331     req = Request.get_open_request(xseg_ctx, vport, name)
 332     req.submit()
 333     req.wait()
 334     ret = req.success()
 335     xseg_ctx.shutdown()
 336     if not ret:
 337         raise Error("vlmc open failed")
 338     if cli:
 339         sys.stdout.write("Volume opened\n")
 340
 341
 342 def close_volume(name, cli=False, **kwargs):
 343     if len(name) < 6:
 344         raise Error("Name should have at least len 6")
 345
 346     ret = False
 347     xseg_ctx = Xseg_ctx(get_segment())
 348     vport = peers['vlmcd'].portno_start
 349     req = Request.get_close_request(xseg_ctx, vport, name)
 350     req.submit()
 351     req.wait()
 352     ret = req.success()
 353     xseg_ctx.shutdown()
 354     if not ret:
 355         raise Error("vlmc close failed")
 356     if cli:
 357         sys.stdout.write("Volume closed\n")
 358
 359
 360 def info(name, cli=False, **kwargs):
 361     if len(name) < 6:
 362         raise Error("Name should have at least len 6")
 363
 364     ret = False
 365     xseg_ctx = Xseg_ctx(get_segment())
 366     mport = peers['mapperd'].portno_start
 367     req = Request.get_info_request(xseg_ctx, mport, name)
 368     req.submit()
 369     req.wait()
 370     ret = req.success()
 371     if ret:
 372         size = req.get_data(xseg_reply_info).contents.size
 373     xseg_ctx.shutdown()
 374     if not ret:
 375         raise Error("vlmc info failed")
 376     if cli:
 377         sys.stdout.write("Volume %s: size: %d\n" % (name, size))
 378
 379
 380 def mapinfo(name, verbose=False, **kwargs):
 381     if len(name) < 6:
 382         raise Error("Name should have at least len 6")
 383
 384     if config['STORAGE'] == "rados":
 385         import rados
 386         cluster = rados.Rados(conffile=config['CEPH_CONF_FILE'])
 387         cluster.connect()
 388         ioctx = cluster.open_ioctx(config['RADOS_POOL_MAPS'])
 389         BLOCKSIZE = 4*1024*1024
 390         try:
 391             mapdata = ioctx.read(ARCHIP_PREFIX + name, length=BLOCKSIZE)
 392         except Exception:
 393             raise Error("Cannot read map data")
 394         if not mapdata:
 395             raise Error("Cannot read map data")
 396         pos = 0
 397         size_uint32t = sizeof(c_uint32)
 398         version = unpack("<L", mapdata[pos:pos+size_uint32t])[0]
 399         pos += size_uint32t
 400         size_uint64t = sizeof(c_uint64)
 401         size = unpack("Q", mapdata[pos:pos+size_uint64t])[0]
 402         pos += size_uint64t
 403         blocks = size / BLOCKSIZE
 404         nr_exists = 0
 405         print ""
 406         print "Volume: " + name
 407         print "Version: " + str(version)
 408         print "Size: " + str(size)
 409         for i in range(blocks):
 410             exists = bool(unpack("B", mapdata[pos:pos+1])[0])
 411             if exists:
 412                 nr_exists += 1
 413             pos += 1
 414             block = hexlify(mapdata[pos:pos+32])
 415             pos += 32
 416             if verbose:
 417                 print block, exists
 418         print "Actual disk usage: " + str(nr_exists * BLOCKSIZE),
 419         print '(' + str(nr_exists) + '/' + str(blocks) + ' blocks)'
 420
 421     elif STORAGE == "files":
 422         raise Error("Mapinfo for file storage not supported")
 423     else:
 424         raise Error("Invalid storage")