X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=cleaner%2Fcleaner;h=1e564e83c45e5c79c6f009716901589af475bfb2;hb=524d3c03d2f3076a349768c177a0e523a7fb333e;hp=b018d0ddccf3f0c3bda24e20a5ce77a558e8b874;hpb=2f883084c4393dc7b730df2e778b9f2fc14f70e0;p=bluesky.git diff --git a/cleaner/cleaner b/cleaner/cleaner index b018d0d..1e564e8 100755 --- a/cleaner/cleaner +++ b/cleaner/cleaner @@ -10,9 +10,11 @@ import base64, os, re, struct, sys import boto +from boto.s3.key import Key # The BlueSky 'struct cloudlog_header' data type. -HEADER_FORMAT = '<4sb16sQIII' +HEADER_FORMAT = '<4s48sb16sQIII' +HEADER_CRYPTBYTES = 48 HEADER_MAGIC = 'AgI-' HEADER_SIZE = struct.calcsize(HEADER_FORMAT) @@ -59,9 +61,63 @@ class FileBackend: m = re.match(r"^log-(\d+)-(\d+)$", name) if m: return (int(m.group(1)), int(m.group(2))) +class S3Backend: + """An interface to BlueSky where the log segments are on in Amazon S3.""" + + def __init__(self, bucket, path='', cachedir="."): + self.conn = boto.connect_s3(is_secure=False) + self.bucket = self.conn.get_bucket(bucket) + self.path = path + self.cachedir = cachedir + self.cache = {} + + def list(self): + files = [] + for k in self.bucket.list(self.path + 'log-'): + files.append((k.key, k.size)) + return files + + def read(self, filename): + if filename in self.cache: + fp = open(os.path.join(self.cachedir, filename), 'rb') + return fp.read() + else: + k = Key(self.bucket) + k.key = self.path + filename + data = k.get_contents_as_string() + fp = open(os.path.join(self.cachedir, filename), 'wb') + fp.write(data) + fp.close() + self.cache[filename] = True + return data + + def write(self, filename, data): + k = Key(self.bucket) + k.key = self.path + filename + k.set_contents_from_string(data) + if filename in self.cache: + del self.cache[filename] + + def delete(self, filename): + k = Key(self.bucket) + k.key = self.path + filename + k.delete() + if filename in self.cache: + del self.cache[filename] + + def loc_to_name(self, location): + return "log-%08d-%08d" % (location) + + def name_to_loc(self, name): + m = re.match(r"^log-(\d+)-(\d+)$", name) + if m: return (int(m.group(1)), int(m.group(2))) + class LogItem: """In-memory representation of a single item stored in a log file.""" + def __init__(self): + self.cryptkeys = '\0' * HEADER_CRYPTBYTES + def __str__(self): return "" % (self.type, self.location, self.size, base64.b16encode(self.id).lower()[0:8]) @@ -80,7 +136,8 @@ class LogItem: link_locs = ''.join(link_locs) header = struct.pack(HEADER_FORMAT, - HEADER_MAGIC, ord(self.type), self.id, self.inum, + HEADER_MAGIC, self.cryptkeys, + ord(self.type), self.id, self.inum, len(self.data), len(link_ids), len(link_locs)) return header + self.data + link_ids + link_locs @@ -159,28 +216,29 @@ class UtilizationTracker: def parse_item(data): if len(data) < HEADER_SIZE: return header = struct.unpack_from(HEADER_FORMAT, data, 0) - size = HEADER_SIZE + sum(header[4:7]) + size = HEADER_SIZE + sum(header[5:8]) if header[0] != HEADER_MAGIC: print "Bad header magic!" return if len(data) != size: - print "Item size does not match!" + print "Item size does not match: %d != %d" % (size, len(data)) return item = LogItem() - item.id = header[2] - item.inum = header[3] + item.cryptkeys = header[1] + item.id = header[3] + item.inum = header[4] item.location = None - item.type = chr(header[1]) + item.type = chr(header[2]) item.size = size - item.data = data[HEADER_SIZE : HEADER_SIZE + header[4]] + item.data = data[HEADER_SIZE : HEADER_SIZE + header[5]] links = [] - link_ids = data[HEADER_SIZE + header[4] - : HEADER_SIZE + header[4] + header[5]] - link_locs = data[HEADER_SIZE + header[4] + header[5] - : HEADER_SIZE + sum(header[4:7])] + link_ids = data[HEADER_SIZE + header[5] + : HEADER_SIZE + header[5] + header[6]] + link_locs = data[HEADER_SIZE + header[5] + header[6] + : HEADER_SIZE + sum(header[5:8])] for i in range(len(link_ids) // 16): id = link_ids[16*i : 16*i + 16] if id == '\0' * 16: @@ -216,7 +274,7 @@ def parse_log(data, location=None): offset = 0 while len(data) - offset >= HEADER_SIZE: header = struct.unpack_from(HEADER_FORMAT, data, offset) - size = HEADER_SIZE + sum(header[4:7]) + size = HEADER_SIZE + sum(header[5:8]) if header[0] != HEADER_MAGIC: print "Bad header magic!" break @@ -232,6 +290,7 @@ def parse_log(data, location=None): def load_checkpoint_record(backend): for (log, size) in reversed(backend.list()): for item in reversed(list(parse_log(backend.read(log), log))): + print item if item.type == ITEM_TYPE.CHECKPOINT: return item @@ -346,7 +405,7 @@ def run_cleaner(backend, inode_map, log): # Determine which segments are poorly utilized and should be cleaned. We # need better heuristics here. for (s, u) in sorted(inode_map.util.segments.items()): - if float(u[1]) / u[0] < 0.6 and u[1] > 0: + if (float(u[1]) / u[0] < 0.6 or u[1] < 32768) and u[1] > 0: print "Should clean segment", s loc = backend.name_to_loc(s) if s: inode_map.obsolete_segments.add(loc) @@ -380,8 +439,9 @@ def run_cleaner(backend, inode_map, log): rewrite_inode(backend, inode_map, i, log, i in dirty_inode_data) if __name__ == '__main__': - backend = FileBackend(".") + backend = S3Backend("mvrable-bluesky", cachedir=".") chkpt = load_checkpoint_record(backend) + print backend.list() imap = InodeMap() imap.build(backend, chkpt) print chkpt