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:
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')
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
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):
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 "<Item ty=%s location=%s size=%d id=%s...>" % (self.type, self.location, self.size, base64.b16encode(self.id).lower()[0:8])
+ return "<Item%s ty=%s location=%s size=%d id=%s...>" % (self.encrypted and '$' or '', self.type, self.location, self.size, base64.b16encode(self.id).lower()[0:8])
@staticmethod
def random_id():
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
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:
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
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):
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()):
# 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)