Merge "Initial implementation of glance replication."

This commit is contained in:
Jenkins 2012-07-10 19:25:36 +00:00 committed by Gerrit Code Review
commit 77d534e708
2 changed files with 864 additions and 0 deletions

656
bin/glance-replicator Executable file
View File

@ -0,0 +1,656 @@
#!/usr/bin/env python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2012 Michael Still and Canonical Inc
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import gettext
import httplib
import json
import logging
import logging.config
import logging.handlers
import optparse
import os
import sys
import urllib
import uuid
# If ../glance/__init__.py exists, add ../ to Python search path, so that
# it will override what happens to be installed in /usr/(local/)lib/python...
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(sys.argv[0]),
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir, 'glance', '__init__.py')):
sys.path.insert(0, possible_topdir)
gettext.install('glance', unicode=1)
COMMANDS = """Commands:
help <command> Output help for one of the commands below
compare What is missing from the slave glance?
dump Dump the contents of a glance instance to local disk.
livecopy Load the contents of one glance instance into another.
load Load the contents of a local directory into glance.
size Determine the size of a glance instance if dumped to disk.
"""
class UploadException(Exception):
pass
class AuthenticationException(Exception):
pass
class ImageService(object):
def __init__(self, conn, auth_token):
""" Initialize the ImageService.
conn: a httplib.HTTPConnection to the glance server
auth_token: authentication token to pass in the x-auth-token header
"""
self.auth_token = auth_token
self.conn = conn
def _http_request(self, method, url, headers, body,
ignore_result_body=False):
"""Perform an HTTP request against the server.
method: the HTTP method to use
url: the URL to request (not including server portion)
headers: headers for the request
body: body to send with the request
ignore_result_body: the body of the result will be ignored
Returns: a httplib response object
"""
if self.auth_token:
headers.setdefault('x-auth-token', self.auth_token)
logging.debug(_('Request: %(method)s http://%(server)s:%(port)s/'
'%(url)s with headers %(headers)s')
% {'method': method,
'server': self.conn.host,
'port': self.conn.port,
'url': url,
'headers': repr(headers)})
self.conn.request(method, url, body, headers)
response = self.conn.getresponse()
headers = self._header_list_to_dict(response.getheaders())
code = response.status
code_description = httplib.responses[code]
logging.debug(_('Response: %(code)s %(status)s %(headers)s')
% {'code': code,
'status': code_description,
'headers': repr(headers)})
if code in [401, 403]:
raise AuthenticationException()
if ignore_result_body:
# NOTE: because we are pipelining requests through a single HTTP
# connection, httplib requires that we read the response body
# before we can make another request. If the caller knows they
# don't care about the body, they can ask us to do that for them.
response.read()
return response
def get_images(self):
"""Return a detailed list of images.
Yields a series of images as dicts containing metadata.
"""
params = {'is_public': None}
while True:
url = 'v1/images/detail'
query = urllib.urlencode(params)
if query:
url += '?%s' % query
response = self._http_request('GET', url, {}, '')
result = json.loads(response.read())
if not result or not 'images' in result or not result['images']:
return
for image in result.get('images', []):
params['marker'] = image['id']
yield image
def get_image(self, image_uuid):
"""Fetch image data from glance.
image_uuid: the id of an image
Returns: a httplib Response object where the body is the image.
"""
url = 'v1/images/%s' % image_uuid
return self._http_request('GET', url, {}, '')
@staticmethod
def _header_list_to_dict(headers):
"""Expand a list of headers into a dictionary.
headers: a list of [(key, value), (key, value), (key, value)]
Returns: a dictionary representation of the list
"""
d = {}
for (header, value) in headers:
if header.startswith('x-image-meta-property-'):
prop = header.replace('x-image-meta-properties', '')
d.setdefault('properties', {})
d['properties'][prop] = value
else:
d[header.replace('x-image-meta-', '')] = value
return d
def get_image_meta(self, image_uuid):
"""Return the metadata for a single image.
image_uuid: the id of an image
Returns: image metadata as a dictionary
"""
url = 'v1/images/%s' % image_uuid
response = self._http_request('HEAD', url, {}, '',
ignore_result_body=True)
return self._header_list_to_dict(response.getheaders())
@staticmethod
def _dict_to_headers(d):
"""Convert a dictionary into one suitable for a HTTP request.
d: a dictionary
Returns: the same dictionary, with x-image-meta added to every key
"""
h = {}
for key in d:
if key == 'properties':
for subkey in d[key]:
h['x-image-meta-property-%s' % subkey] = d[key][subkey]
else:
h['x-image-meta-%s' % key] = d[key]
return h
def add_image(self, image_meta, image_data):
"""Upload an image.
image_meta: image metadata as a dictionary
image_data: image data as a object with a read() method
Returns: a tuple of (http response headers, http response body)
"""
url = 'v1/images'
headers = self._dict_to_headers(image_meta)
headers['Content-Type'] = 'application/octet-stream'
headers['Content-Length'] = int(image_meta['size'])
response = self._http_request('POST', url, headers, image_data)
headers = self._header_list_to_dict(response.getheaders())
logging.debug(_('Image post done'))
body = response.read()
return headers, body
def add_image_meta(self, image_meta):
"""Update image metadata.
image_meta: image metadata as a dictionary
Returns: a tuple of (http response headers, http response body)
"""
url = 'v1/images/%s' % image_meta['id']
headers = self._dict_to_headers(image_meta)
headers['Content-Type'] = 'application/octet-stream'
response = self._http_request('PUT', url, headers, '')
headers = self._header_list_to_dict(response.getheaders())
logging.debug(_('Image post done'))
body = response.read()
return headers, body
def replication_size(options, args):
"""%(prog)s size <server:port>
Determine the size of a glance instance if dumped to disk.
server:port: the location of the glance instance.
"""
server_port = args.pop()
server, port = server_port.split(':')
total_size = 0
count = 0
client = ImageService(httplib.HTTPConnection(server, port),
options.token)
for image in client.get_images():
logging.debug(_('Considering image: %(image)s') % {'image': image})
if image['status'] == 'active':
total_size += int(image['size'])
count += 1
print _('Total size is %d bytes across %d images') % (total_size, count)
def replication_dump(options, args):
"""%(prog)s dump <server:port> <path>
Dump the contents of a glance instance to local disk.
server:port: the location of the glance instance.
path: a directory on disk to contain the data.
"""
path = args.pop()
server_port = args.pop()
server, port = server_port.split(':')
client = ImageService(httplib.HTTPConnection(server, port),
options.token)
for image in client.get_images():
logging.info(_('Considering: %s' % image['id']))
data_path = os.path.join(path, image['id'])
if not os.path.exists(data_path):
logging.info(_('... storing'))
# Dump glance information
f = open(data_path, 'w')
f.write(json.dumps(image))
f.close()
if image['status'] == 'active' and not options.metaonly:
# Now fetch the image. The metadata returned in headers here
# is the same as that which we got from the detailed images
# request earlier, so we can ignore it here. Note that we also
# only dump active images.
logging.info(_('... image is active'))
image_response = client.get_image(image['id'])
f = open(data_path + '.img', 'wb')
while True:
chunk = image_response.read(options.chunksize)
if not chunk:
break
f.write(chunk)
f.close()
def _dict_diff(a, b):
"""A one way dictionary diff.
a: a dictionary
b: a dictionary
Returns: True if the dictionaries are different
"""
# Only things the master has which the slave lacks matter
if set(a.keys()) - set(b.keys()):
logging.debug(_('metadata diff -- master has extra keys: %(keys)s')
% {'keys': ' '.join(set(a.keys()) - set(b.keys()))})
return True
for key in a:
if str(a[key]) != str(b[key]):
logging.debug(_('metadata diff -- value differs for key '
'%(key)s: master "%(master_value)s" vs '
'slave "%(slave_value)s"')
% {'key': key,
'master_value': a[key],
'slave_value': b[key]})
return True
return False
# This is lifted from openstack-common, but copied here to reduce dependancies
def is_uuid_like(value):
try:
uuid.UUID(value)
return True
except Exception:
return False
def replication_load(options, args):
"""%(prog)s load <server:port> <path>
Load the contents of a local directory into glance.
server:port: the location of the glance instance.
path: a directory on disk containing the data.
"""
path = args.pop()
server_port = args.pop()
server, port = server_port.split(':')
client = ImageService(httplib.HTTPConnection(server, port),
options.token)
for ent in os.listdir(path):
if is_uuid_like(ent):
uuid = ent
logging.info(_('Considering: %s') % uuid)
meta_file_name = os.path.join(path, uuid)
meta_file = open(meta_file_name)
meta = json.loads(meta_file.read())
meta_file.close()
# Remove keys which don't make sense for replication
for key in options.dontreplicate.split(' '):
if key in meta:
del meta[key]
if _image_present(client, uuid):
# NOTE(mikal): Perhaps we just need to update the metadata?
# Note that we don't attempt to change an image file once it
# has been uploaded.
headers = client.get_image_meta(uuid)
for key in options.dontreplicate.split(' '):
if key in headers:
del headers[key]
if _dict_diff(meta, headers):
logging.info(_('... metadata has changed'))
headers, body = client.add_image_meta(meta)
_check_upload_response_headers(headers, body)
else:
if not os.path.exists(os.path.join(path, uuid + '.img')):
logging.info(_('... dump is missing image data, skipping'))
continue
# Upload the image itself
img_file = open(os.path.join(path, uuid + '.img'))
headers, body = client.add_image(meta, img_file)
img_file.close()
_check_upload_response_headers(headers, body)
def replication_livecopy(options, args):
"""%(prog)s livecopy <fromserver:port> <toserver:port>
Load the contents of one glance instance into another.
fromserver:port: the location of the master glance instance.
toserver:port: the location of the slave glance instance.
"""
slave_server_port = args.pop()
slave_server, slave_port = slave_server_port.split(':')
slave_conn = httplib.HTTPConnection(slave_server, slave_port)
slave_client = ImageService(slave_conn, options.token)
master_server_port = args.pop()
master_server, master_port = master_server_port.split(':')
master_conn = httplib.HTTPConnection(master_server, master_port)
master_client = ImageService(master_conn, options.token)
for image in master_client.get_images():
logging.info(_('Considering %(id)s') % {'id': image['id']})
if _image_present(slave_client, image['id']):
# NOTE(mikal): Perhaps we just need to update the metadata?
# Note that we don't attempt to change an image file once it
# has been uploaded.
headers = slave_client.get_image_meta(image['id'])
if headers['status'] == 'active':
for key in options.dontreplicate.split(' '):
if key in image:
del image[key]
if key in headers:
del headers[key]
if _dict_diff(image, headers):
logging.info(_('... metadata has changed'))
headers, body = slave_client.add_image_meta(image)
_check_upload_response_headers(headers, body)
elif image['status'] == 'active':
logging.info(_('%s is being synced') % image['id'])
if not options.metaonly:
image_response = master_client.get_image(image['id'])
headers, body = slave_client.add_image(image, image_response)
_check_upload_response_headers(headers, body)
def replication_compare(options, args):
"""%(prog)s compare <fromserver:port> <toserver:port>
Compare the contents of fromserver with those of toserver.
fromserver:port: the location of the master glance instance.
toserver:port: the location of the slave glance instance.
"""
slave_server_port = args.pop()
slave_server, slave_port = slave_server_port.split(':')
slave_conn = httplib.HTTPConnection(slave_server, slave_port)
slave_client = ImageService(slave_conn, options.token)
master_server_port = args.pop()
master_server, master_port = master_server_port.split(':')
master_conn = httplib.HTTPConnection(master_server, master_port)
master_client = ImageService(master_conn, options.token)
for image in master_client.get_images():
if _image_present(slave_client, image['id']):
headers = slave_client.get_image_meta(image['id'])
for key in options.dontreplicate.split(' '):
if key in image:
del image[key]
if key in headers:
del headers[key]
for key in image:
if image[key] != headers.get(key, None):
logging.info(_('%(image_id)s: field %(key)s differs '
'(source is %(master_value)s, destination '
'is %(slave_value)s)')
% {'image_id': image['id'],
'key': key,
'master_value': image[key],
'slave_value': headers.get(key,
'undefined')})
else:
logging.debug(_('%(image_id)s is identical')
% {'image_id': image['id']})
elif image['status'] == 'active':
logging.info(_('%s: entirely missing from the destination')
% image['id'])
def _check_upload_response_headers(headers, body):
"""Check that the headers of an upload are reasonable.
headers: the headers from the upload
body: the body from the upload
"""
if 'status' not in headers:
try:
d = json.loads(body)
if 'image' in d and 'status' in d['image']:
return
except:
raise UploadException('Image upload problem: %s' % body)
def _image_present(client, uuid):
"""Check if an image is present in glance.
client: the ImageService
uuid: the image uuid to check
Returns: True if the image is present
"""
headers = client.get_image_meta(uuid)
return 'status' in headers
def parse_options(parser, cli_args):
"""Returns the parsed CLI options, command to run and its arguments, merged
with any same-named options found in a configuration file
parser: the option parser
cli_args: the arguments passed on the command line
Returns: a tuple of (the parsed options, the command, the command name)
"""
if not cli_args:
cli_args.append('-h') # Show options in usage output...
(options, args) = parser.parse_args(cli_args)
# HACK(sirp): Make the parser available to the print_help method
# print_help is a command, so it only accepts (options, args); we could
# one-off have it take (parser, options, args), however, for now, I think
# this little hack will suffice
options.__parser = parser
if not args:
parser.print_usage()
sys.exit(0)
command_name = args.pop(0)
command = lookup_command(parser, command_name)
return (options, command, args)
def print_help(options, args):
"""Print help specific to a command.
options: the parsed command line options
args: the command line
"""
if len(args) != 1:
print COMMANDS
sys.exit(1)
parser = options.__parser
command_name = args.pop()
command = lookup_command(parser, command_name)
print command.__doc__ % {'prog': os.path.basename(sys.argv[0])}
def lookup_command(parser, command_name):
"""Lookup a command.
parser: the command parser
command_name: the command name
Returns: a method which implements that command
"""
BASE_COMMANDS = {'help': print_help}
REPLICATION_COMMANDS = {
'compare': replication_compare,
'dump': replication_dump,
'livecopy': replication_livecopy,
'load': replication_load,
'size': replication_size,
}
commands = {}
for command_set in (BASE_COMMANDS, REPLICATION_COMMANDS):
commands.update(command_set)
try:
command = commands[command_name]
except KeyError:
parser.print_usage()
sys.exit(_("Unknown command: %s") % command_name)
return command
if __name__ == '__main__':
usage = """
%%prog <command> [options] [args]
%s
""" % COMMANDS
oparser = optparse.OptionParser(version='%%prog',
usage=usage.strip())
# Options
oparser.add_option('-c', '--chunksize', action="store", default=65536,
help="Amount of data to transfer per HTTP write")
oparser.add_option('-d', '--debug', action="store_true", default=False,
help="Print debugging information")
oparser.add_option('-D', '--dontreplicate', action="store",
default=('created_at date deleted_at location '
'updated_at'),
help="List of fields to not replicate")
oparser.add_option('-m', '--metaonly', action="store_true", default=False,
help="Only replicate metadata, not images")
oparser.add_option('-l', '--logfile', action="store", default='',
help="Path of file to log to")
oparser.add_option('-s', '--syslog', action="store_true", default=False,
help="Log to syslog instead of a file")
oparser.add_option('-t', '--token', action="store", default='',
help=("Pass in your authentication token if you have "
"one"))
oparser.add_option('-v', '--verbose', action="store_true", default=False,
help="Print more verbose output")
(options, command, args) = parse_options(oparser, sys.argv[1:])
# Setup logging
root_logger = logging.root
if options.debug:
root_logger.setLevel(logging.DEBUG)
elif options.verbose:
root_logger.setLevel(logging.INFO)
else:
root_logger.setLevel(logging.WARNING)
formatter = logging.Formatter()
if options.syslog:
handler = logging.handlers.SysLogHandler(address='/dev/log')
elif options.logfile:
handler = logging.handlers.WatchedFileHandler(options.logfile)
else:
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(formatter)
root_logger.addHandler(handler)
command(options, args)

View File

@ -0,0 +1,208 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2012 Michael Still and Canonical Inc
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import copy
import imp
import json
import os
import StringIO
import sys
from glance.tests import utils as test_utils
TOPDIR = os.path.normpath(os.path.join(
os.path.dirname(os.path.abspath(__file__)),
os.pardir,
os.pardir,
os.pardir))
GLANCE_REPLICATOR_PATH = os.path.join(TOPDIR, 'bin', 'glance-replicator')
sys.dont_write_bytecode = True
glance_replicator = imp.load_source('glance_replicator',
GLANCE_REPLICATOR_PATH)
sys.dont_write_bytecode = False
IMG_RESPONSE_ACTIVE = {'content-length': '0',
'property-image_state': 'available',
'min_ram': '0',
'disk_format': 'aki',
'updated_at': '2012-06-25T02:10:36',
'date': 'Thu, 28 Jun 2012 07:20:05 GMT',
'owner': '8aef75b5c0074a59aa99188fdb4b9e90',
'id': '6d55dd55-053a-4765-b7bc-b30df0ea3861',
'size': '4660272',
'property-image_location':
('ubuntu-bucket/oneiric-server-cloudimg-amd64-'
'vmlinuz-generic.manifest.xml'),
'property-architecture': 'x86_64',
'etag': 'f46cfe7fb3acaff49a3567031b9b53bb',
'location':
('http://127.0.0.1:9292/v1/images/'
'6d55dd55-053a-4765-b7bc-b30df0ea3861'),
'container_format': 'aki',
'status': 'active',
'deleted': 'False',
'min_disk': '0',
'is_public': 'False',
'name':
('ubuntu-bucket/oneiric-server-cloudimg-amd64-'
'vmlinuz-generic'),
'checksum': 'f46cfe7fb3acaff49a3567031b9b53bb',
'created_at': '2012-06-25T02:10:32',
'protected': 'False',
'content-type': 'text/html; charset=UTF-8'
}
IMG_RESPONSE_QUEUED = copy.copy(IMG_RESPONSE_ACTIVE)
IMG_RESPONSE_QUEUED['status'] = 'queued'
IMG_RESPONSE_QUEUED['id'] = '49b2c782-ee10-4692-84f8-3942e9432c4b'
IMG_RESPONSE_QUEUED['location'] = ('http://127.0.0.1:9292/v1/images/'
+ IMG_RESPONSE_QUEUED['id'])
class FakeHTTPConnection(object):
def __init__(self):
self.count = 0
self.reqs = {}
self.last_req = None
self.host = 'localhost'
self.port = 9292
def prime_request(self, method, url, in_body, in_headers,
out_body, out_headers):
hkeys = in_headers.keys()
hkeys.sort()
hashable = (method, url, in_body, ' '.join(hkeys))
flat_headers = []
for key in out_headers:
flat_headers.append((key, out_headers[key]))
self.reqs[hashable] = (out_body, flat_headers)
def request(self, method, url, body, headers):
self.count += 1
hkeys = headers.keys()
hkeys.sort()
hashable = (method, url, body, ' '.join(hkeys))
if not hashable in self.reqs:
options = []
for h in self.reqs:
options.append(repr(h))
raise Exception('No such primed request: %s "%s"\n'
'%s\n\n'
'Available:\n'
'%s'
% (method, url, hashable, '\n\n'.join(options)))
self.last_req = hashable
def getresponse(self):
class FakeResponse(object):
def __init__(self, (body, headers)):
self.body = StringIO.StringIO(body)
self.headers = headers
self.status = 200
def read(self, count=1000000):
return self.body.read(count)
def getheaders(self):
return self.headers
return FakeResponse(self.reqs[self.last_req])
class ImageServiceTestCase(test_utils.BaseTestCase):
def test_rest_get_images(self):
c = glance_replicator.ImageService(FakeHTTPConnection(), 'noauth')
# Two images, one of which is queued
resp = {'images': [IMG_RESPONSE_ACTIVE, IMG_RESPONSE_QUEUED]}
c.conn.prime_request('GET', 'v1/images/detail?is_public=None', '',
{'x-auth-token': 'noauth'},
json.dumps(resp), {})
c.conn.prime_request('GET',
('v1/images/detail?marker=%s&is_public=None'
% IMG_RESPONSE_QUEUED['id']),
'', {'x-auth-token': 'noauth'},
json.dumps({'images': []}), {})
imgs = list(c.get_images())
self.assertEquals(len(imgs), 2)
self.assertEquals(c.conn.count, 2)
def test_rest_get_image(self):
c = glance_replicator.ImageService(FakeHTTPConnection(), 'noauth')
image_contents = 'THISISTHEIMAGEBODY'
c.conn.prime_request('GET',
'v1/images/%s' % IMG_RESPONSE_ACTIVE['id'],
'', {'x-auth-token': 'noauth'},
image_contents, IMG_RESPONSE_ACTIVE)
body = c.get_image(IMG_RESPONSE_ACTIVE['id'])
self.assertEquals(body.read(), image_contents)
def test_rest_header_list_to_dict(self):
i = [('x-image-meta-banana', 42), ('gerkin', 12)]
o = glance_replicator.ImageService._header_list_to_dict(i)
self.assertTrue('banana' in o)
self.assertTrue('gerkin' in o)
self.assertFalse('x-image-meta-banana' in o)
def test_rest_get_image_meta(self):
c = glance_replicator.ImageService(FakeHTTPConnection(), 'noauth')
c.conn.prime_request('HEAD',
'v1/images/%s' % IMG_RESPONSE_ACTIVE['id'],
'', {'x-auth-token': 'noauth'},
'', IMG_RESPONSE_ACTIVE)
header = c.get_image_meta(IMG_RESPONSE_ACTIVE['id'])
self.assertTrue('id' in header)
def test_rest_dict_to_headers(self):
i = {'banana': 42,
'gerkin': 12}
o = glance_replicator.ImageService._dict_to_headers(i)
self.assertTrue('x-image-meta-banana' in o)
self.assertTrue('x-image-meta-gerkin' in o)
def test_rest_add_image(self):
c = glance_replicator.ImageService(FakeHTTPConnection(), 'noauth')
image_body = 'THISISANIMAGEBODYFORSURE!'
image_meta_with_proto = {}
image_meta_with_proto['x-auth-token'] = 'noauth'
image_meta_with_proto['Content-Type'] = 'application/octet-stream'
image_meta_with_proto['Content-Length'] = len(image_body)
for key in IMG_RESPONSE_ACTIVE:
image_meta_with_proto['x-image-meta-%s' % key] = \
IMG_RESPONSE_ACTIVE[key]
c.conn.prime_request('POST', 'v1/images',
image_body, image_meta_with_proto,
'', IMG_RESPONSE_ACTIVE)
headers, body = c.add_image(IMG_RESPONSE_ACTIVE, image_body)
self.assertEquals(headers, IMG_RESPONSE_ACTIVE)
self.assertEquals(c.conn.count, 1)