Use posixpath to join remote storage paths.
[cumulus.git] / python / cumulus / __init__.py
1 # Cumulus: Efficient Filesystem Backup to the Cloud
2 # Copyright (C) 2008-2009, 2012 The Cumulus Developers
3 # See the AUTHORS file for a list of contributors.
4 #
5 # This program is free software; you can redistribute it and/or modify
6 # it under the terms of the GNU General Public License as published by
7 # the Free Software Foundation; either version 2 of the License, or
8 # (at your option) any later version.
9 #
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13 # GNU General Public License for more details.
14 #
15 # You should have received a copy of the GNU General Public License along
16 # with this program; if not, write to the Free Software Foundation, Inc.,
17 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18
19 """High-level interface for working with Cumulus archives.
20
21 This module provides an easy interface for reading from and manipulating
22 various parts of a Cumulus archive:
23   - listing the snapshots and segments present
24   - reading segment contents
25   - parsing snapshot descriptors and snapshot metadata logs
26   - reading and maintaining the local object database
27 """
28
29 from __future__ import division, print_function, unicode_literals
30
31 import codecs
32 import hashlib
33 import itertools
34 import os
35 import posixpath
36 import re
37 import sqlite3
38 import subprocess
39 import sys
40 import tarfile
41 import tempfile
42 try:
43     import _thread
44 except ImportError:
45     import thread as _thread
46
47 import cumulus.store
48 import cumulus.store.file
49
50 if sys.version < "3":
51     StringTypes = (str, unicode)
52 else:
53     StringTypes = (str,)
54
55 # The largest supported snapshot format that can be understood.
56 FORMAT_VERSION = (0, 11)        # Cumulus Snapshot v0.11
57
58 # Maximum number of nested indirect references allowed in a snapshot.
59 MAX_RECURSION_DEPTH = 3
60
61 # All segments which have been accessed this session.
62 accessed_segments = set()
63
64 # Table of methods used to filter segments before storage, and corresponding
65 # filename extensions.  These are listed in priority order (methods earlier in
66 # the list are tried first).
67 SEGMENT_FILTERS = [
68     (".gpg", "cumulus-filter-gpg --decrypt"),
69     (".gz", "gzip -dc"),
70     (".bz2", "bzip2 -dc"),
71     ("", None),
72 ]
73
74 def to_lines(data):
75     """Decode binary data from a file into a sequence of lines.
76
77     Newline markers are retained."""
78     return list(codecs.iterdecode(data.splitlines(True), "utf-8"))
79
80 def uri_decode(s):
81     """Decode a URI-encoded (%xx escapes) string."""
82     def hex_decode(m): return chr(int(m.group(1), 16))
83     return re.sub(r"%([0-9a-f]{2})", hex_decode, s)
84 def uri_encode(s):
85     """Encode a string to URI-encoded (%xx escapes) form."""
86     def hex_encode(c):
87         if c > '+' and c < '\x7f' and c != '@':
88             return c
89         else:
90             return "%%%02x" % (ord(c),)
91     return ''.join(hex_encode(c) for c in s)
92
93 class Struct:
94     """A class which merely acts as a data container.
95
96     Instances of this class (or its subclasses) are merely used to store data
97     in various attributes.  No methods are provided.
98     """
99
100     def __repr__(self):
101         return "<%s %s>" % (self.__class__, self.__dict__)
102
103 CHECKSUM_ALGORITHMS = {
104     'sha1': hashlib.sha1,
105     'sha224': hashlib.sha224,
106     'sha256': hashlib.sha256,
107 }
108
109 class ChecksumCreator:
110     """Compute a Cumulus checksum for provided data.
111
112     The algorithm used is selectable, but currently defaults to sha1.
113     """
114
115     def __init__(self, algorithm='sha1'):
116         self.algorithm = algorithm
117         self.hash = CHECKSUM_ALGORITHMS[algorithm]()
118
119     def update(self, data):
120         self.hash.update(data)
121         return self
122
123     def compute(self):
124         return "%s=%s" % (self.algorithm, self.hash.hexdigest())
125
126 class ChecksumVerifier:
127     """Verify whether a checksum from a snapshot matches the supplied data."""
128
129     def __init__(self, checksumstr):
130         """Create an object to check the supplied checksum."""
131
132         (algo, checksum) = checksumstr.split("=", 1)
133         self.checksum = checksum
134         self.hash = CHECKSUM_ALGORITHMS[algo]()
135
136     def update(self, data):
137         self.hash.update(data)
138
139     def valid(self):
140         """Return a boolean indicating whether the checksum matches."""
141
142         result = self.hash.hexdigest()
143         return result == self.checksum
144
145 class SearchPathEntry(object):
146     """Item representing a possible search location for Cumulus files.
147
148     Some Cumulus files might be stored in multiple possible file locations: due
149     to format (different compression mechanisms with different extensions),
150     locality (different segments might be placed in different directories to
151     control archiving policies), for backwards compatibility (default location
152     changed over time).  A SearchPathEntry describes a possible location for a
153     file.
154     """
155     def __init__(self, directory_prefix, suffix, context=None):
156         self._directory_prefix = directory_prefix
157         self._suffix = suffix
158         self._context = context
159
160     def __repr__(self):
161         return "%s(%r, %r, %r)" % (self.__class__.__name__,
162                                    self._directory_prefix, self._suffix,
163                                    self._context)
164
165     def build_path(self, basename):
166         """Construct the search path to use for a file with name basename.
167
168         Returns a tuple (pathname, context), where pathname is the path to try
169         and context is any additional data associated with this search entry
170         (if any).
171         """
172         return (posixpath.join(self._directory_prefix, basename + self._suffix),
173                 self._context)
174
175 class SearchPath(object):
176     """A collection of locations to search for files and lookup utilities.
177
178     For looking for a file in a Cumulus storage backend, a SearchPath object
179     contains a list of possible locations to try.  A SearchPath can be used to
180     perform the search as well; when a file is found the search path ordering
181     is updated (moving the successful SearchPathEntry to the front of the list
182     for future searches).
183     """
184     def __init__(self, name_regex, searchpath):
185         self._regex = re.compile(name_regex)
186         self._path = list(searchpath)
187
188     def add_search_entry(self, entry):
189         self._path.append(entry)
190
191     def directories(self):
192         """Return the set of directories to search for a file type."""
193         return set(entry._directory_prefix for entry in self._path)
194
195     def get(self, backend, basename):
196         for (i, entry) in enumerate(self._path):
197             try:
198                 (pathname, context) = entry.build_path(basename)
199                 fp = backend.get(pathname)
200                 # On success, move this entry to the front of the search path
201                 # to speed future searches.
202                 if i > 0:
203                     self._path.pop(i)
204                     self._path.insert(0, entry)
205                 return (fp, pathname, context)
206             except cumulus.store.NotFoundError:
207                 continue
208         raise cumulus.store.NotFoundError(basename)
209
210     def stat(self, backend, basename):
211         for (i, entry) in enumerate(self._path):
212             try:
213                 (pathname, context) = entry.build_path(basename)
214                 stat_data = backend.stat(pathname)
215                 # On success, move this entry to the front of the search path
216                 # to speed future searches.
217                 if i > 0:
218                     self._path.pop(i)
219                     self._path.insert(0, entry)
220                 result = {"path": pathname}
221                 result.update(stat_data)
222                 return result
223             except cumulus.store.NotFoundError:
224                 continue
225         raise cumulus.store.NotFoundError(basename)
226
227     def match(self, filename):
228         return self._regex.match(filename)
229
230     def list(self, backend):
231         success = False
232         for d in self.directories():
233             try:
234                 for f in backend.list(d):
235                     success = True
236                     m = self.match(f)
237                     if m: yield (posixpath.join(d, f), m)
238             except cumulus.store.NotFoundError:
239                 pass
240         if not success:
241             raise cumulus.store.NotFoundError(backend)
242
243 def _build_segments_searchpath(prefix):
244     for (extension, filter) in SEGMENT_FILTERS:
245         yield SearchPathEntry(prefix, extension, filter)
246
247 SEARCH_PATHS = {
248     "checksums": SearchPath(
249         r"^snapshot-(.*)\.(\w+)sums$",
250         [SearchPathEntry("meta", ".sha1sums"),
251          SearchPathEntry("checksums", ".sha1sums"),
252          SearchPathEntry("", ".sha1sums")]),
253     "meta": SearchPath(
254         r"^snapshot-(.*)\.meta(\.\S+)?$",
255         _build_segments_searchpath("meta")),
256     "segments": SearchPath(
257         (r"^([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})"
258          r"\.tar(\.\S+)?$"),
259         itertools.chain(
260             _build_segments_searchpath("segments0"),
261             _build_segments_searchpath("segments1"),
262             _build_segments_searchpath(""),
263             _build_segments_searchpath("segments"))),
264     "snapshots": SearchPath(
265         r"^snapshot-(.*)\.(cumulus|lbs)$",
266         [SearchPathEntry("snapshots", ".cumulus"),
267          SearchPathEntry("snapshots", ".lbs"),
268          SearchPathEntry("", ".cumulus"),
269          SearchPathEntry("", ".lbs")]),
270 }
271
272 class BackendWrapper(object):
273     """Wrapper around a Cumulus storage backend that understands file types.
274
275     The BackendWrapper class understands different Cumulus file types, such as
276     snapshots and segments, and implements higher-level operations such as
277     "retrieve a snapshot with a specific name" (hiding operations such as
278     searching for the correct file name).
279     """
280
281     def __init__(self, backend):
282         """Initializes a wrapper around the specified storage backend.
283
284         store may either be a Store object or URL.
285         """
286         if type(backend) in StringTypes:
287             self._backend = cumulus.store.open(backend)
288         else:
289             self._backend = backend
290
291     @property
292     def raw_backend(self):
293         return self._backend
294
295     def stat_generic(self, basename, filetype):
296         return SEARCH_PATHS[filetype].stat(self._backend, basename)
297
298     def open_generic(self, basename, filetype):
299         return SEARCH_PATHS[filetype].get(self._backend, basename)
300
301     def open_snapshot(self, name):
302         return self.open_generic("snapshot-" + name, "snapshots")
303
304     def open_segment(self, name):
305         return self.open_generic(name + ".tar", "segments")
306
307     def list_generic(self, filetype):
308         return ((x[1].group(1), x[0])
309                 for x in SEARCH_PATHS[filetype].list(self._backend))
310
311     def prefetch_generic(self):
312         """Calls scan on directories to prefetch file metadata."""
313         directories = set()
314         for typeinfo in SEARCH_PATHS.values():
315             directories.update(typeinfo.directories())
316         for d in directories:
317             print("Prefetch", d)
318             self._backend.scan(d)
319
320 class CumulusStore:
321     def __init__(self, backend):
322         if isinstance(backend, BackendWrapper):
323             self.backend = backend
324         else:
325             self.backend = BackendWrapper(backend)
326         self.cachedir = None
327         self.CACHE_SIZE = 16
328         self._lru_list = []
329
330     def get_cachedir(self):
331         if self.cachedir is None:
332             self.cachedir = tempfile.mkdtemp("-cumulus")
333         return self.cachedir
334
335     def cleanup(self):
336         if self.cachedir is not None:
337             # TODO: Avoid use of system, make this safer
338             os.system("rm -rf " + self.cachedir)
339         self.cachedir = None
340
341     @staticmethod
342     def parse_ref(refstr):
343         m = re.match(r"^zero\[(\d+)\]$", refstr)
344         if m:
345             return ("zero", None, None, (0, int(m.group(1)), False))
346
347         m = re.match(r"^([-0-9a-f]+)\/([0-9a-f]+)(\(\S+\))?(\[(=?(\d+)|(\d+)\+(\d+))\])?$", refstr)
348         if not m: return
349
350         segment = m.group(1)
351         object = m.group(2)
352         checksum = m.group(3)
353         slice = m.group(4)
354
355         if checksum is not None:
356             checksum = checksum.lstrip("(").rstrip(")")
357
358         if slice is not None:
359             if m.group(6) is not None:
360                 # Size-assertion slice
361                 slice = (0, int(m.group(6)), True)
362             else:
363                 slice = (int(m.group(7)), int(m.group(8)), False)
364
365         return (segment, object, checksum, slice)
366
367     def list_snapshots(self):
368         return set(x[0] for x in self.backend.list_generic("snapshots"))
369
370     def list_segments(self):
371         return set(x[0] for x in self.backend.list_generic("segments"))
372
373     def load_snapshot(self, snapshot):
374         snapshot_file = self.backend.open_snapshot(snapshot)[0]
375         return to_lines(snapshot_file.read())
376
377     @staticmethod
378     def filter_data(filehandle, filter_cmd):
379         if filter_cmd is None:
380             return filehandle
381         p = subprocess.Popen(filter_cmd, shell=True, stdin=subprocess.PIPE,
382                              stdout=subprocess.PIPE, close_fds=True)
383         input, output = p.stdin, p.stdout
384         def copy_thread(src, dst):
385             BLOCK_SIZE = 4096
386             while True:
387                 block = src.read(BLOCK_SIZE)
388                 if len(block) == 0: break
389                 dst.write(block)
390             src.close()
391             dst.close()
392             p.wait()
393         _thread.start_new_thread(copy_thread, (filehandle, input))
394         return output
395
396     def get_segment(self, segment):
397         accessed_segments.add(segment)
398
399         (segment_fp, path, filter_cmd) = self.backend.open_segment(segment)
400         return self.filter_data(segment_fp, filter_cmd)
401
402     def load_segment(self, segment):
403         seg = tarfile.open(segment, 'r|', self.get_segment(segment))
404         for item in seg:
405             data_obj = seg.extractfile(item)
406             path = item.name.split('/')
407             if len(path) == 2 and path[0] == segment:
408                 yield (path[1], data_obj.read())
409
410     def extract_segment(self, segment):
411         segdir = os.path.join(self.get_cachedir(), segment)
412         os.mkdir(segdir)
413         for (object, data) in self.load_segment(segment):
414             f = open(os.path.join(segdir, object), 'wb')
415             f.write(data)
416             f.close()
417
418     def load_object(self, segment, object):
419         accessed_segments.add(segment)
420         path = os.path.join(self.get_cachedir(), segment, object)
421         if not os.access(path, os.R_OK):
422             self.extract_segment(segment)
423         if segment in self._lru_list: self._lru_list.remove(segment)
424         self._lru_list.append(segment)
425         while len(self._lru_list) > self.CACHE_SIZE:
426             os.system("rm -rf " + os.path.join(self.cachedir,
427                                                self._lru_list[0]))
428             self._lru_list = self._lru_list[1:]
429         return open(path, 'rb').read()
430
431     def get(self, refstr):
432         """Fetch the given object and return it.
433
434         The input should be an object reference, in string form.
435         """
436
437         (segment, object, checksum, slice) = self.parse_ref(refstr)
438
439         if segment == "zero":
440             return "\0" * slice[1]
441
442         data = self.load_object(segment, object)
443
444         if checksum is not None:
445             verifier = ChecksumVerifier(checksum)
446             verifier.update(data)
447             if not verifier.valid():
448                 raise ValueError
449
450         if slice is not None:
451             (start, length, exact) = slice
452             # Note: The following assertion check may need to be commented out
453             # to restore from pre-v0.8 snapshots, as the syntax for
454             # size-assertion slices has changed.
455             if exact and len(data) != length: raise ValueError
456             data = data[start:start+length]
457             if len(data) != length: raise IndexError
458
459         return data
460
461     def prefetch(self):
462         self.backend.prefetch_generic()
463
464 def parse(lines, terminate=None):
465     """Generic parser for RFC822-style "Key: Value" data streams.
466
467     This parser can be used to read metadata logs and snapshot root descriptor
468     files.
469
470     lines must be an iterable object which yields a sequence of lines of input.
471
472     If terminate is specified, it is used as a predicate to determine when to
473     stop reading input lines.
474     """
475
476     dict = {}
477     last_key = None
478
479     for l in lines:
480         # Strip off a trailing newline, if present
481         if len(l) > 0 and l[-1] == "\n":
482             l = l[:-1]
483
484         if terminate is not None and terminate(l):
485             if len(dict) > 0: yield dict
486             dict = {}
487             last_key = None
488             continue
489
490         m = re.match(r"^([-\w]+):\s*(.*)$", l)
491         if m:
492             dict[m.group(1)] = m.group(2)
493             last_key = m.group(1)
494         elif len(l) > 0 and l[0].isspace() and last_key is not None:
495             dict[last_key] += l
496         else:
497             last_key = None
498
499     if len(dict) > 0: yield dict
500
501 def parse_full(lines):
502     try:
503         return next(parse(lines))
504     except StopIteration:
505         return {}
506
507 def parse_metadata_version(s):
508     """Convert a string with the snapshot version format to a tuple."""
509
510     m = re.match(r"^(?:Cumulus|LBS) Snapshot v(\d+(\.\d+)*)$", s)
511     if m is None:
512         return ()
513     else:
514         return tuple([int(d) for d in m.group(1).split(".")])
515
516 def read_metadata(object_store, root):
517     """Iterate through all lines in the metadata log, following references."""
518
519     # Stack for keeping track of recursion when following references to
520     # portions of the log.  The last entry in the stack corresponds to the
521     # object currently being parsed.  Each entry is a list of lines which have
522     # been reversed, so that popping successive lines from the end of each list
523     # will return lines of the metadata log in order.
524     stack = []
525
526     def follow_ref(refstr):
527         if len(stack) >= MAX_RECURSION_DEPTH: raise OverflowError
528         lines = to_lines(object_store.get(refstr))
529         lines.reverse()
530         stack.append(lines)
531
532     follow_ref(root)
533
534     while len(stack) > 0:
535         top = stack[-1]
536         if len(top) == 0:
537             stack.pop()
538             continue
539         line = top.pop()
540
541         # An indirect reference which we must follow?
542         if len(line) > 0 and line[0] == '@':
543             ref = line[1:]
544             ref.strip()
545             follow_ref(ref)
546         else:
547             yield line
548
549 class MetadataItem:
550     """Metadata for a single file (or directory or...) from a snapshot."""
551
552     # Functions for parsing various datatypes that can appear in a metadata log
553     # item.
554     @staticmethod
555     def decode_int(s):
556         """Decode an integer, expressed in decimal, octal, or hexadecimal."""
557         if s.startswith("0x"):
558             return int(s, 16)
559         elif s.startswith("0"):
560             return int(s, 8)
561         else:
562             return int(s, 10)
563
564     @staticmethod
565     def decode_str(s):
566         """Decode a URI-encoded (%xx escapes) string."""
567         return uri_decode(s)
568
569     @staticmethod
570     def raw_str(s):
571         """An unecoded string."""
572         return s
573
574     @staticmethod
575     def decode_user(s):
576         """Decode a user/group to a tuple of uid/gid followed by name."""
577         items = s.split()
578         uid = MetadataItem.decode_int(items[0])
579         name = None
580         if len(items) > 1:
581             if items[1].startswith("(") and items[1].endswith(")"):
582                 name = MetadataItem.decode_str(items[1][1:-1])
583         return (uid, name)
584
585     @staticmethod
586     def decode_device(s):
587         """Decode a device major/minor number."""
588         (major, minor) = map(MetadataItem.decode_int, s.split("/"))
589         return (major, minor)
590
591     class Items: pass
592
593     def __init__(self, fields, object_store):
594         """Initialize from a dictionary of key/value pairs from metadata log."""
595
596         self.fields = fields
597         self.object_store = object_store
598         self.keys = []
599         self.items = self.Items()
600         for (k, v) in fields.items():
601             if k in self.field_types:
602                 decoder = self.field_types[k]
603                 setattr(self.items, k, decoder(v))
604                 self.keys.append(k)
605
606     def data(self):
607         """Return an iterator for the data blocks that make up a file."""
608
609         # This traverses the list of blocks that make up a file, following
610         # indirect references.  It is implemented in much the same way as
611         # read_metadata, so see that function for details of the technique.
612
613         objects = self.fields['data'].split()
614         objects.reverse()
615         stack = [objects]
616
617         def follow_ref(refstr):
618             if len(stack) >= MAX_RECURSION_DEPTH: raise OverflowError
619             objects = self.object_store.get(refstr).split()
620             objects.reverse()
621             stack.append(objects)
622
623         while len(stack) > 0:
624             top = stack[-1]
625             if len(top) == 0:
626                 stack.pop()
627                 continue
628             ref = top.pop()
629
630             # An indirect reference which we must follow?
631             if len(ref) > 0 and ref[0] == '@':
632                 follow_ref(ref[1:])
633             else:
634                 yield ref
635
636 # Description of fields that might appear, and how they should be parsed.
637 MetadataItem.field_types = {
638     'name': MetadataItem.decode_str,
639     'type': MetadataItem.raw_str,
640     'mode': MetadataItem.decode_int,
641     'device': MetadataItem.decode_device,
642     'user': MetadataItem.decode_user,
643     'group': MetadataItem.decode_user,
644     'ctime': MetadataItem.decode_int,
645     'mtime': MetadataItem.decode_int,
646     'links': MetadataItem.decode_int,
647     'inode': MetadataItem.raw_str,
648     'checksum': MetadataItem.decode_str,
649     'size': MetadataItem.decode_int,
650     'contents': MetadataItem.decode_str,
651     'target': MetadataItem.decode_str,
652 }
653
654 def iterate_metadata(object_store, root):
655     for d in parse(read_metadata(object_store, root), lambda l: len(l) == 0):
656         yield MetadataItem(d, object_store)
657
658 class LocalDatabase:
659     """Access to the local database of snapshot contents and object checksums.
660
661     The local database is consulted when creating a snapshot to determine what
662     data can be re-used from old snapshots.  Segment cleaning is performed by
663     manipulating the data in the local database; the local database also
664     includes enough data to guide the segment cleaning process.
665     """
666
667     def __init__(self, path, dbname="localdb.sqlite"):
668         self.db_connection = sqlite3.connect(path + "/" + dbname)
669
670     # Low-level database access.  Use these methods when there isn't a
671     # higher-level interface available.  Exception: do, however, remember to
672     # use the commit() method after making changes to make sure they are
673     # actually saved, even when going through higher-level interfaces.
674     def commit(self):
675         "Commit any pending changes to the local database."
676         self.db_connection.commit()
677
678     def rollback(self):
679         "Roll back any pending changes to the local database."
680         self.db_connection.rollback()
681
682     def cursor(self):
683         "Return a DB-API cursor for directly accessing the local database."
684         return self.db_connection.cursor()
685
686     def list_schemes(self):
687         """Return the list of snapshots found in the local database.
688
689         The returned value is a list of tuples (id, scheme, name, time, intent).
690         """
691
692         cur = self.cursor()
693         cur.execute("select distinct scheme from snapshots")
694         schemes = [row[0] for row in cur.fetchall()]
695         schemes.sort()
696         return schemes
697
698     def list_snapshots(self, scheme):
699         """Return a list of snapshots for the given scheme."""
700         cur = self.cursor()
701         cur.execute("select name from snapshots")
702         snapshots = [row[0] for row in cur.fetchall()]
703         snapshots.sort()
704         return snapshots
705
706     def delete_snapshot(self, scheme, name):
707         """Remove the specified snapshot from the database.
708
709         Warning: This does not garbage collect all dependent data in the
710         database, so it must be followed by a call to garbage_collect() to make
711         the database consistent.
712         """
713         cur = self.cursor()
714         cur.execute("delete from snapshots where scheme = ? and name = ?",
715                     (scheme, name))
716
717     def prune_old_snapshots(self, scheme, intent=1.0):
718         """Delete entries from old snapshots from the database.
719
720         Only snapshots with the specified scheme name will be deleted.  If
721         intent is given, it gives the intended next snapshot type, to determine
722         how aggressively to clean (for example, intent=7 could be used if the
723         next snapshot will be a weekly snapshot).
724         """
725
726         cur = self.cursor()
727
728         # Find the id of the last snapshot to be created.  This is used for
729         # measuring time in a way: we record this value in each segment we
730         # expire on this run, and then on a future run can tell if there have
731         # been intervening backups made.
732         cur.execute("select max(snapshotid) from snapshots")
733         last_snapshotid = cur.fetchone()[0]
734
735         # Get the list of old snapshots for this scheme.  Delete all the old
736         # ones.  Rules for what to keep:
737         #   - Always keep the most recent snapshot.
738         #   - If snapshot X is younger than Y, and X has higher intent, then Y
739         #     can be deleted.
740         cur.execute("""select snapshotid, name, intent,
741                               julianday('now') - timestamp as age
742                        from snapshots where scheme = ?
743                        order by age""", (scheme,))
744
745         first = True
746         max_intent = intent
747         for (id, name, snap_intent, snap_age) in cur.fetchall():
748             can_delete = False
749             if snap_intent < max_intent:
750                 # Delete small-intent snapshots if there is a more recent
751                 # large-intent snapshot.
752                 can_delete = True
753             elif snap_intent == intent:
754                 # Delete previous snapshots with the specified intent level.
755                 can_delete = True
756
757             if can_delete and not first:
758                 print("Delete snapshot %d (%s)" % (id, name))
759                 cur.execute("delete from snapshots where snapshotid = ?",
760                             (id,))
761             first = False
762             max_intent = max(max_intent, snap_intent)
763
764         self.garbage_collect()
765
766     def garbage_collect(self):
767         """Garbage-collect unreachable segment and object data.
768
769         Remove all segments and checksums which is not reachable from the
770         current set of snapshots stored in the local database.
771         """
772         cur = self.cursor()
773
774         # Delete entries in the segment_utilization table which are for
775         # non-existent snapshots.
776         cur.execute("""delete from segment_utilization
777                        where snapshotid not in
778                            (select snapshotid from snapshots)""")
779
780         # Delete segments not referenced by any current snapshots.
781         cur.execute("""delete from segments where segmentid not in
782                            (select segmentid from segment_utilization)""")
783
784         # Delete dangling objects in the block_index table.
785         cur.execute("""delete from block_index
786                        where segmentid not in
787                            (select segmentid from segments)""")
788
789         # Remove sub-block signatures for deleted objects.
790         cur.execute("""delete from subblock_signatures
791                        where blockid not in
792                            (select blockid from block_index)""")
793
794     # Segment cleaning.
795     class SegmentInfo(Struct): pass
796
797     def get_segment_cleaning_list(self, age_boost=0.0):
798         """Return a list of all current segments with information for cleaning.
799
800         Return all segments which are currently known in the local database
801         (there might be other, older segments in the archive itself), and
802         return usage statistics for each to help decide which segments to
803         clean.
804
805         The returned list will be sorted by estimated cleaning benefit, with
806         segments that are best to clean at the start of the list.
807
808         If specified, the age_boost parameter (measured in days) will added to
809         the age of each segment, as a way of adjusting the benefit computation
810         before a long-lived snapshot is taken (for example, age_boost might be
811         set to 7 when cleaning prior to taking a weekly snapshot).
812         """
813
814         cur = self.cursor()
815         segments = []
816         cur.execute("""select segmentid, used, size, mtime,
817                        julianday('now') - mtime as age from segment_info
818                        where expire_time is null""")
819         for row in cur:
820             info = self.SegmentInfo()
821             info.id = row[0]
822             info.used_bytes = row[1]
823             info.size_bytes = row[2]
824             info.mtime = row[3]
825             info.age_days = row[4]
826
827             # If data is not available for whatever reason, treat it as 0.0.
828             if info.age_days is None:
829                 info.age_days = 0.0
830             if info.used_bytes is None:
831                 info.used_bytes = 0.0
832
833             # Benefit calculation: u is the estimated fraction of each segment
834             # which is utilized (bytes belonging to objects still in use
835             # divided by total size; this doesn't take compression or storage
836             # overhead into account, but should give a reasonable estimate).
837             #
838             # The total benefit is a heuristic that combines several factors:
839             # the amount of space that can be reclaimed (1 - u), an ageing
840             # factor (info.age_days) that favors cleaning old segments to young
841             # ones and also is more likely to clean segments that will be
842             # rewritten for long-lived snapshots (age_boost), and finally a
843             # penalty factor for the cost of re-uploading data (u + 0.1).
844             u = info.used_bytes / info.size_bytes
845             info.cleaning_benefit \
846                 = (1 - u) * (info.age_days + age_boost) / (u + 0.1)
847
848             segments.append(info)
849
850         segments.sort(cmp, key=lambda s: s.cleaning_benefit, reverse=True)
851         return segments
852
853     def mark_segment_expired(self, segment):
854         """Mark a segment for cleaning in the local database.
855
856         The segment parameter should be either a SegmentInfo object or an
857         integer segment id.  Objects in the given segment will be marked as
858         expired, which means that any future snapshots that would re-use those
859         objects will instead write out a new copy of the object, and thus no
860         future snapshots will depend upon the given segment.
861         """
862
863         if isinstance(segment, int):
864             id = segment
865         elif isinstance(segment, self.SegmentInfo):
866             id = segment.id
867         else:
868             raise TypeError("Invalid segment: %s, must be of type int or SegmentInfo, not %s" % (segment, type(segment)))
869
870         cur = self.cursor()
871         cur.execute("select max(snapshotid) from snapshots")
872         last_snapshotid = cur.fetchone()[0]
873         cur.execute("update segments set expire_time = ? where segmentid = ?",
874                     (last_snapshotid, id))
875         cur.execute("update block_index set expired = 0 where segmentid = ?",
876                     (id,))
877
878     def balance_expired_objects(self):
879         """Analyze expired objects in segments to be cleaned and group by age.
880
881         Update the block_index table of the local database to group expired
882         objects by age.  The exact number of buckets and the cutoffs for each
883         are dynamically determined.  Calling this function after marking
884         segments expired will help in the segment cleaning process, by ensuring
885         that when active objects from clean segments are rewritten, they will
886         be placed into new segments roughly grouped by age.
887         """
888
889         # The expired column of the block_index table is used when generating a
890         # new Cumulus snapshot.  A null value indicates that an object may be
891         # re-used.  Otherwise, an object must be written into a new segment if
892         # needed.  Objects with distinct expired values will be written into
893         # distinct segments, to allow for some grouping by age.  The value 0 is
894         # somewhat special in that it indicates any rewritten objects can be
895         # placed in the same segment as completely new objects; this can be
896         # used for very young objects which have been expired, or objects not
897         # expected to be encountered.
898         #
899         # In the balancing process, all objects which are not used in any
900         # current snapshots will have expired set to 0.  Objects which have
901         # been seen will be sorted by age and will have expired values set to
902         # 0, 1, 2, and so on based on age (with younger objects being assigned
903         # lower values).  The number of buckets and the age cutoffs is
904         # determined by looking at the distribution of block ages.
905
906         cur = self.cursor()
907
908         # Mark all expired objects with expired = 0; these objects will later
909         # have values set to indicate groupings of objects when repacking.
910         cur.execute("""update block_index set expired = 0
911                        where expired is not null""")
912
913         # We will want to aim for at least one full segment for each bucket
914         # that we eventually create, but don't know how many bytes that should
915         # be due to compression.  So compute the average number of bytes in
916         # each expired segment as a rough estimate for the minimum size of each
917         # bucket.  (This estimate could be thrown off by many not-fully-packed
918         # segments, but for now don't worry too much about that.)  If we can't
919         # compute an average, it's probably because there are no expired
920         # segments, so we have no more work to do.
921         cur.execute("""select avg(size) from segments
922                        where segmentid in
923                            (select distinct segmentid from block_index
924                             where expired is not null)""")
925         segment_size_estimate = cur.fetchone()[0]
926         if not segment_size_estimate:
927             return
928
929         # Next, extract distribution of expired objects (number and size) by
930         # age.  Save the timestamp for "now" so that the classification of
931         # blocks into age buckets will not change later in the function, after
932         # time has passed.  Set any timestamps in the future to now, so we are
933         # guaranteed that for the rest of this function, age is always
934         # non-negative.
935         cur.execute("select julianday('now')")
936         now = cur.fetchone()[0]
937
938         cur.execute("""update block_index set timestamp = ?
939                        where timestamp > ? and expired is not null""",
940                     (now, now))
941
942         cur.execute("""select round(? - timestamp) as age, count(*), sum(size)
943                        from block_index where expired = 0
944                        group by age order by age""", (now,))
945         distribution = cur.fetchall()
946
947         # Start to determine the buckets for expired objects.  Heuristics used:
948         #   - An upper bound on the number of buckets is given by the number of
949         #     segments we estimate it will take to store all data.  In fact,
950         #     aim for a couple of segments per bucket.
951         #   - Place very young objects in bucket 0 (place with new objects)
952         #     unless there are enough of them to warrant a separate bucket.
953         #   - Try not to create unnecessarily many buckets, since fewer buckets
954         #     will allow repacked data to be grouped based on spatial locality
955         #     (while more buckets will group by temporal locality).  We want a
956         #     balance.
957         MIN_AGE = 4
958         total_bytes = sum([i[2] for i in distribution])
959         target_buckets = 2 * (total_bytes / segment_size_estimate) ** 0.4
960         min_size = 1.5 * segment_size_estimate
961         target_size = max(2 * segment_size_estimate,
962                           total_bytes / target_buckets)
963
964         print("segment_size:", segment_size_estimate)
965         print("distribution:", distribution)
966         print("total_bytes:", total_bytes)
967         print("target_buckets:", target_buckets)
968         print("min, target size:", min_size, target_size)
969
970         # Chosen cutoffs.  Each bucket consists of objects with age greater
971         # than one cutoff value, but not greater than the next largest cutoff.
972         cutoffs = []
973
974         # Starting with the oldest objects, begin grouping together into
975         # buckets of size at least target_size bytes.
976         distribution.reverse()
977         bucket_size = 0
978         min_age_bucket = False
979         for (age, items, size) in distribution:
980             if bucket_size >= target_size \
981                 or (age < MIN_AGE and not min_age_bucket):
982                 if bucket_size < target_size and len(cutoffs) > 0:
983                     cutoffs.pop()
984                 cutoffs.append(age)
985                 bucket_size = 0
986
987             bucket_size += size
988             if age < MIN_AGE:
989                 min_age_bucket = True
990
991         # The last (youngest) bucket will be group 0, unless it has enough data
992         # to be of size min_size by itself, or there happen to be no objects
993         # less than MIN_AGE at all.
994         if bucket_size >= min_size or not min_age_bucket:
995             cutoffs.append(-1)
996         cutoffs.append(-1)
997
998         print("cutoffs:", cutoffs)
999
1000         # Update the database to assign each object to the appropriate bucket.
1001         cutoffs.reverse()
1002         for i in range(len(cutoffs)):
1003             cur.execute("""update block_index set expired = ?
1004                            where round(? - timestamp) > ?
1005                              and expired is not null""",
1006                         (i, now, cutoffs[i]))