code.grnet.gr Git - pithos/blob - tools/pithos-sync

   1 #!/usr/bin/env python
   2
   3 # Copyright 2011 GRNET S.A. All rights reserved.
   4 #
   5 # Redistribution and use in source and binary forms, with or
   6 # without modification, are permitted provided that the following
   7 # conditions are met:
   8 #
   9 #   1. Redistributions of source code must retain the above
  10 #      copyright notice, this list of conditions and the following
  11 #      disclaimer.
  12 #
  13 #   2. Redistributions in binary form must reproduce the above
  14 #      copyright notice, this list of conditions and the following
  15 #      disclaimer in the documentation and/or other materials
  16 #      provided with the distribution.
  17 #
  18 # THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  19 # OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  20 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  21 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  22 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  23 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  24 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  25 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  26 # AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  27 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  28 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  29 # POSSIBILITY OF SUCH DAMAGE.
  30 #
  31 # The views and conclusions contained in the software and
  32 # documentation are those of the authors and should not be
  33 # interpreted as representing official policies, either expressed
  34 # or implied, of GRNET S.A.
  35
  36 import os
  37 import sqlite3
  38 import sys
  39 import shutil
  40 import pickle
  41
  42 from lib import transfer
  43 from lib.client import Pithos_Client, Fault
  44 from lib.hashmap import HashMap, merkle
  45 from lib.util import get_user, get_auth, get_server
  46
  47
  48 DEFAULT_CONTAINER = 'pithos'
  49
  50 def get_container():
  51     try:
  52         return os.environ['PITHOS_SYNC_CONTAINER']
  53     except KeyError:
  54         return DEFAULT_CONTAINER
  55
  56
  57 def create_dir(path):
  58     if not os.path.exists(path):
  59         os.makedirs(path)
  60     if not os.path.isdir(path):
  61         raise RuntimeError("Cannot open '%s'" % (path,))
  62
  63
  64 def copy_file(src, dst):
  65     print '***', 'COPYING', src, dst
  66     path = os.path.dirname(dst)
  67     create_dir(path)
  68     shutil.copyfile(src, dst)
  69
  70
  71 client = None
  72 conf = None
  73 confdir = None
  74 trash = None
  75 lstate = None
  76 cstate = None
  77 rstate = None
  78
  79
  80 class Trash(object):
  81     def __init__(self):
  82         self.path = os.path.join(confdir, 'trash')
  83         create_dir(self.path)
  84
  85         dbpath = os.path.join(confdir, 'trash.db')
  86         self.conn = sqlite3.connect(dbpath)
  87         sql = '''CREATE TABLE IF NOT EXISTS files (
  88                     path TEXT PRIMARY KEY, hash TEXT)'''
  89         self.conn.execute(sql)
  90         self.conn.commit()
  91
  92     def put(self, fullpath, path, hash):
  93         copy_file(fullpath, os.path.join(self.path, path))
  94         os.remove(fullpath)
  95         sql = 'INSERT OR REPLACE INTO files VALUES (?, ?)'
  96         self.conn.execute(sql, (path, hash))
  97         self.conn.commit()
  98
  99     def search(self, hash):
 100         sql = 'SELECT path FROM files WHERE hash = ?'
 101         ret = self.conn.execute(sql, (hash,)).fetchone()
 102         return ret[0] if ret else None
 103
 104     def empty(self):
 105         sql = 'DELETE FROM files'
 106         self.conn.execute(sql)
 107         self.conn.commit()
 108         shutil.rmtree(self.path)
 109
 110     def fullpath(self, path):
 111         return os.path.join(self.path, path)
 112
 113
 114 class LocalState(object):
 115     def __init__(self, path):
 116         self.path = path
 117
 118         dbpath = os.path.join(confdir, 'state.db')
 119         self.conn = sqlite3.connect(dbpath)
 120         sql = '''CREATE TABLE IF NOT EXISTS files (
 121                     path TEXT PRIMARY KEY, hash TEXT)'''
 122         self.conn.execute(sql)
 123         self.conn.commit()
 124
 125     def get(self, path):
 126         sql = 'SELECT hash FROM files WHERE path = ?'
 127         ret = self.conn.execute(sql, (path,)).fetchone()
 128         return ret[0] if ret else 'DEL'
 129
 130     def put(self, path, hash):
 131         sql = 'INSERT OR REPLACE INTO files VALUES (?, ?)'
 132         self.conn.execute(sql, (path, hash))
 133         self.conn.commit()
 134
 135     def search(self, hash):
 136         sql = 'SELECT path FROM files WHERE hash = ?'
 137         ret = self.conn.execute(sql, (hash,)).fetchone()
 138         return ret[0] if ret else None
 139
 140     def fullpath(self, path):
 141         return os.path.join(self.path, path)
 142
 143
 144 class CurrentState(object):
 145     def __init__(self, path):
 146         self.path = path
 147
 148     def get(self, path):
 149         fullpath = os.path.join(self.path, path)
 150         if os.path.exists(fullpath):
 151             if os.path.isdir(fullpath):
 152                 return 'DIR'
 153             else:
 154                 return merkle(fullpath, conf['blocksize'], conf['blockhash'])
 155         else:
 156             return 'DEL'
 157
 158     def fullpath(self, path):
 159         return os.path.join(self.path, path)
 160
 161
 162 class RemoteState(object):
 163     def __init__(self, client):
 164         self.client = client
 165         self.container = get_container()
 166
 167     def get(self, path):
 168         try:
 169             meta = self.client.retrieve_object_metadata(self.container, path)
 170         except Fault:
 171             return 'DEL'
 172         if meta.get('content-type', None) == 'application/directory':
 173             return 'DIR'
 174         else:
 175             return meta['x-object-hash']
 176
 177
 178 def update_local(path, S):
 179     # XXX If something is already here, put it in trash and delete it.
 180     # XXX If we have a directory already here, put all files in trash.
 181     fullpath = cstate.fullpath(path)
 182     if S == 'DEL':
 183         trash.put(fullpath, path, S)
 184     elif S == 'DIR':
 185         if os.path.exists(fullpath):
 186             trash.put(fullpath, path, S)
 187         # XXX Strip trailing slash (or escape).
 188         os.mkdir(fullpath)
 189     else:
 190         # First, search for local copy
 191         file = lstate.search(S)
 192         if file:
 193             copy_file(lstate.fullpath(file), fullpath)
 194         else:
 195             # Search for copy in trash
 196             file = trash.search(S)
 197             if file:
 198                 # XXX Move from trash (not copy).
 199                 copy_file(trash.fullpath(file), fullpath)
 200             else:
 201                 # Download
 202                 transfer.download(client, get_container(), path, fullpath)
 203         assert cstate.get(path) == S
 204
 205
 206 def update_remote(path, S):
 207     fullpath = cstate.fullpath(path)
 208     if S == 'DEL':
 209         client.delete_object(get_container(), path)
 210     elif S == 'DIR':
 211         client.create_directory_marker(get_container(), path)
 212     else:
 213         prefix, name = os.path.split(path)
 214         if prefix:
 215             prefix += '/'
 216         transfer.upload(client, fullpath, get_container(), prefix, name)
 217         assert rstate.get(path) == S
 218
 219
 220 def resolve_conflict(path):
 221     # XXX Check if this works with dirs.
 222     fullpath = cstate.fullpath(path)
 223     if os.path.exists(fullpath):
 224         os.rename(fullpath, fullpath + '.local')
 225
 226
 227 def sync(path):
 228     L = lstate.get(path)
 229     C = cstate.get(path)
 230     R = rstate.get(path)
 231
 232     if C == L:
 233         # No local changes
 234         if R != L:
 235             update_local(path, R)
 236             lstate.put(path, R)
 237         return
 238
 239     if R == L:
 240         # No remote changes
 241         if C != L:
 242             update_remote(path, C)
 243             lstate.put(path, C)
 244         return
 245
 246     # At this point both local and remote states have changes since last sync
 247
 248     if C == R:
 249         # We were lucky, both had the same change
 250         lstate.put(path, R)
 251     else:
 252         # Conflict, try to resolve it
 253         resolve_conflict(path)
 254         update_local(path, R)
 255         lstate.put(path, R)
 256
 257
 258 def walk(dir):
 259     pending = ['']
 260
 261     while pending:
 262         dirs = set()
 263         files = set()
 264         root = pending.pop(0)
 265         if root:
 266             yield root
 267
 268         dirpath = os.path.join(dir, root)
 269         if os.path.exists(dirpath):
 270             for filename in os.listdir(dirpath):
 271                 path = os.path.join(root, filename)
 272                 if os.path.isdir(os.path.join(dir, path)):
 273                     dirs.add(path)
 274                 else:
 275                     files.add(path)
 276
 277         for object in client.list_objects(get_container(), prefix=root,
 278                                             delimiter='/', format='json'):
 279             # XXX Check subdirs.
 280             if 'subdir' in object:
 281                 continue
 282             name = str(object['name'])
 283             if object['content_type'] == 'application/directory':
 284                 dirs.add(name)
 285             else:
 286                 files.add(name)
 287
 288         pending += sorted(dirs)
 289         for path in files:
 290             yield path
 291
 292
 293 def main():
 294     global client, conf, confdir, trash, lstate, cstate, rstate
 295
 296     if len(sys.argv) != 2:
 297         print 'syntax: %s <dir>' % sys.argv[0]
 298         sys.exit(1)
 299
 300     dir = sys.argv[1]
 301     client = Pithos_Client(get_server(), get_auth(), get_user())
 302
 303     container = get_container()
 304     try:
 305         meta = client.retrieve_container_metadata(container)
 306     except Fault:
 307         raise RuntimeError("Cannot open container '%s'" % (container,))
 308
 309     conf = {'local': dir,
 310             'remote': container,
 311             'blocksize': int(meta['x-container-block-size']),
 312             'blockhash': meta['x-container-block-hash']}
 313     confdir = os.path.expanduser('~/.pithos-sync/')
 314
 315     conffile = os.path.join(confdir, 'config')
 316     if os.path.isfile(conffile):
 317         try:
 318             if (conf != pickle.loads(open(conffile, 'rb').read())):
 319                 raise ValueError
 320         except:
 321             shutil.rmtree(confdir)
 322     create_dir(confdir)
 323
 324     trash = Trash()
 325     lstate = LocalState(dir)
 326     cstate = CurrentState(dir)
 327     rstate = RemoteState(client)
 328
 329     for path in walk(dir):
 330         print 'Syncing', path
 331         sync(path)
 332
 333     f = open(conffile, 'wb')
 334     f.write(pickle.dumps(conf))
 335     f.close()
 336
 337     trash.empty()
 338
 339
 340 if __name__ == '__main__':
 341     main()