+++ /dev/null
-#!/usr/bin/python
-
-#
-# Ceph - scalable distributed file system
-#
-# Copyright (C) 2011 New Dream Network
-#
-# This is free software; you can redistribute it and/or
-# modify it under the terms of the GNU Lesser General Public
-# License version 2.1, as published by the Free Software
-# Foundation. See file COPYING.
-#
-
-"""
-boto_tool.py: s3cmd-like tool for operating on s3
-
-A lot of common s3 clients can't handle weird names.
-But this little script can do it!
-"""
-
-from boto.s3.connection import OrdinaryCallingFormat
-from boto.s3.connection import S3Connection
-from boto.s3.key import Key
-from optparse import OptionParser
-from sys import stderr
-import boto
-import os
-import string
-import sys
-
-global conn
-
-def getenv(a):
- if os.environ.has_key(a):
- return os.environ[a]
- else:
- return None
-
-def strip_prefix(prefix, s):
- if not (s[0:len(prefix)] == prefix):
- return None
- return s[len(prefix):]
-
-def list_all_buckets(args):
- parser = OptionParser()
- parser.add_option("-v", "--verbose", action="store_true",
- dest="verbose", default=False, help="verbose output")
- (opts, args) = parser.parse_args(args)
- blrs = conn.get_all_buckets()
- for b in blrs:
- if (opts.verbose):
- print b.__dict__
- else:
- print b
-
-def mkbucket(args):
- if (len(args) < 1):
- print "must give an argument to mkbucket"
- return 255
- bucket_name = args[0]
- print "creating bucket '%s' ..." % bucket_name
- bucket = conn.create_bucket(bucket_name)
- print "done."
- return 0
-
-def rmbucket(args):
- if (len(args) < 1):
- print "must give an argument to rmbucket"
- return 255
- bucket = conn.get_bucket(args[0])
- print "deleting bucket '%s' ..." % args[0]
- bucket.delete()
- print "done."
- return 0
-
-def bucket_exists(args):
- if (len(args) < 1):
- print "must give an argument to exists"
- return 255
- bucket = conn.get_bucket(opts.bucket_exists)
- if (bucket == None):
- print "bucket '%s' does not exist"
- return 1
- else:
- print "found bucket '%s'."
- return 0
-
-def put_obj(bucket_name, args):
- parser = OptionParser()
- parser.add_option("-f", "--filename", dest="filename",
- help="file name (default stdin)")
- (opts, args) = parser.parse_args(args)
- if (len(args) < 1):
- print "put requires an argument: the object name"
- return 255
- obj_name = args[0]
- print "uploading to bucket: '%s', object name: '%s'" % (bucket_name, obj_name)
- bucket = conn.get_bucket(bucket_name)
- k = Key(bucket)
- k.key = obj_name
- if (opts.filename == None):
- print "sorry, no support for put-from-stdin yet. use -f"
- return 255
- else:
- k.set_contents_from_filename(opts.filename)
-
-def put_obj_acl(bucket_name, args):
- parser = OptionParser()
- parser.add_option("-f", "--filename", dest="filename",
- help="file name (default stdin)")
- (opts, args) = parser.parse_args(args)
- if (len(args) < 1):
- print "put_acl requires an argument: the object name"
- return 255
- obj_name = args[0]
- print "uploading object ACL to bucket: '%s', object name: '%s'" \
- % (bucket_name, obj_name)
- bucket = conn.get_bucket(bucket_name)
- k = Key(bucket)
- k.key = obj_name
- if (opts.filename == None):
- print "sorry, no support for put-from-stdin yet. use -f"
- return 255
- else:
- f = open(opts.filename, "r")
- try:
- xml = f.read()
- finally:
- f.close()
- k.set_xml_acl(xml)
-
-def get_obj(bucket_name, args):
- parser = OptionParser()
- parser.add_option("-f", "--filename", dest="filename",
- help="file name (default stdin)")
- (opts, args) = parser.parse_args(args)
- if (len(args) < 1):
- print "get requires an argument: the object name"
- return 255
- obj_name = args[0]
- print "downloading from bucket: '%s', object name: '%s'" % (bucket_name, obj_name)
- bucket = conn.get_bucket(bucket_name)
- k = Key(bucket)
- k.key = obj_name
- if (opts.filename == None):
- k.get_contents_to_file(sys.stdout)
- else:
- k.get_contents_to_filename(opts.filename)
-
-def get_obj_acl(bucket_name, args):
- parser = OptionParser()
- parser.add_option("-f", "--filename", dest="filename",
- help="file name (default stdin)")
- (opts, args) = parser.parse_args(args)
- if (len(args) < 1):
- print "get_acl requires an argument: the object name to get the acl for"
- return 255
- obj_name = args[0]
- print "downloading object acl from bucket: '%s', object name: '%s'" % \
- (bucket_name, obj_name)
- bucket = conn.get_bucket(bucket_name)
- k = Key(bucket)
- k.key = obj_name
- xml = k.get_xml_acl()
- if (opts.filename == None):
- print xml
- else:
- f = open(opts.filename, "w")
- try:
- f.write(xml)
- finally:
- f.close()
-
-def list_obj(bucket_name, args):
- if (len(args) < 1):
- prefix = None
- else:
- prefix = args[0]
- bucket = conn.get_bucket(bucket_name)
- for key in bucket.list(prefix = prefix):
- print key.name
-
-def rm_obj(bucket_name, args):
- if (len(args) < 1):
- obj_name = None
- else:
- obj_name = args[0]
- print "removing from bucket: '%s', object name: '%s'" % (bucket_name, obj_name)
- bucket = conn.get_bucket(bucket_name)
- bucket.delete_key(obj_name)
- print "done."
-
-def head_obj(bucket_name, args):
- parser = OptionParser()
- parser.add_option("-f", "--filename", dest="filename",
- help="file name (default stdin)")
- (opts, args) = parser.parse_args(args)
- if (len(args) < 1):
- print "get requires an argument: the object name"
- return 255
- obj_name = args[0]
- print "downloading from bucket: '%s', object name: '%s'" % (bucket_name, obj_name)
- bucket = conn.get_bucket(bucket_name)
- k = bucket.get_key(k, obj_name)
- print k
-
-def usage():
- print """
-boto_tool.py
- ./boto_tool.py -h
- ./boto_tool.py --help
- Show this help
- ./boto_tool.py <host> ls
- Lists all buckets in a host
- ./boto_tool.py <host> <bucket> ls
- Lists all objects in a bucket
- ./boto_tool.py <host> <bucket> ls <prefix>
- Lists all objects in a bucket that have a given prefix
- ./boto_tool.py <host> mkbucket <bucket>
- Create a new bucket
- ./boto_tool.py <host> rmbucket <bucket>
- Remove a bucket
- ./boto_tool.py <host> exists <bucket>
- Tests if a bucket exists
- ./boto_tool.py <host> <bucket> put <object> [opts]
- Upload an object
- opts:
- -f filename file name (default stdin)
- ./boto_tool.py <host> <bucket> get <object> [opts]
- Gets an object
- opts:
- -f filename file name (default stdout)
- ./boto_tool.py <host> <bucket> head <object> [opts]
- Gets the headers of an object
-"""
-
-if (len(sys.argv) < 3):
- usage()
- sys.exit(255)
-
-if (sys.argv[1] == "-h") or (sys.argv[1] == "--help"):
- usage()
- sys.exit(0)
-
-host = sys.argv[1]
-
-conn = S3Connection(calling_format=OrdinaryCallingFormat(), is_secure=False,
- host = host,
- aws_access_key_id=getenv("AKEY"),
- aws_secret_access_key=getenv("SKEY"))
-
-if (sys.argv[2] == "ls"):
- sys.exit(list_all_buckets(sys.argv[3:]))
-elif (sys.argv[2] == "mkbucket"):
- sys.exit(mkbucket(sys.argv[3:]))
-elif (sys.argv[2] == "rmbucket"):
- sys.exit(rmbucket(sys.argv[3:]))
-elif (sys.argv[2] == "exists"):
- sys.exit(bucket_exists(sys.argv[3:]))
-else:
- # bucket operations
- wb = strip_prefix("bucket:", sys.argv[2])
- if (wb):
- bucket_name = wb
- else:
- bucket_name = sys.argv[2]
- if (len(sys.argv) < 4):
- print "too few arguments. -h for help."
- sys.exit(255)
- if (sys.argv[3] == "put"):
- sys.exit(put_obj(bucket_name, sys.argv[4:]))
- if (sys.argv[3] == "putacl"):
- sys.exit(put_obj_acl(bucket_name, sys.argv[4:]))
- elif (sys.argv[3] == "get"):
- sys.exit(get_obj(bucket_name, sys.argv[4:]))
- elif (sys.argv[3] == "getacl"):
- sys.exit(get_obj_acl(bucket_name, sys.argv[4:]))
- elif (sys.argv[3] == "ls"):
- sys.exit(list_obj(bucket_name, sys.argv[4:]))
- elif (sys.argv[3] == "rm"):
- sys.exit(rm_obj(bucket_name, sys.argv[4:]))
- elif (sys.argv[3] == "head"):
- sys.exit(head_obj(bucket_name, sys.argv[4:]))
- else:
- print "unknown operation on bucket"
- sys.exit(255)
-
+++ /dev/null
-#!/usr/bin/env python
-
-#
-# Ceph - scalable distributed file system
-#
-# Copyright (C) 2011 New Dream Network
-#
-# This is free software; you can redistribute it and/or
-# modify it under the terms of the GNU Lesser General Public
-# License version 2.1, as published by the Free Software
-# Foundation. See file COPYING.
-#
-
-"""
-obsync.py: the object synchronizer
-"""
-
-from boto.s3.connection import OrdinaryCallingFormat
-from boto.s3.connection import S3Connection
-from boto.s3.key import Key
-from optparse import OptionParser
-from sys import stderr
-from lxml import etree
-import base64
-import boto
-import errno
-import hashlib
-import mimetypes
-import os
-from StringIO import StringIO
-import rados
-import re
-import shutil
-import string
-import sys
-import tempfile
-import traceback
-
-global opts
-global xuser
-
-###### Constants classes #######
-RGW_META_BUCKET_NAME = ".rgw"
-
-###### Exception classes #######
-class LocalFileIsAcl(Exception):
- pass
-
-class InvalidLocalName(Exception):
- pass
-
-class NonexistentStore(Exception):
- pass
-
-###### Helper functions #######
-def mkdir_p(path):
- try:
- os.makedirs(path)
- except OSError, exc:
- if exc.errno != errno.EEXIST:
- raise
- if (not os.path.isdir(path)):
- raise
-
-def bytes_to_str(b):
- return ''.join(["%02x"% ord(x) for x in b]).strip()
-
-def get_md5(f, block_size=2**20):
- md5 = hashlib.md5()
- while True:
- data = f.read(block_size)
- if not data:
- break
- md5.update(data)
- return "%s" % md5.hexdigest()
-
-def strip_prefix(prefix, s):
- if not (s[0:len(prefix)] == prefix):
- return None
- return s[len(prefix):]
-
-def etag_to_md5(etag):
- if (etag[:1] == '"'):
- start = 1
- else:
- start = 0
- if (etag[-1:] == '"'):
- end = -1
- else:
- end = None
- return etag[start:end]
-
-def getenv(a, b):
- if os.environ.has_key(a):
- return os.environ[a]
- elif b and os.environ.has_key(b):
- return os.environ[b]
- else:
- return None
-
-# Escaping functions.
-#
-# Valid names for local files are a little different than valid object
-# names for S3. So these functions are needed to translate.
-#
-# Basically, in local names, every sequence starting with a dollar sign is
-# reserved as a special escape sequence. If you want to create an S3 object
-# with a dollar sign in the name, the local file should have a double dollar
-# sign ($$).
-#
-# TODO: translate local files' control characters into escape sequences.
-# Most S3 clients (boto included) cannot handle control characters in S3 object
-# names.
-# TODO: check for invalid utf-8 in local file names. Ideally, escape it, but
-# if not, just reject the local file name. S3 object names must be valid
-# utf-8.
-#
-# ---------- -----------
-# In S3 Locally
-# ---------- -----------
-# foo/ foo$slash
-#
-# $money $$money
-#
-# obj-with-acl obj-with-acl
-# .obj-with-acl$acl
-def s3_name_to_local_name(s3_name):
- s3_name = re.sub(r'\$', "$$", s3_name)
- if (s3_name[-1:] == "/"):
- s3_name = s3_name[:-1] + "$slash"
- return s3_name
-
-def local_name_to_s3_name(local_name):
- if local_name.find(r'$acl') != -1:
- raise LocalFileIsAcl()
- local_name = re.sub(r'\$slash', "/", local_name)
- mre = re.compile("[$][^$]")
- if mre.match(local_name):
- raise InvalidLocalName("Local name contains a dollar sign escape \
-sequence we don't understand.")
- local_name = re.sub(r'\$\$', "$", local_name)
- return local_name
-
-def get_local_acl_file_name(local_name):
- if local_name.find(r'$acl') != -1:
- raise LocalFileIsAcl()
- return os.path.dirname(local_name) + "/." + \
- os.path.basename(local_name) + "$acl"
-
-###### ACLs #######
-
-# for buckets: allow list
-# for object: allow grantee to read object data and metadata
-READ = 1
-
-# for buckets: allow create, overwrite, or deletion of any object in the bucket
-WRITE = 2
-
-# for buckets: allow grantee to read the bucket ACL
-# for objects: allow grantee to read the object ACL
-READ_ACP = 4
-
-# for buckets: allow grantee to write the bucket ACL
-# for objects: allow grantee to write the object ACL
-WRITE_ACP = 8
-
-# all of the above
-FULL_CONTROL = READ | WRITE | READ_ACP | WRITE_ACP
-
-ACL_TYPE_CANON_USER = "canon:"
-ACL_TYPE_EMAIL_USER = "email:"
-ACL_TYPE_GROUP = "group:"
-ALL_ACL_TYPES = [ ACL_TYPE_CANON_USER, ACL_TYPE_EMAIL_USER, ACL_TYPE_GROUP ]
-
-S3_GROUP_AUTH_USERS = ACL_TYPE_GROUP + "AuthenticatedUsers"
-S3_GROUP_ALL_USERS = ACL_TYPE_GROUP + "AllUsers"
-S3_GROUP_LOG_DELIVERY = ACL_TYPE_GROUP + "LogDelivery"
-
-NS = "http://s3.amazonaws.com/doc/2006-03-01/"
-NS2 = "http://www.w3.org/2001/XMLSchema-instance"
-
-def get_user_type(utype):
- for ut in [ ACL_TYPE_CANON_USER, ACL_TYPE_EMAIL_USER, ACL_TYPE_GROUP ]:
- if utype[:len(ut)] == ut:
- return ut
- raise Exception("unknown user type for user %s" % utype)
-
-def strip_user_type(utype):
- for ut in [ ACL_TYPE_CANON_USER, ACL_TYPE_EMAIL_USER, ACL_TYPE_GROUP ]:
- if utype[:len(ut)] == ut:
- return utype[len(ut):]
- raise Exception("unknown user type for user %s" % utype)
-
-def grantee_attribute_to_user_type(utype):
- if (utype == "Canonical User"):
- return ACL_TYPE_CANON_USER
- elif (utype == "CanonicalUser"):
- return ACL_TYPE_CANON_USER
- elif (utype == "Group"):
- return ACL_TYPE_GROUP
- elif (utype == "Email User"):
- return ACL_TYPE_EMAIL_USER
- elif (utype == "EmailUser"):
- return ACL_TYPE_EMAIL_USER
- else:
- raise Exception("unknown user type for user %s" % utype)
-
-def user_type_to_attr(t):
- if (t == ACL_TYPE_CANON_USER):
- return "CanonicalUser"
- elif (t == ACL_TYPE_GROUP):
- return "Group"
- elif (t == ACL_TYPE_EMAIL_USER):
- return "EmailUser"
- else:
- raise Exception("unknown user type %s" % t)
-
-def add_user_type(user):
- """ All users that are not specifically marked as something else
-are treated as canonical users"""
- for atype in ALL_ACL_TYPES:
- if (user[:len(atype)] == atype):
- return user
- return ACL_TYPE_CANON_USER + user
-
-class AclGrant(object):
- def __init__(self, user_id, display_name, permission):
- self.user_id = user_id
- self.display_name = display_name
- self.permission = permission
- def translate_users(self, xusers):
- # Keep in mind that xusers contains user_ids of the form "type:value"
- # So typical contents might be like { canon:XYZ => canon.123 }
- if (xusers.has_key(self.user_id)):
- self.user_id = xusers[self.user_id]
- # It's not clear what the new pretty-name should be, so just leave it blank.
- self.display_name = ""
- def equals(self, rhs):
- if (self.user_id != rhs.user_id):
- return False
- if (self.permission != rhs.permission):
- return False
- # ignore display_name
- return True
-
-class AclPolicy(object):
- def __init__(self, owner_id, owner_display_name, grants):
- self.owner_id = owner_id
- self.owner_display_name = owner_display_name
- self.grants = grants # dict of { string -> ACLGrant }
- @staticmethod
- def from_xml(s):
- root = etree.parse(StringIO(s))
- owner_id_node = root.find("{%s}Owner/{%s}ID" % (NS,NS))
- owner_id = owner_id_node.text
- owner_display_name_node = root.find("{%s}Owner/{%s}DisplayName" \
- % (NS,NS))
- if (owner_display_name_node != None):
- owner_display_name = owner_display_name_node.text
- else:
- owner_display_name = None
- grantlist = root.findall("{%s}AccessControlList/{%s}Grant" \
- % (NS,NS))
- grants = { }
- for g in grantlist:
- grantee = g.find("{%s}Grantee" % NS)
- user_id = grantee.find("{%s}ID" % NS).text
- if (grantee.attrib.has_key("type")):
- user_type = grantee.attrib["type"]
- else:
- user_type = grantee.attrib["{%s}type" % NS2]
- display_name = grantee.find("{%s}DisplayName" % NS).text
- permission = g.find("{%s}Permission" % NS).text
- grant_user_id = grantee_attribute_to_user_type(user_type) + user_id
- grants[grant_user_id] = AclGrant(grant_user_id, display_name, permission)
- return AclPolicy(owner_id, owner_display_name, grants)
- def to_xml(self):
- root = etree.Element("AccessControlPolicy", nsmap={None: NS})
- owner = etree.SubElement(root, "Owner")
- id_elem = etree.SubElement(owner, "ID")
- id_elem.text = self.owner_id
- if (self.owner_display_name and self.owner_display_name != ""):
- display_name_elem = etree.SubElement(owner, "DisplayName")
- display_name_elem.text = self.owner_display_name
- access_control_list = etree.SubElement(root, "AccessControlList")
- for k,g in self.grants.items():
- grant_elem = etree.SubElement(access_control_list, "Grant")
- grantee_elem = etree.SubElement(grant_elem, "{%s}Grantee" % NS,
- nsmap={None: NS, "xsi" : NS2})
- grantee_elem.set("{%s}type" % NS2, user_type_to_attr(get_user_type(g.user_id)))
- user_id_elem = etree.SubElement(grantee_elem, "{%s}ID" % NS)
- user_id_elem.text = strip_user_type(g.user_id)
- display_name_elem = etree.SubElement(grantee_elem, "{%s}DisplayName" % NS)
- display_name_elem.text = g.display_name
- permission_elem = etree.SubElement(grant_elem, "{%s}Permission" % NS)
- permission_elem.text = g.permission
- return etree.tostring(root, encoding="UTF-8")
- def translate_users(self, xusers):
- # Owner ids are always expressed in terms of canonical user id
- if (xusers.has_key(ACL_TYPE_CANON_USER + self.owner_id)):
- self.owner_id = \
- strip_user_type(xusers[ACL_TYPE_CANON_USER + self.owner_id])
- self.owner_display_name = ""
- for k,g in self.grants.items():
- g.translate_users(xusers)
- def set_owner(self, owner_id):
- self.owner_id = owner_id
- self.owner_display_name = ""
- def equals(self, rhs):
- if (self.owner_id != rhs.owner_id):
- return False
- for k,g in self.grants.items():
- if (not rhs.grants.has_key(k)):
- return False
- if (not g.equals(rhs.grants[k])):
- return False
- for l,r in rhs.grants.items():
- if (not self.grants.has_key(l)):
- return False
- if (not r.equals(self.grants[l])):
- return False
- return True
-
-def compare_xml(xml1, xml2):
- tree1 = etree.parse(StringIO(xml1))
- out1 = etree.tostring(tree1, encoding="UTF-8")
- tree2 = etree.parse(StringIO(xml2))
- out2 = etree.tostring(tree2, encoding="UTF-8")
- out1 = out1.replace("xsi:type", "type")
- out2 = out2.replace("xsi:type", "type")
- if out1 != out2:
- print "out1 = %s" % out1
- print "out2 = %s" % out2
- raise Exception("compare xml failed")
-
-#<?xml version="1.0" encoding="UTF-8"?>
-def test_acl_policy():
- test1_xml = \
-"<AccessControlPolicy xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">" + \
-"<Owner><ID>foo</ID><DisplayName>MrFoo</DisplayName></Owner><AccessControlList>" + \
-"<Grant><Grantee xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" " + \
-"xsi:type=\"CanonicalUser\"><ID>*** Owner-Canonical-User-ID ***</ID>" + \
-"<DisplayName>display-name</DisplayName></Grantee>" + \
-"<Permission>FULL_CONTROL</Permission></Grant></AccessControlList></AccessControlPolicy>"
- test1 = AclPolicy.from_xml(test1_xml)
- compare_xml(test1_xml, test1.to_xml())
-
-###### Object #######
-class Object(object):
- def __init__(self, name, md5, size):
- self.name = name
- self.md5 = md5
- self.size = int(size)
- def equals(self, rhs):
- if (self.name != rhs.name):
- return False
- if (self.md5 != rhs.md5):
- return False
- if (self.size != rhs.size):
- return False
- return True
- def local_name(self):
- return s3_name_to_local_name(self.name)
- @staticmethod
- def from_file(obj_name, path):
- f = open(path, 'r')
- try:
- md5 = get_md5(f)
- finally:
- f.close()
- size = os.path.getsize(path)
- #print "Object.from_file: path="+path+",md5=" + bytes_to_str(md5) +",size=" + str(size)
- return Object(obj_name, md5, size)
-
-###### Store #######
-class Store(object):
- @staticmethod
- def make_store(url, create, akey, skey):
- s3_url = strip_prefix("s3://", url)
- if (s3_url):
- return S3Store(s3_url, create, akey, skey)
- rados_url = strip_prefix("rados:", url)
- if (rados_url):
- return RadosStore(rados_url, create, akey, skey)
- file_url = strip_prefix("file://", url)
- if (file_url):
- return FileStore(file_url, create)
- if (url[0:1] == "/"):
- return FileStore(url, create)
- if (url[0:2] == "./"):
- return FileStore(url, create)
- raise Exception("Failed to find a prefix of s3://, file://, /, or ./ \
-Cannot handle this URL.")
- def __init__(self, url):
- self.url = url
-
-###### LocalCopy ######
-class LocalCopy(object):
- def __init__(self, obj_name, path, path_is_temp):
- self.obj_name = obj_name
- self.path = path
- self.path_is_temp = path_is_temp
- def remove(self):
- if ((self.path_is_temp == True) and (self.path != None)):
- os.unlink(self.path)
- self.path = None
- self.path_is_temp = False
- def __del__(self):
- self.remove()
-
-class LocalAcl(object):
- @staticmethod
- def from_file(obj_name, file_name):
- f = open(file_name, "r")
- try:
- xml = f.read()
- finally:
- f.close()
- return LocalAcl.from_xml(obj_name, xml)
- @staticmethod
- def from_xml(obj_name, xml):
- acl_policy = AclPolicy.from_xml(xml)
- return LocalAcl(obj_name, acl_policy)
- @staticmethod
- def get_empty(obj_name):
- return LocalAcl(obj_name, None)
- def __init__(self, obj_name, acl_policy):
- self.obj_name = obj_name
- self.acl_policy = acl_policy
- def equals(self, rhs):
- """ Compare two LocalAcls """
- if (self.acl_policy == None):
- return (rhs.acl_policy == None)
- if (rhs.acl_policy == None):
- return (self.acl_policy == None)
- return self.acl_policy.equals(rhs.acl_policy)
- def translate_users(self, xusers):
- """ Translate the users in this ACL """
- if (self.acl_policy == None):
- return
- self.acl_policy.translate_users(xusers)
- def set_owner(self, owner_id):
- if (self.acl_policy == None):
- return
- self.acl_policy.set_owner(owner_id)
- def write_to_file(self, file_name):
- """ Write this ACL to a file """
- if (self.acl_policy == None):
- return
- xml = self.acl_policy.to_xml()
- f = open(file_name, 'w')
- try:
- f.write(xml)
- finally:
- f.close()
-
-###### S3 store #######
-class S3StoreIterator(object):
- """S3Store iterator"""
- def __init__(self, blrs):
- self.blrs = blrs
- def __iter__(self):
- return self
- def next(self):
- # This will raise StopIteration when there are no more objects to
- # iterate on
- key = self.blrs.next()
- ret = Object(key.name, etag_to_md5(key.etag), key.size)
- return ret
-
-class S3Store(Store):
- def __init__(self, url, create, akey, skey):
- # Parse the s3 url
- host_end = string.find(url, "/")
- if (host_end == -1):
- raise Exception("S3Store URLs are of the form \
-s3://host/bucket/key_prefix. Failed to find the host.")
- self.host = url[0:host_end]
- bucket_end = url.find("/", host_end+1)
- if (bucket_end == -1):
- self.bucket_name = url[host_end+1:]
- self.key_prefix = ""
- else:
- self.bucket_name = url[host_end+1:bucket_end]
- self.key_prefix = url[bucket_end+1:]
- if (self.bucket_name == ""):
- raise Exception("S3Store URLs are of the form \
-s3://host/bucket/key_prefix. Failed to find the bucket.")
- if (opts.more_verbose):
- print "self.host = '" + self.host + "', ",
- print "self.bucket_name = '" + self.bucket_name + "' ",
- print "self.key_prefix = '" + self.key_prefix + "'"
- self.conn = S3Connection(calling_format=OrdinaryCallingFormat(),
- host=self.host, is_secure=False,
- aws_access_key_id=akey, aws_secret_access_key=skey)
- self.bucket = self.conn.lookup(self.bucket_name)
- if (self.bucket == None):
- if (create):
- if (opts.dry_run):
- raise Exception("logic error: this should be unreachable.")
- self.bucket = self.conn.create_bucket(bucket_name = self.bucket_name)
- else:
- raise RuntimeError("%s: no such bucket as %s" % \
- (url, self.bucket_name))
- Store.__init__(self, "s3://" + url)
- def __str__(self):
- return "s3://" + self.host + "/" + self.bucket_name + "/" + self.key_prefix
- def get_acl(self, obj):
- acl_xml = self.bucket.get_xml_acl(obj.name)
- return LocalAcl.from_xml(obj.name, acl_xml)
- def make_local_copy(self, obj):
- k = Key(self.bucket)
- k.key = obj.name
- temp_file = tempfile.NamedTemporaryFile(mode='w+b', delete=False).name
- try:
- k.get_contents_to_filename(temp_file)
- except:
- os.unlink(temp_file)
- raise
- return LocalCopy(obj.name, temp_file, True)
- def all_objects(self):
- blrs = self.bucket.list(prefix = self.key_prefix)
- return S3StoreIterator(blrs.__iter__())
- def locate_object(self, obj):
- k = self.bucket.get_key(obj.name)
- if (k == None):
- return None
- return Object(obj.name, etag_to_md5(k.etag), k.size)
- def upload(self, local_copy, src_acl, obj):
- if (opts.more_verbose):
- print "S3Store.UPLOAD: local_copy.path='" + local_copy.path + "' " + \
- "obj='" + obj.name + "'"
- if (opts.dry_run):
- return
-# mime = mimetypes.guess_type(local_copy.path)[0]
-# if (mime == NoneType):
-# mime = "application/octet-stream"
- k = Key(self.bucket)
- k.key = obj.name
- #k.set_metadata("Content-Type", mime)
- k.set_contents_from_filename(local_copy.path)
- if (src_acl.acl_policy != None):
- try:
- xml = src_acl.acl_policy.to_xml()
- self.bucket.set_xml_acl(xml, k)
- except Exception, e:
- print >>stderr, "ERROR SETTING ACL on object '" + sobj.name + "'"
- print >>stderr
- print >>stderr, "************* ACL: *************"
- print >>stderr, str(xml)
- print >>stderr, "********************************"
- raise
-
- def remove(self, obj):
- if (opts.dry_run):
- return
- self.bucket.delete_key(obj.name)
- if (opts.more_verbose):
- print "S3Store: removed %s" % obj.name
-
-###### FileStore #######
-class FileStoreIterator(object):
- """FileStore iterator"""
- def __init__(self, base):
- self.base = base
- if (opts.follow_symlinks):
- self.generator = os.walk(base, followlinks=True)
- else:
- self.generator = os.walk(base)
- self.path = ""
- self.files = []
- def __iter__(self):
- return self
- def next(self):
- while True:
- if (len(self.files) == 0):
- self.path, dirs, self.files = self.generator.next()
- continue
- path = self.path + "/" + self.files[0]
- self.files = self.files[1:]
- # Ignore non-files when iterating.
- if (not os.path.isfile(path)):
- continue
- try:
- obj_name = local_name_to_s3_name(path[len(self.base)+1:])
- except LocalFileIsAcl, e:
- # ignore ACL side files when iterating
- continue
- return Object.from_file(obj_name, path)
-
-class FileStore(Store):
- def __init__(self, url, create):
- # Parse the file url
- self.base = url
- if (self.base[-1:] == '/'):
- self.base = self.base[:-1]
- if (create):
- if (opts.dry_run):
- raise Exception("logic error: this should be unreachable.")
- mkdir_p(self.base)
- elif (not os.path.isdir(self.base)):
- raise NonexistentStore()
- Store.__init__(self, "file://" + url)
- def __str__(self):
- return "file://" + self.base
- def get_acl(self, obj):
- acl_name = get_local_acl_file_name(obj.local_name())
- acl_path = self.base + "/" + acl_name
- if (os.path.exists(acl_path)):
- return LocalAcl.from_file(obj.name, acl_path)
- else:
- return LocalAcl.get_empty(obj.name)
- def make_local_copy(self, obj):
- local_name = obj.local_name()
- return LocalCopy(obj.name, self.base + "/" + local_name, False)
- def all_objects(self):
- return FileStoreIterator(self.base)
- def locate_object(self, obj):
- path = self.base + "/" + obj.local_name()
- found = os.path.isfile(path)
- if (opts.more_verbose):
- if (found):
- print "FileStore::locate_object: found object '" + \
- obj.name + "'"
- else:
- print "FileStore::locate_object: did not find object '" + \
- obj.name + "'"
- if (not found):
- return None
- return Object.from_file(obj.name, path)
- def upload(self, local_copy, src_acl, obj):
- if (opts.more_verbose):
- print "FileStore.UPLOAD: local_copy.path='" + local_copy.path + "' " + \
- "obj='" + obj.name + "'"
- if (opts.dry_run):
- return
- s = local_copy.path
- lname = obj.local_name()
- d = self.base + "/" + lname
- #print "s='" + s +"', d='" + d + "'"
- mkdir_p(os.path.dirname(d))
- shutil.copy(s, d)
- if (src_acl.acl_policy != None):
- src_acl.write_to_file(self.base + "/" + get_local_acl_file_name(lname))
- def remove(self, obj):
- if (opts.dry_run):
- return
- os.unlink(self.base + "/" + obj.name)
- if (opts.more_verbose):
- print "FileStore: removed %s" % obj.name
-
-###### Rados store #######
-class RadosStoreIterator(object):
- """RadosStore iterator"""
- def __init__(self, it, rados_store):
- self.it = it # has type rados.ObjectIterator
- self.rados_store = rados_store
- self.prefix = self.rados_store.prefix
- self.prefix_len = len(self.rados_store.prefix)
- def __iter__(self):
- return self
- def next(self):
- rados_obj = None
- while True:
- # This will raise StopIteration when there are no more objects to
- # iterate on
- rados_obj = self.it.next()
- # do the prefixes match?
- if rados_obj.key[:self.prefix_len] == self.prefix:
- break
- ret = self.rados_store.obsync_obj_from_rgw(rados_obj.key)
- if (ret == None):
- raise Exception("internal iterator error")
- return ret
-
-class RadosStore(Store):
- def __init__(self, url, create, akey, skey):
- # Parse the rados url
- conf_end = string.find(url, ":")
- if (conf_end == -1):
- raise Exception("RadosStore URLs are of the form \
-rados:path/to/ceph/conf:bucket:key_prefix. Failed to find the path to the conf.")
- self.conf_file_path = url[0:conf_end]
- bucket_end = url.find(":", conf_end+1)
- if (bucket_end == -1):
- self.rgw_bucket_name = url[conf_end+1:]
- self.key_prefix = ""
- else:
- self.rgw_bucket_name = url[conf_end+1:bucket_end]
- self.key_prefix = url[bucket_end+1:]
- if (self.rgw_bucket_name == ""):
- raise Exception("RadosStore URLs are of the form \
-rados:/path/to/ceph/conf:pool:key_prefix. Failed to find the bucket.")
- if (opts.more_verbose):
- print "self.conf_file_path = '" + self.conf_file_path + "', ",
- print "self.rgw_bucket_name = '" + self.rgw_bucket_name + "' ",
- print "self.key_prefix = '" + self.key_prefix + "'"
- acl_hack = getenv("ACL_HACK", None)
- if (acl_hack == None):
- raise Exception("RadosStore error: You must specify an environment " +
- "variable called ACL_HACK containing the name of a file. This " +
- "file contains a serialized RGW ACL that you want " +
- "to insert into the user.rgw.acl extended attribute of all " +
- "the objects you create. This is a hack and yes, it will go " +
- "away soon.")
- acl_hack_f = open(acl_hack, "r")
- try:
- self.acl_hack = acl_hack_f.read()
- finally:
- acl_hack_f.close()
- self.rados = rados.Rados()
- self.rados.conf_read_file(self.conf_file_path)
- self.rados.connect()
- if (not self.rados.pool_exists(self.rgw_bucket_name)):
- if (create):
- self.create_rgw_bucket(self.rgw_bucket_name)
- else:
- raise NonexistentStore()
- self.ioctx = self.rados.open_ioctx(self.rgw_bucket_name)
- Store.__init__(self, "rados:" + url)
- def create_rgw_bucket(self, rgw_bucket_name):
- """ Create an rgw bucket named 'rgw_bucket_name' """
- self.rados.create_pool(self.rgw_bucket_name)
- meta_ctx = None
- try:
- meta_ctx = self.rados.open_ioctx(RGW_META_BUCKET_NAME)
- meta_ctx.write(rgw_bucket_name, "", 0)
- print "meta_ctx.set_xattr(rgw_bucket_name=" + rgw_bucket_name + ", " + \
- "user.rgw.acl, self.acl_hack=" + self.acl_hack + ")"
- meta_ctx.set_xattr(rgw_bucket_name, "user.rgw.acl", self.acl_hack)
- finally:
- if (meta_ctx):
- meta_ctx.close()
- def obsync_obj_from_rgw(self, key):
- """Create an obsync object from a Rados object"""
- try:
- size, tm = self.ioctx.stat(key)
- except rados.ObjectNotFound:
- return None
- md5 = self.ioctx.get_xattr(key, "user.rgw.etag")
- return Object(key, md5, size)
- def __str__(self):
- return "rados:" + self.conf_file_path + ":" + self.rgw_bucket_name + ":" + self.key_prefix
- def get_acl(self, obj):
- acl = LocalAcl(obj.name)
- # todo: set XML ACL
- return acl
- def make_local_copy(self, obj):
- temp_file = None
- temp_file_f = None
- try:
- # read the object from rados in chunks
- temp_file = tempfile.NamedTemporaryFile(mode='w+b', delete=False)
- temp_file_f = open(temp_file.name, 'w')
- while True:
- buf = self.ioctx.read(obj.name, off, 8192)
- if (len(buf) == 0):
- break
- temp_file_f.write(buf)
- if (len(buf) < 8192):
- break
- off += 8192
- temp_file_f.close()
- # TODO: implement ACLs
- except:
- if (temp_file_f):
- temp_file_f.close()
- if (temp_file):
- os.unlink(temp_file.name)
- raise
- return LocalCopy(obj.name, temp_file.name, True)
- def all_objects(self):
- it = self.bucket.list_objects()
- return RadosStoreIterator(it, self.key_prefix)
- def locate_object(self, obj):
- return self.obsync_obj_from_rgw(obj.name)
- def upload(self, local_copy, src_acl, obj):
- if (opts.more_verbose):
- print "RadosStore.UPLOAD: local_copy.path='" + local_copy.path + "' " + \
- "obj='" + obj.name + "'"
- if (opts.dry_run):
- return
- local_copy_f = open(local_copy.path, 'r')
- off = 0
- while True:
- buf = local_copy_f.read(8192)
- if ((len(buf) == 0) and (off != 0)):
- break
- self.ioctx.write(obj.name, buf, off)
- if (len(buf) < 8192):
- break
- off += 8192
- self.ioctx.set_xattr(obj.name, "user.rgw.etag", obj.md5)
- self.ioctx.set_xattr(obj.name, "user.rgw.acl", self.acl_hack)
- self.ioctx.set_xattr(obj.name, "user.rgw.content_type",
- "application/octet-stream")
- def remove(self, obj):
- if (opts.dry_run):
- return
- self.ioctx.remove_object(obj.name)
- if (opts.more_verbose):
- print "RadosStore: removed %s" % obj.name
-###### Functions #######
-def delete_unreferenced(src, dst):
- """ delete everything from dst that is not referenced in src """
- if (opts.more_verbose):
- print "handling deletes."
- for dobj in dst.all_objects():
- sobj = src.locate_object(dobj)
- if (sobj == None):
- dst.remove(dobj)
-
-def xuser_cb(opt, opt_str, value, parser):
- """ handle an --xuser argument """
- equals = value.find(r'=')
- if equals == -1:
- print >>stderr, "Error parsing --xuser: You must give both a source \
-and destination user name, like so:\n\
---xuser SOURCE_USER=DEST_USER\n\
-\n\
-This will translate the user SOURCE_USER in the source to the user DEST_USER \n\
-in the destination."
- sys.exit(1)
- src_user = value[:equals]
- dst_user = value[equals+1:]
- if ((len(src_user) == 0) or (len(dst_user) == 0)):
- print >>stderr, "Error parsing --xuser: can't have a zero-length \
-user name."
- sys.exit(1)
- src_user = add_user_type(src_user)
- dst_user = add_user_type(dst_user)
- if (xuser.has_key(src_user)):
- print >>stderr, "Error parsing --xuser: we are already translating \
-\"%s\" to \"%s\"; we cannot translate it to \"%s\"" % \
-(src_user, xuser[src_user], dst_user)
- sys.exit(1)
- xuser[src_user] = dst_user
-
-USAGE = """
-obsync synchronizes S3, Rados, and local objects. The source and destination
-can both be local or both remote.
-
-Examples:
-# copy contents of mybucket to disk
-obsync -v s3://myhost/mybucket file://mydir
-
-# copy contents of mydir to an S3 bucket
-obsync -v file://mydir s3://myhost/mybucket
-
-# synchronize two S3 buckets
-SRC_AKEY=... SRC_SKEY=... \
-DST_AKEY=... DST_SKEY=... \
-obsync -v s3://myhost/mybucket1 s3://myhost2/mybucket2
- --xuser bob=robert --xuser joe=joseph -O bob
-
-Note: You must specify an AWS access key and secret access key when accessing
-S3. obsync honors these environment variables:
-SRC_AKEY Access key for the source URL
-SRC_SKEY Secret access key for the source URL
-DST_AKEY Access key for the destination URL
-DST_SKEY Secret access key for the destination URL
-AKEY Access key for both source and dest
-SKEY Secret access key for both source and dest
-
-If these environment variables are not given, we will fall back on libboto
-defaults.
-
-obsync (options) [source] [destination]"""
-
-parser = OptionParser(USAGE)
-parser.add_option("-n", "--dry-run", action="store_true", \
- dest="dry_run", default=False)
-parser.add_option("-c", "--create-dest", action="store_true", \
- dest="create", help="create the destination if it doesn't already exist")
-parser.add_option("--delete-before", action="store_true", \
- dest="delete_before", help="delete objects that aren't in SOURCE from \
-DESTINATION before transferring any objects")
-parser.add_option("-d", "--delete-after", action="store_true", \
- dest="delete_after", help="delete objects that aren't in SOURCE from \
-DESTINATION after doing all transfers.")
-parser.add_option("-L", "--follow-symlinks", action="store_true", \
- dest="follow_symlinks", help="follow symlinks (please avoid symlink " + \
- "loops when using this option!)")
-parser.add_option("--no-preserve-acls", action="store_true", \
- dest="no_preserve_acls", help="don't preserve ACLs when copying objects.")
-parser.add_option("-v", "--verbose", action="store_true", \
- dest="verbose", help="be verbose")
-parser.add_option("-V", "--more-verbose", action="store_true", \
- dest="more_verbose", help="be really, really verbose (developer mode)")
-parser.add_option("-x", "--xuser", type="string", nargs=1, action="callback", \
- dest="SRC=DST", callback=xuser_cb, help="set up a user tranlation. You \
-can specify multiple user translations with multiple --xuser arguments.")
-parser.add_option("--force", action="store_true", \
- dest="force", help="overwrite all destination objects, even if they \
-appear to be the same as the source objects.")
-parser.add_option("--unit", action="store_true", \
- dest="run_unit_tests", help="run unit tests and quit")
-xuser = {}
-(opts, args) = parser.parse_args()
-if (opts.run_unit_tests):
- test_acl_policy()
- sys.exit(0)
-
-opts.preserve_acls = not opts.no_preserve_acls
-if (opts.create and opts.dry_run):
- raise Exception("You can't run with both --create-dest and --dry-run! \
-By definition, a dry run never changes anything.")
-if (len(args) < 2):
- print >>stderr, "Expected two positional arguments: source and destination"
- print >>stderr, USAGE
- sys.exit(1)
-elif (len(args) > 2):
- print >>stderr, "Too many positional arguments."
- print >>stderr, USAGE
- sys.exit(1)
-if (opts.more_verbose):
- print >>stderr, "User translations:"
- for k,v in xuser.items():
- print >>stderr, "\"%s\" ==> \"%s\"" % (k, v)
- print >>stderr, ""
-if (opts.more_verbose):
- opts.verbose = True
- boto.set_stream_logger("stdout")
- boto.log.info("Enabling verbose boto logging.")
-if (opts.delete_before and opts.delete_after):
- print >>stderr, "It doesn't make sense to specify both --delete-before \
-and --delete-after."
- sys.exit(1)
-src_name = args[0]
-dst_name = args[1]
-try:
- if (opts.more_verbose):
- print "SOURCE: " + src_name
- src = Store.make_store(src_name, False,
- getenv("SRC_AKEY", "AKEY"), getenv("SRC_SKEY", "SKEY"))
-except NonexistentStore, e:
- print >>stderr, "Fatal error: Source " + src_name + " does not exist."
- sys.exit(1)
-except Exception, e:
- print >>stderr, "error creating source: " + str(e)
- traceback.print_exc(100000, stderr)
- sys.exit(1)
-try:
- if (opts.more_verbose):
- print "DESTINATION: " + dst_name
- dst = Store.make_store(dst_name, opts.create,
- getenv("DST_AKEY", "AKEY"), getenv("DST_SKEY", "SKEY"))
-except NonexistentStore, e:
- print >>stderr, "Fatal error: Destination " + dst_name + " does " +\
- "not exist. Run with -c or --create-dest to create it automatically."
- sys.exit(1)
-except Exception, e:
- print >>stderr, "error creating destination: " + str(e)
- traceback.print_exc(100000, stderr)
- sys.exit(1)
-
-if (opts.delete_before):
- delete_unreferenced(src, dst)
-
-for sobj in src.all_objects():
- if (opts.more_verbose):
- print "handling " + sobj.name
- dobj = dst.locate_object(sobj)
- upload = False
- src_acl = None
- dst_acl = None
- if (opts.force):
- if (opts.verbose):
- print "F " + sobj.name
- upload = True
- elif (dobj == None):
- if (opts.verbose):
- print "+ " + sobj.name
- upload = True
- elif not sobj.equals(dobj):
- if (opts.verbose):
- print "> " + sobj.name
- upload = True
- elif (opts.preserve_acls):
- # Do the ACLs match?
- src_acl = src.get_acl(sobj)
- dst_acl = dst.get_acl(dobj)
- src_acl.translate_users(xuser)
- #src_acl.set_owner()
- if (not src_acl.equals(dst_acl)):
- upload = True
- if (opts.verbose):
- print "^ " + sobj.name
- else:
- if (opts.verbose):
- print ". " + sobj.name
- if (upload):
- if (not opts.preserve_acls):
- # Just default to an empty ACL
- src_acl = LocalAcl.get_empty(sobj.name)
- else:
- if (src_acl == None):
- src_acl = src.get_acl(sobj)
- src_acl.translate_users(xuser)
- #src_acl.set_owner()
- local_copy = src.make_local_copy(sobj)
- try:
- dst.upload(local_copy, src_acl, sobj)
- finally:
- local_copy.remove()
-
-if (opts.delete_after):
- delete_unreferenced(src, dst)
-
-if (opts.more_verbose):
- print "finished."
-
-sys.exit(0)
projects / ceph.git / commitdiff