lbs.py

   1 """High-level interface for working with LBS archives.
   2
   3 This module provides an easy interface for reading from and manipulating
   4 various parts of an LBS archive:
   5   - listing the snapshots and segments present
   6   - reading segment contents
   7   - parsing snapshot descriptors and snapshot metadata logs
   8   - reading and maintaining the local object database
   9 """
  10
  11 from __future__ import division
  12 import os, re, sha, tarfile, tempfile, thread
  13 from pysqlite2 import dbapi2 as sqlite3
  14
  15 # Maximum number of nested indirect references allowed in a snapshot.
  16 MAX_RECURSION_DEPTH = 3
  17
  18 class Struct:
  19     """A class which merely acts as a data container.
  20
  21     Instances of this class (or its subclasses) are merely used to store data
  22     in various attributes.  No methods are provided.
  23     """
  24
  25     def __repr__(self):
  26         return "<%s %s>" % (self.__class__, self.__dict__)
  27
  28 CHECKSUM_ALGORITHMS = {
  29     'sha1': sha.new
  30 }
  31
  32 class ChecksumCreator:
  33     """Compute an LBS checksum for provided data.
  34
  35     The algorithm used is selectable, but currently defaults to sha1.
  36     """
  37
  38     def __init__(self, algorithm='sha1'):
  39         self.algorithm = algorithm
  40         self.hash = CHECKSUM_ALGORITHMS[algorithm]()
  41
  42     def update(self, data):
  43         self.hash.update(data)
  44         return self
  45
  46     def compute(self):
  47         return "%s=%s" % (self.algorithm, self.hash.hexdigest())
  48
  49 class ChecksumVerifier:
  50     """Verify whether a checksum from a snapshot matches the supplied data."""
  51
  52     def __init__(self, checksumstr):
  53         """Create an object to check the supplied checksum."""
  54
  55         (algo, checksum) = checksumstr.split("=", 1)
  56         self.checksum = checksum
  57         self.hash = CHECKSUM_ALGORITHMS[algo]()
  58
  59     def update(self, data):
  60         self.hash.update(data)
  61
  62     def valid(self):
  63         """Return a boolean indicating whether the checksum matches."""
  64
  65         result = self.hash.hexdigest()
  66         return result == self.checksum
  67
  68 class LowlevelDataStore:
  69     """Access to the backup store containing segments and snapshot descriptors.
  70
  71     Instances of this class are used to get direct filesystem-level access to
  72     the backup data.  To read a backup, a caller will ordinarily not care about
  73     direct access to backup segments, but will instead merely need to access
  74     objects from those segments.  The ObjectStore class provides a suitable
  75     wrapper around a DataStore to give this high-level access.
  76     """
  77
  78     def __init__(self, path):
  79         self.path = path
  80
  81     # Low-level filesystem access.  These methods could be overwritten to
  82     # provide access to remote data stores.
  83     def lowlevel_list(self):
  84         """Get a listing of files stored."""
  85
  86         return os.listdir(self.path)
  87
  88     def lowlevel_open(self, filename):
  89         """Return a file-like object for reading data from the given file."""
  90
  91         return open(os.path.join(self.path, filename), 'rb')
  92
  93     def lowlevel_stat(self, filename):
  94         """Return a dictionary of information about the given file.
  95
  96         Currently, the only defined field is 'size', giving the size of the
  97         file in bytes.
  98         """
  99
 100         stat = os.stat(os.path.join(self.path, filename))
 101         return {'size': stat.st_size}
 102
 103     # Slightly higher-level list methods.
 104     def list_snapshots(self):
 105         for f in self.lowlevel_list():
 106             m = re.match(r"^snapshot-(.*)\.lbs$", f)
 107             if m:
 108                 yield m.group(1)
 109
 110     def list_segments(self):
 111         for f in self.lowlevel_list():
 112             m = re.match(r"^([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})(\.\S+)?$", f)
 113             if m:
 114                 yield m.group(1)
 115
 116 class ObjectStore:
 117     def __init__(self, data_store):
 118         self.store = data_store
 119         self.cachedir = None
 120         self.CACHE_SIZE = 16
 121         self.lru_list = []
 122
 123     def get_cachedir(self):
 124         if self.cachedir is None:
 125             self.cachedir = tempfile.mkdtemp(".lbs")
 126         return self.cachedir
 127
 128     def cleanup(self):
 129         if self.cachedir is not None:
 130             # TODO: Avoid use of system, make this safer
 131             os.system("rm -rv " + self.cachedir)
 132         self.cachedir = None
 133
 134     @staticmethod
 135     def parse_ref(refstr):
 136         m = re.match(r"^([-0-9a-f]+)\/([0-9a-f]+)(\(\S+\))?(\[(\d+)\+(\d+)\])?$", refstr)
 137         if not m: return
 138
 139         segment = m.group(1)
 140         object = m.group(2)
 141         checksum = m.group(3)
 142         slice = m.group(4)
 143
 144         if checksum is not None:
 145             checksum = checksum.lstrip("(").rstrip(")")
 146
 147         if slice is not None:
 148             slice = (int(m.group(5)), int(m.group(6)))
 149
 150         return (segment, object, checksum, slice)
 151
 152     def get_segment(self, segment):
 153         raw = self.store.lowlevel_open(segment + ".tar.gpg")
 154
 155         (input, output) = os.popen2("lbs-filter-gpg --decrypt")
 156         def copy_thread(src, dst):
 157             BLOCK_SIZE = 4096
 158             while True:
 159                 block = src.read(BLOCK_SIZE)
 160                 if len(block) == 0: break
 161                 dst.write(block)
 162             dst.close()
 163
 164         thread.start_new_thread(copy_thread, (raw, input))
 165         return output
 166
 167     def load_segment(self, segment):
 168         seg = tarfile.open(segment, 'r|', self.get_segment(segment))
 169         for item in seg:
 170             data_obj = seg.extractfile(item)
 171             path = item.name.split('/')
 172             if len(path) == 2 and path[0] == segment:
 173                 yield (path[1], data_obj.read())
 174
 175     def load_snapshot(self, snapshot):
 176         file = self.store.lowlevel_open("snapshot-" + snapshot + ".lbs")
 177         return file.read().splitlines(True)
 178
 179     def extract_segment(self, segment):
 180         segdir = os.path.join(self.get_cachedir(), segment)
 181         os.mkdir(segdir)
 182         for (object, data) in self.load_segment(segment):
 183             f = open(os.path.join(segdir, object), 'wb')
 184             f.write(data)
 185             f.close()
 186
 187     def load_object(self, segment, object):
 188         path = os.path.join(self.get_cachedir(), segment, object)
 189         if not os.access(path, os.R_OK):
 190             print "Extracting", segment
 191             self.extract_segment(segment)
 192         if segment in self.lru_list: self.lru_list.remove(segment)
 193         self.lru_list.append(segment)
 194         while len(self.lru_list) > self.CACHE_SIZE:
 195             os.system("rm -rv " + os.path.join(self.cachedir, self.lru_list[0]))
 196             self.lru_list = self.lru_list[1:]
 197         return open(path, 'rb').read()
 198
 199     def get(self, refstr):
 200         """Fetch the given object and return it.
 201
 202         The input should be an object reference, in string form.
 203         """
 204
 205         (segment, object, checksum, slice) = self.parse_ref(refstr)
 206
 207         data = self.load_object(segment, object)
 208
 209         if checksum is not None:
 210             verifier = ChecksumVerifier(checksum)
 211             verifier.update(data)
 212             if not verifier.valid():
 213                 raise ValueError
 214
 215         if slice is not None:
 216             (start, length) = slice
 217             data = data[start:start+length]
 218             if len(data) != length: raise IndexError
 219
 220         return data
 221
 222 def parse(lines, terminate=None):
 223     """Generic parser for RFC822-style "Key: Value" data streams.
 224
 225     This parser can be used to read metadata logs and snapshot root descriptor
 226     files.
 227
 228     lines must be an iterable object which yields a sequence of lines of input.
 229
 230     If terminate is specified, it is used as a predicate to determine when to
 231     stop reading input lines.
 232     """
 233
 234     dict = {}
 235     last_key = None
 236
 237     for l in lines:
 238         # Strip off a trailing newline, if present
 239         if len(l) > 0 and l[-1] == "\n":
 240             l = l[:-1]
 241
 242         if terminate is not None and terminate(l):
 243             if len(dict) > 0: yield dict
 244             dict = {}
 245             last_key = None
 246             continue
 247
 248         m = re.match(r"^(\w+):\s*(.*)$", l)
 249         if m:
 250             dict[m.group(1)] = m.group(2)
 251             last_key = m.group(1)
 252         elif len(l) > 0 and l[0].isspace() and last_key is not None:
 253             dict[last_key] += l
 254         else:
 255             last_key = None
 256
 257     if len(dict) > 0: yield dict
 258
 259 def parse_full(lines):
 260     try:
 261         return parse(lines).next()
 262     except StopIteration:
 263         return {}
 264
 265 def read_metadata(object_store, root):
 266     """Iterate through all lines in the metadata log, following references."""
 267
 268     # Stack for keeping track of recursion when following references to
 269     # portions of the log.  The last entry in the stack corresponds to the
 270     # object currently being parsed.  Each entry is a list of lines which have
 271     # been reversed, so that popping successive lines from the end of each list
 272     # will return lines of the metadata log in order.
 273     stack = []
 274
 275     def follow_ref(refstr):
 276         if len(stack) >= MAX_RECURSION_DEPTH: raise OverflowError
 277         lines = object_store.get(refstr).splitlines(True)
 278         lines.reverse()
 279         stack.append(lines)
 280
 281     follow_ref(root)
 282
 283     while len(stack) > 0:
 284         top = stack[-1]
 285         if len(top) == 0:
 286             stack.pop()
 287             continue
 288         line = top.pop()
 289
 290         # An indirect reference which we must follow?
 291         if len(line) > 0 and line[0] == '@':
 292             ref = line[1:]
 293             ref.strip()
 294             follow_ref(ref)
 295         else:
 296             yield line
 297
 298 class MetadataItem:
 299     """Metadata for a single file (or directory or...) from a snapshot."""
 300
 301     def __init__(self, fields, object_store):
 302         """Initialize from a dictionary of key/value pairs from metadata log."""
 303
 304         self.fields = fields
 305         self.object_store = object_store
 306
 307     def data(self):
 308         """Return an iterator for the data blocks that make up a file."""
 309
 310         # This traverses the list of blocks that make up a file, following
 311         # indirect references.  It is implemented in much the same way as
 312         # read_metadata, so see that function for details of the technique.
 313
 314         objects = self.fields['data'].split()
 315         objects.reverse()
 316         stack = [objects]
 317
 318         def follow_ref(refstr):
 319             if len(stack) >= MAX_RECURSION_DEPTH: raise OverflowError
 320             objects = self.object_store.get(refstr).split()
 321             objects.reverse()
 322             stack.append(objects)
 323
 324         while len(stack) > 0:
 325             top = stack[-1]
 326             if len(top) == 0:
 327                 stack.pop()
 328                 continue
 329             ref = top.pop()
 330
 331             # An indirect reference which we must follow?
 332             if len(ref) > 0 and ref[0] == '@':
 333                 follow_ref(ref[1:])
 334             else:
 335                 yield ref
 336
 337 def iterate_metadata(object_store, root):
 338     for d in parse(read_metadata(object_store, root), lambda l: len(l) == 0):
 339         yield MetadataItem(d, object_store)
 340
 341 class LocalDatabase:
 342     """Access to the local database of snapshot contents and object checksums.
 343
 344     The local database is consulted when creating a snapshot to determine what
 345     data can be re-used from old snapshots.  Segment cleaning is performed by
 346     manipulating the data in the local database; the local database also
 347     includes enough data to guide the segment cleaning process.
 348     """
 349
 350     def __init__(self, path, dbname="localdb.sqlite"):
 351         self.db_connection = sqlite3.connect(path + "/" + dbname)
 352
 353     # Low-level database access.  Use these methods when there isn't a
 354     # higher-level interface available.  Exception: do, however, remember to
 355     # use the commit() method after making changes to make sure they are
 356     # actually saved, even when going through higher-level interfaces.
 357     def commit(self):
 358         "Commit any pending changes to the local database."
 359         self.db_connection.commit()
 360
 361     def rollback(self):
 362         "Roll back any pending changes to the local database."
 363         self.db_connection.rollback()
 364
 365     def cursor(self):
 366         "Return a DB-API cursor for directly accessing the local database."
 367         return self.db_connection.cursor()
 368
 369     def garbage_collect(self):
 370         """Delete entries from old snapshots from the database."""
 371
 372         cur = self.cursor()
 373
 374         # Delete old snapshots.
 375         cur.execute("""delete from snapshots
 376                        where snapshotid < (select max(snapshotid)
 377                                            from snapshots)""")
 378
 379         # Delete entries in the snapshot_contents table which are for
 380         # non-existent snapshots.
 381         cur.execute("""delete from snapshot_contents
 382                        where snapshotid not in
 383                            (select snapshotid from snapshots)""")
 384
 385         # Find segments which contain no objects used by any current snapshots,
 386         # and delete them from the segment table.
 387         cur.execute("""delete from segments where segmentid not in
 388                            (select distinct segmentid from snapshot_contents
 389                                 natural join block_index)""")
 390
 391         # Finally, delete objects contained in non-existent segments.  We can't
 392         # simply delete unused objects, since we use the set of unused objects
 393         # to determine the used/free ratio of segments.
 394         cur.execute("""delete from block_index
 395                        where segmentid not in
 396                            (select segmentid from segments)""")
 397
 398     # Segment cleaning.
 399     class SegmentInfo(Struct): pass
 400
 401     def get_segment_cleaning_list(self, age_boost=0.0):
 402         """Return a list of all current segments with information for cleaning.
 403
 404         Return all segments which are currently known in the local database
 405         (there might be other, older segments in the archive itself), and
 406         return usage statistics for each to help decide which segments to
 407         clean.
 408
 409         The returned list will be sorted by estimated cleaning benefit, with
 410         segments that are best to clean at the start of the list.
 411
 412         If specified, the age_boost parameter (measured in days) will added to
 413         the age of each segment, as a way of adjusting the benefit computation
 414         before a long-lived snapshot is taken (for example, age_boost might be
 415         set to 7 when cleaning prior to taking a weekly snapshot).
 416         """
 417
 418         cur = self.cursor()
 419         segments = []
 420         cur.execute("""select segmentid, used, size, mtime,
 421                        julianday('now') - mtime as age from segment_info""")
 422         for row in cur:
 423             info = self.SegmentInfo()
 424             info.id = row[0]
 425             info.used_bytes = row[1]
 426             info.size_bytes = row[2]
 427             info.mtime = row[3]
 428             info.age_days = row[4]
 429
 430             # Benefit calculation: u is the estimated fraction of each segment
 431             # which is utilized (bytes belonging to objects still in use
 432             # divided by total size; this doesn't take compression or storage
 433             # overhead into account, but should give a reasonable estimate).
 434             #
 435             # The total benefit is a heuristic that combines several factors:
 436             # the amount of space that can be reclaimed (1 - u), an ageing
 437             # factor (info.age_days) that favors cleaning old segments to young
 438             # ones and also is more likely to clean segments that will be
 439             # rewritten for long-lived snapshots (age_boost), and finally a
 440             # penalty factor for the cost of re-uploading data (u + 0.1).
 441             u = info.used_bytes / info.size_bytes
 442             info.cleaning_benefit \
 443                 = (1 - u) * (info.age_days + age_boost) / (u + 0.1)
 444
 445             segments.append(info)
 446
 447         segments.sort(cmp, key=lambda s: s.cleaning_benefit, reverse=True)
 448         return segments
 449
 450     def mark_segment_expired(self, segment):
 451         """Mark a segment for cleaning in the local database.
 452
 453         The segment parameter should be either a SegmentInfo object or an
 454         integer segment id.  Objects in the given segment will be marked as
 455         expired, which means that any future snapshots that would re-use those
 456         objects will instead write out a new copy of the object, and thus no
 457         future snapshots will depend upon the given segment.
 458         """
 459
 460         if isinstance(segment, int):
 461             id = segment
 462         elif isinstance(segment, self.SegmentInfo):
 463             id = segment.id
 464         else:
 465             raise TypeError("Invalid segment: %s, must be of type int or SegmentInfo, not %s" % (segment, type(segment)))
 466
 467         cur = self.cursor()
 468         cur.execute("update block_index set expired = 1 where segmentid = ?",
 469                     (id,))
 470
 471     def balance_expired_objects(self):
 472         """Analyze expired objects in segments to be cleaned and group by age.
 473
 474         Update the block_index table of the local database to group expired
 475         objects by age.  The exact number of buckets and the cutoffs for each
 476         are dynamically determined.  Calling this function after marking
 477         segments expired will help in the segment cleaning process, by ensuring
 478         that when active objects from clean segments are rewritten, they will
 479         be placed into new segments roughly grouped by age.
 480         """
 481
 482         # The expired column of the block_index table is used when generating a
 483         # new LBS snapshot.  A null value indicates that an object may be
 484         # re-used.  Otherwise, an object must be written into a new segment if
 485         # needed.  Objects with distinct expired values will be written into
 486         # distinct segments, to allow for some grouping by age.  The value 0 is
 487         # somewhat special in that it indicates any rewritten objects can be
 488         # placed in the same segment as completely new objects; this can be
 489         # used for very young objects which have been expired, or objects not
 490         # expected to be encountered.
 491         #
 492         # In the balancing process, all objects which are not used in any
 493         # current snapshots will have expired set to 0.  Objects which have
 494         # been seen will be sorted by age and will have expired values set to
 495         # 0, 1, 2, and so on based on age (with younger objects being assigned
 496         # lower values).  The number of buckets and the age cutoffs is
 497         # determined by looking at the distribution of block ages.
 498
 499         cur = self.cursor()
 500
 501         # First step: Mark all unused-and-expired objects with expired = -1,
 502         # which will cause us to mostly ignore these objects when rebalancing.
 503         # At the end, we will set these objects to be in group expired = 0.
 504         # Mark expired objects which still seem to be in use with expired = 0;
 505         # these objects will later have values set to indicate groupings of
 506         # objects when repacking.
 507         cur.execute("""update block_index set expired = -1
 508                        where expired is not null""")
 509
 510         cur.execute("""update block_index set expired = 0
 511                        where expired is not null and blockid in
 512                            (select blockid from snapshot_contents)""")
 513
 514         # We will want to aim for at least one full segment for each bucket
 515         # that we eventually create, but don't know how many bytes that should
 516         # be due to compression.  So compute the average number of bytes in
 517         # each expired segment as a rough estimate for the minimum size of each
 518         # bucket.  (This estimate could be thrown off by many not-fully-packed
 519         # segments, but for now don't worry too much about that.)  If we can't
 520         # compute an average, it's probably because there are no expired
 521         # segments, so we have no more work to do.
 522         cur.execute("""select avg(size) from segment_info
 523                        where segmentid in
 524                            (select distinct segmentid from block_index
 525                             where expired is not null)""")
 526         segment_size_estimate = cur.fetchone()[0]
 527         if not segment_size_estimate:
 528             return
 529
 530         # Next, extract distribution of expired objects (number and size) by
 531         # age.  Save the timestamp for "now" so that the classification of
 532         # blocks into age buckets will not change later in the function, after
 533         # time has passed.  Set any timestamps in the future to now, so we are
 534         # guaranteed that for the rest of this function, age is always
 535         # non-negative.
 536         cur.execute("select julianday('now')")
 537         now = cur.fetchone()[0]
 538
 539         cur.execute("""update block_index set timestamp = ?
 540                        where timestamp > ? and expired is not null""",
 541                     (now, now))
 542
 543         cur.execute("""select round(? - timestamp) as age, count(*), sum(size)
 544                        from block_index where expired = 0
 545                        group by age order by age""", (now,))
 546         distribution = cur.fetchall()
 547
 548         # Start to determine the buckets for expired objects.  Heuristics used:
 549         #   - An upper bound on the number of buckets is given by the number of
 550         #     segments we estimate it will take to store all data.  In fact,
 551         #     aim for a couple of segments per bucket.
 552         #   - Place very young objects in bucket 0 (place with new objects)
 553         #     unless there are enough of them to warrant a separate bucket.
 554         #   - Try not to create unnecessarily many buckets, since fewer buckets
 555         #     will allow repacked data to be grouped based on spatial locality
 556         #     (while more buckets will group by temporal locality).  We want a
 557         #     balance.
 558         MIN_AGE = 4
 559         total_bytes = sum([i[2] for i in distribution])
 560         target_buckets = 2 * (total_bytes / segment_size_estimate) ** 0.4
 561         min_size = 1.5 * segment_size_estimate
 562         target_size = max(2 * segment_size_estimate,
 563                           total_bytes / target_buckets)
 564
 565         print "segment_size:", segment_size_estimate
 566         print "distribution:", distribution
 567         print "total_bytes:", total_bytes
 568         print "target_buckets:", target_buckets
 569         print "min, target size:", min_size, target_size
 570
 571         # Chosen cutoffs.  Each bucket consists of objects with age greater
 572         # than one cutoff value, but not greater than the next largest cutoff.
 573         cutoffs = []
 574
 575         # Starting with the oldest objects, begin grouping together into
 576         # buckets of size at least target_size bytes.
 577         distribution.reverse()
 578         bucket_size = 0
 579         min_age_bucket = False
 580         for (age, items, size) in distribution:
 581             if bucket_size >= target_size \
 582                 or (age < MIN_AGE and not min_age_bucket):
 583                 if bucket_size < target_size and len(cutoffs) > 0:
 584                     cutoffs.pop()
 585                 cutoffs.append(age)
 586                 bucket_size = 0
 587
 588             bucket_size += size
 589             if age < MIN_AGE:
 590                 min_age_bucket = True
 591
 592         # The last (youngest) bucket will be group 0, unless it has enough data
 593         # to be of size min_size by itself, or there happen to be no objects
 594         # less than MIN_AGE at all.
 595         if bucket_size >= min_size or not min_age_bucket:
 596             cutoffs.append(-1)
 597         cutoffs.append(-1)
 598
 599         print "cutoffs:", cutoffs
 600
 601         # Update the database to assign each object to the appropriate bucket.
 602         cutoffs.reverse()
 603         for i in range(len(cutoffs)):
 604             cur.execute("""update block_index set expired = ?
 605                            where round(? - timestamp) > ? and expired >= 0""",
 606                         (i, now, cutoffs[i]))
 607         cur.execute("update block_index set expired = 0 where expired = -1")