X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=cleaner%2Fcleaner;h=95bbb08c4ac01e98ebaad135487e9c7f567240f5;hb=c1f4367b3c65b01562bc6484226dff0135cd4780;hp=336537da3b50c7bacc5539c65a06893de6e02387;hpb=1d2357b567318ccc6172f4b9edbbadbeba713086;p=bluesky.git diff --git a/cleaner/cleaner b/cleaner/cleaner index 336537d..95bbb08 100755 --- a/cleaner/cleaner +++ b/cleaner/cleaner @@ -13,8 +13,10 @@ import boto from boto.s3.key import Key # The BlueSky 'struct cloudlog_header' data type. -HEADER_FORMAT = '<4sb16sQIII' -HEADER_MAGIC = 'AgI-' +HEADER_FORMAT = '<4s48sb16sQIII' +HEADER_CRYPTBYTES = 48 +HEADER_MAGIC1 = 'AgI-' # Unencrypted data +HEADER_MAGIC2 = 'AgI=' # Encrypted data HEADER_SIZE = struct.calcsize(HEADER_FORMAT) class ITEM_TYPE: @@ -41,9 +43,14 @@ class FileBackend: return [(f, os.stat(os.path.join(self.path, f)).st_size) for f in files] - def read(self, filename): + def read(self, filename, offset=0, length=None): fp = open(os.path.join(self.path, filename), 'rb') - return fp.read() + if offset > 0: + fp.seek(offset) + if legnth is None: + return fp.read() + else: + return fp.read(length) def write(self, filename, data): fp = open(os.path.join(self.path, filename), 'wb') @@ -76,10 +83,15 @@ class S3Backend: files.append((k.key, k.size)) return files - def read(self, filename): + def read(self, filename, offset=0, length=None): if filename in self.cache: fp = open(os.path.join(self.cachedir, filename), 'rb') - return fp.read() + if offset > 0: + fp.seek(offset) + if length is None: + return fp.read() + else: + return fp.read(length) else: k = Key(self.bucket) k.key = self.path + filename @@ -88,6 +100,10 @@ class S3Backend: fp.write(data) fp.close() self.cache[filename] = True + if offset > 0: + data = data[offset:] + if length is not None: + data = data[0:length] return data def write(self, filename, data): @@ -114,8 +130,12 @@ class S3Backend: class LogItem: """In-memory representation of a single item stored in a log file.""" + def __init__(self): + self.cryptkeys = '\0' * HEADER_CRYPTBYTES + self.encrypted = False + def __str__(self): - return "" % (self.type, self.location, self.size, base64.b16encode(self.id).lower()[0:8]) + return "" % (self.encrypted and '$' or '', self.type, self.location, self.size, base64.b16encode(self.id).lower()[0:8]) @staticmethod def random_id(): @@ -131,8 +151,13 @@ class LogItem: link_ids = ''.join(link_ids) link_locs = ''.join(link_locs) + if self.encrypted: + magic = HEADER_MAGIC2 + else: + magic = HEADER_MAGIC1 header = struct.pack(HEADER_FORMAT, - HEADER_MAGIC, ord(self.type), self.id, self.inum, + 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 @@ -211,28 +236,30 @@ 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: + if header[0] not in (HEADER_MAGIC1, HEADER_MAGIC2): 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] + if header[0] == HEADER_MAGIC2: item.encrypted = True + 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: @@ -250,7 +277,7 @@ def load_item(backend, location): The elements of the tuple are (directory, sequence, offset, size).""" filename = backend.loc_to_name((location[0], location[1])) - data = backend.read(filename)[location[2] : location[2] + location[3]] + data = backend.read(filename, location[2], location[3]) item = parse_item(data) item.location = location return item @@ -268,8 +295,8 @@ 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]) - if header[0] != HEADER_MAGIC: + size = HEADER_SIZE + sum(header[5:8]) + if header[0] not in (HEADER_MAGIC1, HEADER_MAGIC2): print "Bad header magic!" break if size + offset > len(data): @@ -395,7 +422,7 @@ def rewrite_inode(backend, inode_map, inum, log, copy_data=True): log.write(inode, 1) inode_map.mark_updated(inum) -def run_cleaner(backend, inode_map, log): +def run_cleaner(backend, inode_map, log, repack_inodes=False): # 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()): @@ -410,7 +437,10 @@ def run_cleaner(backend, inode_map, log): # Given that list of segments to clean, scan through those segments to find # data which is still live and mark relevant inodes as needing to be # rewritten. - dirty_inodes = set() + if repack_inodes: + dirty_inodes = set(inode_map.inodes) + else: + dirty_inodes = set() dirty_inode_data = set() for s in inode_map.obsolete_segments: filename = backend.loc_to_name(s)