X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=cleaner%2Fcleaner;h=336537da3b50c7bacc5539c65a06893de6e02387;hb=1d2357b567318ccc6172f4b9edbbadbeba713086;hp=445bdeb4cd59e5e246e9824d35de272fa323a285;hpb=885814beba8d6191c042d56e7734bda0e680ac6a;p=bluesky.git

diff --git a/cleaner/cleaner b/cleaner/cleaner
index 445bdeb..336537d 100755
--- a/cleaner/cleaner
+++ b/cleaner/cleaner
@@ -10,6 +10,7 @@
 
 import base64, os, re, struct, sys
 import boto
+from boto.s3.key import Key
 
 # The BlueSky 'struct cloudlog_header' data type.
 HEADER_FORMAT = '<4sb16sQIII'
@@ -49,6 +50,60 @@ class FileBackend:
         fp.write(data)
         fp.close()
 
+    def delete(self, filename):
+        os.unlink(os.path.join(self.path, 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 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)
 
@@ -229,6 +284,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
 
@@ -268,8 +324,10 @@ class InodeMap:
         print
         print "Segment utilizations:"
         for (s, u) in sorted(util.segments.items()):
-            #if u[1] > 0:
             print "%s: %s %s" % (s, u, float(u[1]) / u[0])
+            if u[1] == 0:
+                print "Deleting..."
+                backend.delete(s)
 
         self.inodes = inodes
         self.util = util
@@ -341,7 +399,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.95 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)
@@ -375,13 +433,14 @@ 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
 
-    log_dir = LogDirectory(backend, 1)
+    log_dir = LogDirectory(backend, 0)
     run_cleaner(backend, imap, log_dir)
     imap.write(backend, log_dir)
     log_dir.close_all()