src/allmydata/storage/server.py

   1 import os, re, weakref, struct, time
   2
   3 from foolscap import Referenceable
   4 from twisted.application import service
   5
   6 from zope.interface import implements
   7 from allmydata.interfaces import RIStorageServer, IStatsProducer
   8 from allmydata.util import base32, fileutil, log, time_format
   9 import allmydata # for __full_version__
  10
  11 from allmydata.storage.common import si_b2a, si_a2b, storage_index_to_dir
  12 _pyflakes_hush = [si_b2a, si_a2b, storage_index_to_dir] # re-exported
  13 from allmydata.storage.lease import LeaseInfo
  14 from allmydata.storage.mutable import MutableShareFile, EmptyShare, \
  15      create_mutable_sharefile
  16 from allmydata.storage.immutable import ShareFile, BucketWriter, BucketReader
  17 from allmydata.storage.crawler import BucketCountingCrawler
  18
  19 # storage/
  20 # storage/shares/incoming
  21 #   incoming/ holds temp dirs named $START/$STORAGEINDEX/$SHARENUM which will
  22 #   be moved to storage/shares/$START/$STORAGEINDEX/$SHARENUM upon success
  23 # storage/shares/$START/$STORAGEINDEX
  24 # storage/shares/$START/$STORAGEINDEX/$SHARENUM
  25
  26 # Where "$START" denotes the first 10 bits worth of $STORAGEINDEX (that's 2
  27 # base-32 chars).
  28
  29 # $SHARENUM matches this regex:
  30 NUM_RE=re.compile("^[0-9]+$")
  31
  32
  33
  34 class StorageServer(service.MultiService, Referenceable):
  35     implements(RIStorageServer, IStatsProducer)
  36     name = 'storage'
  37
  38     def __init__(self, storedir, nodeid, reserved_space=0,
  39                  discard_storage=False, readonly_storage=False,
  40                  stats_provider=None):
  41         service.MultiService.__init__(self)
  42         assert isinstance(nodeid, str)
  43         assert len(nodeid) == 20
  44         self.my_nodeid = nodeid
  45         self.storedir = storedir
  46         sharedir = os.path.join(storedir, "shares")
  47         fileutil.make_dirs(sharedir)
  48         self.sharedir = sharedir
  49         # we don't actually create the corruption-advisory dir until necessary
  50         self.corruption_advisory_dir = os.path.join(storedir,
  51                                                     "corruption-advisories")
  52         self.reserved_space = int(reserved_space)
  53         self.no_storage = discard_storage
  54         self.readonly_storage = readonly_storage
  55         self.stats_provider = stats_provider
  56         if self.stats_provider:
  57             self.stats_provider.register_producer(self)
  58         self.incomingdir = os.path.join(sharedir, 'incoming')
  59         self._clean_incomplete()
  60         fileutil.make_dirs(self.incomingdir)
  61         self._active_writers = weakref.WeakKeyDictionary()
  62         lp = log.msg("StorageServer created", facility="tahoe.storage")
  63
  64         if reserved_space:
  65             if self.get_available_space() is None:
  66                 log.msg("warning: [storage]reserved_space= is set, but this platform does not support statvfs(2), so this reservation cannot be honored",
  67                         umin="0wZ27w", level=log.UNUSUAL)
  68
  69         self.latencies = {"allocate": [], # immutable
  70                           "write": [],
  71                           "close": [],
  72                           "read": [],
  73                           "get": [],
  74                           "writev": [], # mutable
  75                           "readv": [],
  76                           "add-lease": [], # both
  77                           "renew": [],
  78                           "cancel": [],
  79                           }
  80
  81         statefile = os.path.join(storedir, "bucket_counter.state")
  82         self.bucket_counter = BucketCountingCrawler(self, statefile)
  83         self.bucket_counter.setServiceParent(self)
  84
  85     def count(self, name, delta=1):
  86         if self.stats_provider:
  87             self.stats_provider.count("storage_server." + name, delta)
  88
  89     def add_latency(self, category, latency):
  90         a = self.latencies[category]
  91         a.append(latency)
  92         if len(a) > 1000:
  93             self.latencies[category] = a[-1000:]
  94
  95     def get_latencies(self):
  96         """Return a dict, indexed by category, that contains a dict of
  97         latency numbers for each category. Each dict will contain the
  98         following keys: mean, 01_0_percentile, 10_0_percentile,
  99         50_0_percentile (median), 90_0_percentile, 95_0_percentile,
 100         99_0_percentile, 99_9_percentile. If no samples have been collected
 101         for the given category, then that category name will not be present
 102         in the return value."""
 103         # note that Amazon's Dynamo paper says they use 99.9% percentile.
 104         output = {}
 105         for category in self.latencies:
 106             if not self.latencies[category]:
 107                 continue
 108             stats = {}
 109             samples = self.latencies[category][:]
 110             samples.sort()
 111             count = len(samples)
 112             stats["mean"] = sum(samples) / count
 113             stats["01_0_percentile"] = samples[int(0.01 * count)]
 114             stats["10_0_percentile"] = samples[int(0.1 * count)]
 115             stats["50_0_percentile"] = samples[int(0.5 * count)]
 116             stats["90_0_percentile"] = samples[int(0.9 * count)]
 117             stats["95_0_percentile"] = samples[int(0.95 * count)]
 118             stats["99_0_percentile"] = samples[int(0.99 * count)]
 119             stats["99_9_percentile"] = samples[int(0.999 * count)]
 120             output[category] = stats
 121         return output
 122
 123     def log(self, *args, **kwargs):
 124         if "facility" not in kwargs:
 125             kwargs["facility"] = "tahoe.storage"
 126         return log.msg(*args, **kwargs)
 127
 128     def _clean_incomplete(self):
 129         fileutil.rm_dir(self.incomingdir)
 130
 131     def do_statvfs(self):
 132         return os.statvfs(self.storedir)
 133
 134     def get_stats(self):
 135         # remember: RIStatsProvider requires that our return dict
 136         # contains numeric values.
 137         stats = { 'storage_server.allocated': self.allocated_size(), }
 138         stats["storage_server.reserved_space"] = self.reserved_space
 139         for category,ld in self.get_latencies().items():
 140             for name,v in ld.items():
 141                 stats['storage_server.latencies.%s.%s' % (category, name)] = v
 142         writeable = True
 143         if self.readonly_storage:
 144             writeable = False
 145         try:
 146             s = self.do_statvfs()
 147             disk_total = s.f_bsize * s.f_blocks
 148             disk_used = s.f_bsize * (s.f_blocks - s.f_bfree)
 149             # spacetime predictors should look at the slope of disk_used.
 150             disk_free_for_root = s.f_bsize * s.f_bfree
 151             disk_free_for_nonroot = s.f_bsize * s.f_bavail
 152
 153             # include our local policy here: if we stop accepting shares when
 154             # the available space drops below 1GB, then include that fact in
 155             # disk_avail.
 156             disk_avail = disk_free_for_nonroot - self.reserved_space
 157             disk_avail = max(disk_avail, 0)
 158             if self.readonly_storage:
 159                 disk_avail = 0
 160             if disk_avail == 0:
 161                 writeable = False
 162
 163             # spacetime predictors should use disk_avail / (d(disk_used)/dt)
 164             stats["storage_server.disk_total"] = disk_total
 165             stats["storage_server.disk_used"] = disk_used
 166             stats["storage_server.disk_free_for_root"] = disk_free_for_root
 167             stats["storage_server.disk_free_for_nonroot"] = disk_free_for_nonroot
 168             stats["storage_server.disk_avail"] = disk_avail
 169         except AttributeError:
 170             # os.statvfs is available only on unix
 171             pass
 172         stats["storage_server.accepting_immutable_shares"] = int(writeable)
 173         return stats
 174
 175
 176     def stat_disk(self, d):
 177         s = os.statvfs(d)
 178         # s.f_bavail: available to non-root users
 179         disk_avail = s.f_bsize * s.f_bavail
 180         return disk_avail
 181
 182     def get_available_space(self):
 183         # returns None if it cannot be measured (windows)
 184         try:
 185             disk_avail = self.stat_disk(self.storedir)
 186             disk_avail -= self.reserved_space
 187         except AttributeError:
 188             disk_avail = None
 189         if self.readonly_storage:
 190             disk_avail = 0
 191         return disk_avail
 192
 193     def allocated_size(self):
 194         space = 0
 195         for bw in self._active_writers:
 196             space += bw.allocated_size()
 197         return space
 198
 199     def remote_get_version(self):
 200         remaining_space = self.get_available_space()
 201         if remaining_space is None:
 202             # we're on a platform that doesn't have 'df', so make a vague
 203             # guess.
 204             remaining_space = 2**64
 205         version = { "http://allmydata.org/tahoe/protocols/storage/v1" :
 206                     { "maximum-immutable-share-size": remaining_space,
 207                       "tolerates-immutable-read-overrun": True,
 208                       "delete-mutable-shares-with-zero-length-writev": True,
 209                       },
 210                     "application-version": str(allmydata.__full_version__),
 211                     }
 212         return version
 213
 214     def remote_allocate_buckets(self, storage_index,
 215                                 renew_secret, cancel_secret,
 216                                 sharenums, allocated_size,
 217                                 canary, owner_num=0):
 218         # owner_num is not for clients to set, but rather it should be
 219         # curried into the PersonalStorageServer instance that is dedicated
 220         # to a particular owner.
 221         start = time.time()
 222         self.count("allocate")
 223         alreadygot = set()
 224         bucketwriters = {} # k: shnum, v: BucketWriter
 225         si_dir = storage_index_to_dir(storage_index)
 226         si_s = si_b2a(storage_index)
 227
 228         log.msg("storage: allocate_buckets %s" % si_s)
 229
 230         # in this implementation, the lease information (including secrets)
 231         # goes into the share files themselves. It could also be put into a
 232         # separate database. Note that the lease should not be added until
 233         # the BucketWriter has been closed.
 234         expire_time = time.time() + 31*24*60*60
 235         lease_info = LeaseInfo(owner_num,
 236                                renew_secret, cancel_secret,
 237                                expire_time, self.my_nodeid)
 238
 239         max_space_per_bucket = allocated_size
 240
 241         remaining_space = self.get_available_space()
 242         limited = remaining_space is not None
 243         if limited:
 244             # this is a bit conservative, since some of this allocated_size()
 245             # has already been written to disk, where it will show up in
 246             # get_available_space.
 247             remaining_space -= self.allocated_size()
 248
 249         # fill alreadygot with all shares that we have, not just the ones
 250         # they asked about: this will save them a lot of work. Add or update
 251         # leases for all of them: if they want us to hold shares for this
 252         # file, they'll want us to hold leases for this file.
 253         for (shnum, fn) in self._get_bucket_shares(storage_index):
 254             alreadygot.add(shnum)
 255             sf = ShareFile(fn)
 256             sf.add_or_renew_lease(lease_info)
 257
 258         # self.readonly_storage causes remaining_space=0
 259
 260         for shnum in sharenums:
 261             incominghome = os.path.join(self.incomingdir, si_dir, "%d" % shnum)
 262             finalhome = os.path.join(self.sharedir, si_dir, "%d" % shnum)
 263             if os.path.exists(finalhome):
 264                 # great! we already have it. easy.
 265                 pass
 266             elif os.path.exists(incominghome):
 267                 # Note that we don't create BucketWriters for shnums that
 268                 # have a partial share (in incoming/), so if a second upload
 269                 # occurs while the first is still in progress, the second
 270                 # uploader will use different storage servers.
 271                 pass
 272             elif (not limited) or (remaining_space >= max_space_per_bucket):
 273                 # ok! we need to create the new share file.
 274                 bw = BucketWriter(self, incominghome, finalhome,
 275                                   max_space_per_bucket, lease_info, canary)
 276                 if self.no_storage:
 277                     bw.throw_out_all_data = True
 278                 bucketwriters[shnum] = bw
 279                 self._active_writers[bw] = 1
 280                 if limited:
 281                     remaining_space -= max_space_per_bucket
 282             else:
 283                 # bummer! not enough space to accept this bucket
 284                 pass
 285
 286         if bucketwriters:
 287             fileutil.make_dirs(os.path.join(self.sharedir, si_dir))
 288
 289         self.add_latency("allocate", time.time() - start)
 290         return alreadygot, bucketwriters
 291
 292     def _iter_share_files(self, storage_index):
 293         for shnum, filename in self._get_bucket_shares(storage_index):
 294             f = open(filename, 'rb')
 295             header = f.read(32)
 296             f.close()
 297             if header[:32] == MutableShareFile.MAGIC:
 298                 sf = MutableShareFile(filename, self)
 299                 # note: if the share has been migrated, the renew_lease()
 300                 # call will throw an exception, with information to help the
 301                 # client update the lease.
 302             elif header[:4] == struct.pack(">L", 1):
 303                 sf = ShareFile(filename)
 304             else:
 305                 continue # non-sharefile
 306             yield sf
 307
 308     def remote_add_lease(self, storage_index, renew_secret, cancel_secret,
 309                          owner_num=1):
 310         start = time.time()
 311         self.count("add-lease")
 312         new_expire_time = time.time() + 31*24*60*60
 313         lease_info = LeaseInfo(owner_num,
 314                                renew_secret, cancel_secret,
 315                                new_expire_time, self.my_nodeid)
 316         for sf in self._iter_share_files(storage_index):
 317             sf.add_or_renew_lease(lease_info)
 318         self.add_latency("add-lease", time.time() - start)
 319         return None
 320
 321     def remote_renew_lease(self, storage_index, renew_secret):
 322         start = time.time()
 323         self.count("renew")
 324         new_expire_time = time.time() + 31*24*60*60
 325         found_buckets = False
 326         for sf in self._iter_share_files(storage_index):
 327             found_buckets = True
 328             sf.renew_lease(renew_secret, new_expire_time)
 329         self.add_latency("renew", time.time() - start)
 330         if not found_buckets:
 331             raise IndexError("no such lease to renew")
 332
 333     def remote_cancel_lease(self, storage_index, cancel_secret):
 334         start = time.time()
 335         self.count("cancel")
 336
 337         total_space_freed = 0
 338         found_buckets = False
 339         for sf in self._iter_share_files(storage_index):
 340             # note: if we can't find a lease on one share, we won't bother
 341             # looking in the others. Unless something broke internally
 342             # (perhaps we ran out of disk space while adding a lease), the
 343             # leases on all shares will be identical.
 344             found_buckets = True
 345             # this raises IndexError if the lease wasn't present XXXX
 346             total_space_freed += sf.cancel_lease(cancel_secret)
 347
 348         if found_buckets:
 349             storagedir = os.path.join(self.sharedir,
 350                                       storage_index_to_dir(storage_index))
 351             if not os.listdir(storagedir):
 352                 os.rmdir(storagedir)
 353
 354         if self.stats_provider:
 355             self.stats_provider.count('storage_server.bytes_freed',
 356                                       total_space_freed)
 357         self.add_latency("cancel", time.time() - start)
 358         if not found_buckets:
 359             raise IndexError("no such storage index")
 360
 361     def bucket_writer_closed(self, bw, consumed_size):
 362         if self.stats_provider:
 363             self.stats_provider.count('storage_server.bytes_added', consumed_size)
 364         del self._active_writers[bw]
 365
 366     def _get_bucket_shares(self, storage_index):
 367         """Return a list of (shnum, pathname) tuples for files that hold
 368         shares for this storage_index. In each tuple, 'shnum' will always be
 369         the integer form of the last component of 'pathname'."""
 370         storagedir = os.path.join(self.sharedir, storage_index_to_dir(storage_index))
 371         try:
 372             for f in os.listdir(storagedir):
 373                 if NUM_RE.match(f):
 374                     filename = os.path.join(storagedir, f)
 375                     yield (int(f), filename)
 376         except OSError:
 377             # Commonly caused by there being no buckets at all.
 378             pass
 379
 380     def remote_get_buckets(self, storage_index):
 381         start = time.time()
 382         self.count("get")
 383         si_s = si_b2a(storage_index)
 384         log.msg("storage: get_buckets %s" % si_s)
 385         bucketreaders = {} # k: sharenum, v: BucketReader
 386         for shnum, filename in self._get_bucket_shares(storage_index):
 387             bucketreaders[shnum] = BucketReader(self, filename,
 388                                                 storage_index, shnum)
 389         self.add_latency("get", time.time() - start)
 390         return bucketreaders
 391
 392     def get_leases(self, storage_index):
 393         """Provide an iterator that yields all of the leases attached to this
 394         bucket. Each lease is returned as a tuple of (owner_num,
 395         renew_secret, cancel_secret, expiration_time).
 396
 397         This method is not for client use.
 398         """
 399
 400         # since all shares get the same lease data, we just grab the leases
 401         # from the first share
 402         try:
 403             shnum, filename = self._get_bucket_shares(storage_index).next()
 404             sf = ShareFile(filename)
 405             return sf.iter_leases()
 406         except StopIteration:
 407             return iter([])
 408
 409     def remote_slot_testv_and_readv_and_writev(self, storage_index,
 410                                                secrets,
 411                                                test_and_write_vectors,
 412                                                read_vector):
 413         start = time.time()
 414         self.count("writev")
 415         si_s = si_b2a(storage_index)
 416         lp = log.msg("storage: slot_writev %s" % si_s)
 417         si_dir = storage_index_to_dir(storage_index)
 418         (write_enabler, renew_secret, cancel_secret) = secrets
 419         # shares exist if there is a file for them
 420         bucketdir = os.path.join(self.sharedir, si_dir)
 421         shares = {}
 422         if os.path.isdir(bucketdir):
 423             for sharenum_s in os.listdir(bucketdir):
 424                 try:
 425                     sharenum = int(sharenum_s)
 426                 except ValueError:
 427                     continue
 428                 filename = os.path.join(bucketdir, sharenum_s)
 429                 msf = MutableShareFile(filename, self)
 430                 msf.check_write_enabler(write_enabler, si_s)
 431                 shares[sharenum] = msf
 432         # write_enabler is good for all existing shares.
 433
 434         # Now evaluate test vectors.
 435         testv_is_good = True
 436         for sharenum in test_and_write_vectors:
 437             (testv, datav, new_length) = test_and_write_vectors[sharenum]
 438             if sharenum in shares:
 439                 if not shares[sharenum].check_testv(testv):
 440                     self.log("testv failed: [%d]: %r" % (sharenum, testv))
 441                     testv_is_good = False
 442                     break
 443             else:
 444                 # compare the vectors against an empty share, in which all
 445                 # reads return empty strings.
 446                 if not EmptyShare().check_testv(testv):
 447                     self.log("testv failed (empty): [%d] %r" % (sharenum,
 448                                                                 testv))
 449                     testv_is_good = False
 450                     break
 451
 452         # now gather the read vectors, before we do any writes
 453         read_data = {}
 454         for sharenum, share in shares.items():
 455             read_data[sharenum] = share.readv(read_vector)
 456
 457         ownerid = 1 # TODO
 458         expire_time = time.time() + 31*24*60*60   # one month
 459         lease_info = LeaseInfo(ownerid,
 460                                renew_secret, cancel_secret,
 461                                expire_time, self.my_nodeid)
 462
 463         if testv_is_good:
 464             # now apply the write vectors
 465             for sharenum in test_and_write_vectors:
 466                 (testv, datav, new_length) = test_and_write_vectors[sharenum]
 467                 if new_length == 0:
 468                     if sharenum in shares:
 469                         shares[sharenum].unlink()
 470                 else:
 471                     if sharenum not in shares:
 472                         # allocate a new share
 473                         allocated_size = 2000 # arbitrary, really
 474                         share = self._allocate_slot_share(bucketdir, secrets,
 475                                                           sharenum,
 476                                                           allocated_size,
 477                                                           owner_num=0)
 478                         shares[sharenum] = share
 479                     shares[sharenum].writev(datav, new_length)
 480                     # and update the lease
 481                     shares[sharenum].add_or_renew_lease(lease_info)
 482
 483             if new_length == 0:
 484                 # delete empty bucket directories
 485                 if not os.listdir(bucketdir):
 486                     os.rmdir(bucketdir)
 487
 488
 489         # all done
 490         self.add_latency("writev", time.time() - start)
 491         return (testv_is_good, read_data)
 492
 493     def _allocate_slot_share(self, bucketdir, secrets, sharenum,
 494                              allocated_size, owner_num=0):
 495         (write_enabler, renew_secret, cancel_secret) = secrets
 496         my_nodeid = self.my_nodeid
 497         fileutil.make_dirs(bucketdir)
 498         filename = os.path.join(bucketdir, "%d" % sharenum)
 499         share = create_mutable_sharefile(filename, my_nodeid, write_enabler,
 500                                          self)
 501         return share
 502
 503     def remote_slot_readv(self, storage_index, shares, readv):
 504         start = time.time()
 505         self.count("readv")
 506         si_s = si_b2a(storage_index)
 507         lp = log.msg("storage: slot_readv %s %s" % (si_s, shares),
 508                      facility="tahoe.storage", level=log.OPERATIONAL)
 509         si_dir = storage_index_to_dir(storage_index)
 510         # shares exist if there is a file for them
 511         bucketdir = os.path.join(self.sharedir, si_dir)
 512         if not os.path.isdir(bucketdir):
 513             self.add_latency("readv", time.time() - start)
 514             return {}
 515         datavs = {}
 516         for sharenum_s in os.listdir(bucketdir):
 517             try:
 518                 sharenum = int(sharenum_s)
 519             except ValueError:
 520                 continue
 521             if sharenum in shares or not shares:
 522                 filename = os.path.join(bucketdir, sharenum_s)
 523                 msf = MutableShareFile(filename, self)
 524                 datavs[sharenum] = msf.readv(readv)
 525         log.msg("returning shares %s" % (datavs.keys(),),
 526                 facility="tahoe.storage", level=log.NOISY, parent=lp)
 527         self.add_latency("readv", time.time() - start)
 528         return datavs
 529
 530     def remote_advise_corrupt_share(self, share_type, storage_index, shnum,
 531                                     reason):
 532         fileutil.make_dirs(self.corruption_advisory_dir)
 533         now = time_format.iso_utc(sep="T")
 534         si_s = base32.b2a(storage_index)
 535         # windows can't handle colons in the filename
 536         fn = os.path.join(self.corruption_advisory_dir,
 537                           "%s--%s-%d" % (now, si_s, shnum)).replace(":","")
 538         f = open(fn, "w")
 539         f.write("report: Share Corruption\n")
 540         f.write("type: %s\n" % share_type)
 541         f.write("storage_index: %s\n" % si_s)
 542         f.write("share_number: %d\n" % shnum)
 543         f.write("\n")
 544         f.write(reason)
 545         f.write("\n")
 546         f.close()
 547         log.msg(format=("client claims corruption in (%(share_type)s) " +
 548                         "%(si)s-%(shnum)d: %(reason)s"),
 549                 share_type=share_type, si=si_s, shnum=shnum, reason=reason,
 550                 level=log.SCARY, umid="SGx2fA")
 551         return None
 552