Merge lp:~cerberus/nova/xs_migrations into lp:~hudson-openstack/nova/trunk

[Matt Dietz]

Implementation for XenServer migrations. There are several places for optimization but I based the current implementation on the chance scheduler just to be safe. Beyond that, a few features are missing, such as ensuring the IP address is transferred along with the migrated instance. This will be added in a subsequent patch. Finally, everything is implemented through the Openstack API resize hooks, but actual resizing of the instance RAM and hard drive space is not yet implemented.

[Rick Harris (rconradharris)]

== Meta ==
First of all, really nice job. There's a lot of code here, so I am going to review this in passes, starting from the bottom (code I'm a bit more familiar with).

I looks like you and I are sharing quite a bit in terms of utility code for working with VHDs, executing commands from subprocess, etc. We should definitely get together and see if we can create a useful library we can both share. Perhaps this can be phase 2 of both of our patches.

== Review ==

1397 +def get_sr_path(session):
1398 + sr_ref = find_sr(session)
1399 +
1400 + if sr_ref is None:
1401 + raise Exception('Cannot find SR to read VDI from')
1402 +
1403 + sr_rec = session.xenapi.SR.get_record(sr_ref)
1404 + sr_uuid = sr_rec["uuid"]
1405 + sr_path = os.path.join(FILE_SR_PATH, sr_uuid)
1406 + return sr_path
1407 +
1408 +def find_sr(session):
1409 + host = get_this_host(session)
1410 + srs = session.xenapi.SR.get_all()
1411 + for sr in srs:
1412 + sr_rec = session.xenapi.SR.get_record(sr)
1413 + if not ('i18n-key' in sr_rec['other_config'] and
1414 + sr_rec['other_config']['i18n-key'] == 'local-storage'):
1415 + continue
1416 + for pbd in sr_rec['PBDs']:
1417 + pbd_rec = session.xenapi.PBD.get_record(pbd)
1418 + if pbd_rec['host'] == host:
1419 + return sr
1420 + return None

This code already exists `vm_utils`. Rather than duplicate here, it might be better to pass `sr_path` in as an argument to your plugin functions. (This is the approach I took with xs-unified-images).

This also has the advantage in that I added a FLAG for SR's base_path (/var/run/sr-mount), so it's only specified in one place now.

1491 + rsync_args = ['nohup', RSYNC, '-av', '--progress', '-e', ssh_cmd,
1492 + source_path, dest_path]

Optional, but you can use the `shlex` module to convert a normal command string into the args list. Makes it a little easier to read/deal with.

1499 + if rsync_proc.returncode != 0:
1500 + raise Exception("Unexpected VHD transfer failure")

I notice that we're both using subprocess to execute commands and then trying to handle non-zero exit codes appropriately. I wrote a method called `assert_subprocess_success` in xs-unified-images. It might be worth it to refactor this (and perhaps other similar utility code) to a file we can both share.

1462 + subprocess.call([VHD_UTIL, 'modify', '-n', new_cow_path, '-p',
1463 + new_base_copy_path])

Another candidate for a function we can share. Perhaps we need a `link_vhds` method (this exists in xs-unified-images) inside this utility file. Is pluginlib_nova the right place for this?

1434 + sr_temp_path = "%s/images/" % sr_path

The pattern you're using here of creating a temp_path is similar to what I'm doing (again :). We can definitely re-use some code here. `_make_staging_area` and `cleanup_staging_area` seem like candidates for utility file.

1389 +SSH_HOSTS = '/root/.ssh/known_hosts'
1390 +DEVNULL = '/dev/null'
1391 +KEYSCAN = '/usr/bin/ssh-keyscan'
1392 +RSYNC = '/usr/bin/rsync'

Lots of these are things that don't look likely to be tweaked and don't...

review: Needs Fixing

[Jay Pipes (jaypipes)]

Hi! Please set the merge proposal status to Work In Progress while you work on any changes, and then set back to Needs Review when you've pushed changes. This way, reviewers get notified. Thanks, Matt!

[Matt Dietz (cerberus)]

Rick: I'm pretty sure we need the default_sr_scan idea because the instance in question doesn't actually exist on the destination machine during the step in which we link the VHDs. It might be possible refactor, but with the way the VM spawn code works, I don't think it would be trivial.

[Rick Harris (rconradharris)]

> 957 + # TODO(mdietz): replace this with the flag once unified-images merges
> 958 + return '/var/run/sr-mount/%s' % cls.get_sr(session, sr_label)

xs-unified is merged now. :-)

> 968 + @classmethod
> 969 + def scan_sr(cls, session, instance_id=None, sr_ref=None):
> 970 + """Scans the SR specified by sr_ref"""
> 971 + if sr_ref:
> 972 + LOG.debug(_("Re-scanning SR %s"), sr_ref)
> 973 + task = session.call_xenapi('Async.SR.scan', sr_ref)
> 974 + session.wait_for_task(instance_id, task)

Not sure I understand what's going on here. It looks like, sr_ref isn't
required; if it's not passed, this is a NOOP, if it is passed, it scans the
sr_ref.

I think scan_sr should take sr_ref as a positional, required argument and then
always scan the desired sr. Since `wait for task` takes an instance_id, that
needs to be required as well.

> I'm pretty sure we need the default_sr_scan idea because the instance in question doesn't actually exist on the destination machine

Hmm, I'm not sure I understand here. Is this because of the instance_id
argument?

> 948 + def get_sr(cls, session, sr_label='slices'):

Since multiple SRs can exist on a XS machine, I think it might be better to
call it `get_default_sr` for now to clarify that this is just one of many and
in some way a 'special' SR.

> 1293 - 'x-image-meta-type': 'vhd'
> 1294 - }
> 1295 + 'x-image-meta-type': 'vhd', }

Not sure about this change. Seems messy to leave that spurious comma in there
(even though it's syntactically valid).

> 1224 + task = None

Line is unecessary since `task` is guaranteed to be defined by if/else block.

> 1192 + 'new_cow_uuid': new_cow_uuid,

Should probably be using one space here, 'key': 'value'.

PEP8:
    - More than one space around an assignment (or other) operator to
      align it with another.

      Yes:

          x = 1
          y = 2
          long_variable = 3

      No:

          x = 1
          y = 2
          long_variable = 3

> 1093 + with self._get_snapshot(instance) as snapshot:

I'm mixed on this, context-managers are cool and all, but in this case,
creating the Snapshot class just to create a contextmanager seems like
overkill.

I like the idea of having a Snapshot class, but, I'd prefer for it inherit much
more of the functionality to create/destroy/upload snapshots (basically OOify
more of the code). Thoughts?

1071 + #Have a look at the VDI and see if it has a PV kernel

Femto-nit: needs a space betwee # and first letter of comment

> 104 + raise exception.Error(_("No finished migrations found for \
> 105 + instance"))

Should this be exception.NotFound()?

[Matt Dietz (cerberus)]

Wow, good feedback

Regarding the scan_default_sr concept, yes, I'm referring to the instance_id. Specifically the lack of an actual instance to infer the SR from or associate with wait_for_task when rescanning the SR.

I agree there would be a lot of value in a dedicated snapshot class. I'm hesitant to create one because it would likely be a substantial patch to get right (Since we would want to implement it abstractly with virt specific calls.) Could we agree to a refactor bug/blueprint down the road that makes better sense of this? If you'd like, I can back out the context manager change. I was simply trying to avoid having to try finally the snapshot deletion, as it seemed uglier than I what I did.

> 1295 + 'x-image-meta-type': 'vhd', }

Regarding this, I make this for two reasons. It saves me from stupid syntax errors when I add something to the list, and there's a lower chance that PEP8 is going to find extra whitespace at the end of the line. I could really go either way on that though.

I'll get on making the rest of the changes soon. Thanks!

[Rick Harris (rconradharris)]

> Regarding the scan_default_sr concept, yes, I'm referring to the instance_id

I'm sure there is a better way here. In an ideal world, I'd say we pair and try to come up with a solution to this, but, I think we're *both* pressed for time at the moment by other BPs. So, maybe we just throw a 'FIXME(sirp): refactor scan_default_sr in there', and come back to this.

I have this feeling we're starting to build up some technical debt in the XenAPI virt-layer, so, I'm hoping to do some cleanup refactorings in the coming weeks (OO-ification, etc).

> If you'd like, I can back out the context manager change.

I'd vote we back out the change, but I'm just one person. I'm curious other reviewers thing about this?

> It saves me from stupid syntax errors when I add something to the list, and there's a lower chance that PEP8 is going to find extra whitespace at the end of the line

Yeah, I've seen this done in other code bases for exactly this reason. Personally, I don't like the practice, but that' just personal preference, certainly not something that should be decree'd.

I think my real concern is that this isn't a Nova-wide convention. One of the strengths of our current code-base, IMHO, is that we have a pretty consistent style, and that greatly enhances its readability. Having some dicts use the trailing-comma and some not is a small increase in entropy, and that's a (small) lose.

If you think this is a good-style convention, I'd recommend that you propose this as a style-guideline and we debate it on the ML.

[Josh Kearney (jk0)]

Couple things I noticed -- the current migration number is at 008, so you'll want to move yours to 009. I'd also merge trunk and re-push since there have been some changes to vmops. _shutdown() will merge nicely with your change. There's also a _start() method now, but your _power_on() handles exceptions better, so I'd rename your method to _start() (since _start() is called at several places in vmops).

[Josh Kearney (jk0)]

Just a couple things:

930 + template_vm_ref = session.wait_for_task(instance_id, task)
974 + session.wait_for_task(instance_id, task)
1158 + self._session.wait_for_task(instance.id, task)
1170 + self._session.wait_for_task(instance.id, task)
1198 + self._session.wait_for_task(instance.id, task)

wait_for_task() now takes the `task` as the first arg and `id` as the second (so id can be None by default). This is likely why your unit tests were failing. You'll want to reverse the args on the lines above (wait_for_task(task, instance.id)) -- also may want to double check for others that I may have missed in your branch.

Other than that, I'd probably get rid of the extra space padding around your docstrings """[ ]Lorem Ipsum[ ]""", but that's mostly just a personal preference. Reason I like doing that is otherwise the extra spaces have to be stripped when you call __doc__.

review: Needs Fixing

[Josh Kearney (jk0)]

Looks great!

review: Approve

[Rick Harris (rconradharris)]

Code looks solid.

I think there is some room to refactor both the 'xs-unified' and 'xs-migration' branches to use some common code (quite a bit I think).

That said, getting some test failures (re-iterating here to capture this): http://paste.openstack.org/show/810/

review: Needs Fixing

[Matt Dietz (cerberus)]

Rick: I fixed the tests in question

[Rick Harris (rconradharris)]

lgtm

review: Approve

=== modified file 'nova/api/openstack/servers.py'
--- nova/api/openstack/servers.py	2011-03-01 16:53:19 +0000
+++ nova/api/openstack/servers.py	2011-03-07 17:23:15 +0000
@@ -203,10 +203,58 @@
         return exc.HTTPNoContent()
 
     def action(self, req, id):
-        """ Multi-purpose method used to reboot, rebuild, and
-        resize a server """
+        """Multi-purpose method used to reboot, rebuild, or
+        resize a server"""
+
+        actions = {
+            'reboot':        self._action_reboot,
+            'resize':        self._action_resize,
+            'confirmResize': self._action_confirm_resize,
+            'revertResize':  self._action_revert_resize,
+            'rebuild':       self._action_rebuild,
+            }
+
         input_dict = self._deserialize(req.body, req)
-        #TODO(sandy): rebuild/resize not supported.
+        for key in actions.keys():
+            if key in input_dict:
+                return actions[key](input_dict, req, id)
+        return faults.Fault(exc.HTTPNotImplemented())
+
+    def _action_confirm_resize(self, input_dict, req, id):
+        try:
+            self.compute_api.confirm_resize(req.environ['nova.context'], id)
+        except Exception, e:
+            LOG.exception(_("Error in confirm-resize %s"), e)
+            return faults.Fault(exc.HTTPBadRequest())
+        return exc.HTTPNoContent()
+
+    def _action_revert_resize(self, input_dict, req, id):
+        try:
+            self.compute_api.revert_resize(req.environ['nova.context'], id)
+        except Exception, e:
+            LOG.exception(_("Error in revert-resize %s"), e)
+            return faults.Fault(exc.HTTPBadRequest())
+        return exc.HTTPAccepted()
+
+    def _action_rebuild(self, input_dict, req, id):
+        return faults.Fault(exc.HTTPNotImplemented())
+
+    def _action_resize(self, input_dict, req, id):
+        """ Resizes a given instance to the flavor size requested """
+        try:
+            if 'resize' in input_dict and 'flavorId' in input_dict['resize']:
+                flavor_id = input_dict['resize']['flavorId']
+                self.compute_api.resize(req.environ['nova.context'], id,
+                        flavor_id)
+            else:
+                LOG.exception(_("Missing arguments for resize"))
+                return faults.Fault(exc.HTTPUnprocessableEntity())
+        except Exception, e:
+            LOG.exception(_("Error in resize %s"), e)
+            return faults.Fault(exc.HTTPBadRequest())
+        return faults.Fault(exc.HTTPAccepted())
+
+    def _action_reboot(self, input_dict, req, id):
         try:
             reboot_type = input_dict['reboot']['type']
         except Exception:

=== modified file 'nova/compute/api.py'
--- nova/compute/api.py	2011-03-03 18:21:54 +0000
+++ nova/compute/api.py	2011-03-07 17:23:15 +0000
@@ -404,6 +404,10 @@
         kwargs = {'method': method, 'args': params}
         return rpc.call(context, queue, kwargs)
 
+    def _cast_scheduler_message(self, context, args):
+        """Generic handler for RPC calls to the scheduler"""
+        rpc.cast(context, FLAGS.scheduler_topic, args)
+
     def snapshot(self, context, instance_id, name):
         """Snapshot the given instance.
 
@@ -420,6 +424,45 @@
         """Reboot the given instance."""
         self._cast_compute_message('reboot_instance', context, instance_id)
 
+    def revert_resize(self, context, instance_id):
+        """Reverts a resize, deleting the 'new' instance in the process"""
+        context = context.elevated()
+        migration_ref = self.db.migration_get_by_instance_and_status(context,
+                instance_id, 'finished')
+        if not migration_ref:
+            raise exception.NotFound(_("No finished migrations found for "
+                    "instance"))
+
+        params = {'migration_id': migration_ref['id']}
+        self._cast_compute_message('revert_resize', context, instance_id,
+                migration_ref['dest_compute'], params=params)
+
+    def confirm_resize(self, context, instance_id):
+        """Confirms a migration/resize, deleting the 'old' instance in the
+        process."""
+        context = context.elevated()
+        migration_ref = self.db.migration_get_by_instance_and_status(context,
+                instance_id, 'finished')
+        if not migration_ref:
+            raise exception.NotFound(_("No finished migrations found for "
+                    "instance"))
+        instance_ref = self.db.instance_get(context, instance_id)
+        params = {'migration_id': migration_ref['id']}
+        self._cast_compute_message('confirm_resize', context, instance_id,
+                migration_ref['source_compute'], params=params)
+
+        self.db.migration_update(context, migration_id,
+                {'status': 'confirmed'})
+        self.db.instance_update(context, instance_id,
+                {'host': migration_ref['dest_compute'], })
+
+    def resize(self, context, instance_id, flavor):
+        """Resize a running instance."""
+        self._cast_scheduler_message(context,
+                    {"method": "prep_resize",
+                     "args": {"topic": FLAGS.compute_topic,
+                              "instance_id": instance_id, }},)
+
     def pause(self, context, instance_id):
         """Pause the given instance."""
         self._cast_compute_message('pause_instance', context, instance_id)

=== modified file 'nova/compute/manager.py'
--- nova/compute/manager.py	2011-03-01 19:26:31 +0000
+++ nova/compute/manager.py	2011-03-07 17:23:15 +0000
@@ -413,6 +413,112 @@
 
     @exception.wrap_exception
     @checks_instance_lock
+    def confirm_resize(self, context, instance_id, migration_id):
+        """Destroys the source instance"""
+        context = context.elevated()
+        instance_ref = self.db.instance_get(context, instance_id)
+        migration_ref = self.db.migration_get(context, migration_id)
+        self.driver.destroy(instance_ref)
+
+    @exception.wrap_exception
+    @checks_instance_lock
+    def revert_resize(self, context, instance_id, migration_id):
+        """Destroys the new instance on the destination machine,
+        reverts the model changes, and powers on the old
+        instance on the source machine"""
+        instance_ref = self.db.instance_get(context, instance_id)
+        migration_ref = self.db.migration_get(context, migration_id)
+
+        #TODO(mdietz): we may want to split these into separate methods.
+        if migration_ref['source_compute'] == FLAGS.host:
+            self.driver._start(instance_ref)
+            self.db.migration_update(context, migration_id,
+                    {'status': 'reverted'})
+        else:
+            self.driver.destroy(instance_ref)
+            topic = self.db.queue_get_for(context, FLAGS.compute_topic,
+                    instance_ref['host'])
+            rpc.cast(context, topic,
+                    {'method': 'revert_resize',
+                     'args': {
+                           'migration_id': migration_ref['id'],
+                           'instance_id': instance_id, },
+                    })
+
+    @exception.wrap_exception
+    @checks_instance_lock
+    def prep_resize(self, context, instance_id):
+        """Initiates the process of moving a running instance to another
+        host, possibly changing the RAM and disk size in the process"""
+        context = context.elevated()
+        instance_ref = self.db.instance_get(context, instance_id)
+        if instance_ref['host'] == FLAGS.host:
+            raise exception.Error(_(
+                    'Migration error: destination same as source!'))
+
+        migration_ref = self.db.migration_create(context,
+                {'instance_id': instance_id,
+                 'source_compute': instance_ref['host'],
+                 'dest_compute': FLAGS.host,
+                 'dest_host':   self.driver.get_host_ip_addr(),
+                 'status':      'pre-migrating'})
+        LOG.audit(_('instance %s: migrating to '), instance_id,
+                context=context)
+        topic = self.db.queue_get_for(context, FLAGS.compute_topic,
+                instance_ref['host'])
+        rpc.cast(context, topic,
+                {'method': 'resize_instance',
+                 'args': {
+                       'migration_id': migration_ref['id'],
+                       'instance_id': instance_id, },
+                })
+
+    @exception.wrap_exception
+    @checks_instance_lock
+    def resize_instance(self, context, instance_id, migration_id):
+        """Starts the migration of a running instance to another host"""
+        migration_ref = self.db.migration_get(context, migration_id)
+        instance_ref = self.db.instance_get(context, instance_id)
+        self.db.migration_update(context, migration_id,
+                {'status': 'migrating', })
+
+        disk_info = self.driver.migrate_disk_and_power_off(instance_ref,
+                                  migration_ref['dest_host'])
+        self.db.migration_update(context, migration_id,
+                {'status': 'post-migrating', })
+
+        #TODO(mdietz): This is where we would update the VM record
+        #after resizing
+        service = self.db.service_get_by_host_and_topic(context,
+                migration_ref['dest_compute'], FLAGS.compute_topic)
+        topic = self.db.queue_get_for(context, FLAGS.compute_topic,
+                migration_ref['dest_compute'])
+        rpc.cast(context, topic,
+                {'method': 'finish_resize',
+                 'args': {
+                       'migration_id': migration_id,
+                       'instance_id': instance_id,
+                       'disk_info': disk_info, },
+                })
+
+    @exception.wrap_exception
+    @checks_instance_lock
+    def finish_resize(self, context, instance_id, migration_id, disk_info):
+        """Completes the migration process by setting up the newly transferred
+        disk and turning on the instance on its new host machine"""
+        migration_ref = self.db.migration_get(context, migration_id)
+        instance_ref = self.db.instance_get(context,
+                migration_ref['instance_id'])
+
+        # this may get passed into the following spawn instead
+        new_disk_info = self.driver.attach_disk(instance_ref, disk_info)
+        self.driver.spawn(instance_ref, disk=new_disk_info)
+
+        self.db.migration_update(context, migration_id,
+                {'status': 'finished', })
+
+    @exception.wrap_exception
+    @checks_instance_lock
     def pause_instance(self, context, instance_id):
         """Pause an instance on this server."""
         context = context.elevated()

=== modified file 'nova/db/api.py'
--- nova/db/api.py	2011-02-28 19:44:17 +0000
+++ nova/db/api.py	2011-03-07 17:23:15 +0000
@@ -80,10 +80,15 @@
 
 
 def service_get(context, service_id):
-    """Get an service or raise if it does not exist."""
+    """Get a service or raise if it does not exist."""
     return IMPL.service_get(context, service_id)
 
 
+def service_get_by_host_and_topic(context, host, topic):
+    """Get a service by host it's on and topic it listens to"""
+    return IMPL.service_get_by_host_and_topic(context, host, topic)
+
+
 def service_get_all(context, disabled=False):
     """Get all services."""
     return IMPL.service_get_all(context, disabled)
@@ -254,6 +259,28 @@
 
 ####################
 
+def migration_update(context, id, values):
+    """Update a migration instance"""
+    return IMPL.migration_update(context, id, values)
+
+
+def migration_create(context, values):
+    """Create a migration record"""
+    return IMPL.migration_create(context, values)
+
+
+def migration_get(context, migration_id):
+    """Finds a migration by the id"""
+    return IMPL.migration_get(context, migration_id)
+
+
+def migration_get_by_instance_and_status(context, instance_id, status):
+    """Finds a migration by the instance id its migrating"""
+    return IMPL.migration_get_by_instance_and_status(context, instance_id,
+            status)
+
+####################
+
 
 def fixed_ip_associate(context, address, instance_id):
     """Associate fixed ip to instance.

=== modified file 'nova/db/sqlalchemy/api.py'
--- nova/db/sqlalchemy/api.py	2011-03-03 00:37:02 +0000
+++ nova/db/sqlalchemy/api.py	2011-03-07 17:23:15 +0000
@@ -155,6 +155,17 @@
 
 
 @require_admin_context
+def service_get_by_host_and_topic(context, host, topic):
+    session = get_session()
+    return session.query(models.Service).\
+                   filter_by(deleted=False).\
+                   filter_by(disabled=False).\
+                   filter_by(host=host).\
+                   filter_by(topic=topic).\
+                   first()
+
+
+@require_admin_context
 def service_get_all_by_host(context, host):
     session = get_session()
     return session.query(models.Service).\
@@ -1972,6 +1983,51 @@
                        all()
 
 
+###################
+
+
+@require_admin_context
+def migration_create(context, values):
+    migration = models.Migration()
+    migration.update(values)
+    migration.save()
+    return migration
+
+
+@require_admin_context
+def migration_update(context, id, values):
+    session = get_session()
+    with session.begin():
+        migration = migration_get(context, id, session=session)
+        migration.update(values)
+        migration.save(session=session)
+        return migration
+
+
+@require_admin_context
+def migration_get(context, id, session=None):
+    if not session:
+        session = get_session()
+    result = session.query(models.Migration).\
+                     filter_by(id=id).first()
+    if not result:
+        raise exception.NotFound(_("No migration found with id %s")
+                % migration_id)
+    return result
+
+
+@require_admin_context
+def migration_get_by_instance_and_status(context, instance_id, status):
+    session = get_session()
+    result = session.query(models.Migration).\
+                     filter_by(instance_id=instance_id).\
+                     filter_by(status=status).first()
+    if not result:
+        raise exception.NotFound(_("No migration found with instance id %s")
+                % migration_id)
+    return result
+
+
 ##################
 
 

=== added file 'nova/db/sqlalchemy/migrate_repo/versions/009_add_instance_migrations.py'
--- nova/db/sqlalchemy/migrate_repo/versions/009_add_instance_migrations.py	1970-01-01 00:00:00 +0000
+++ nova/db/sqlalchemy/migrate_repo/versions/009_add_instance_migrations.py	2011-03-07 17:23:15 +0000
@@ -0,0 +1,61 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 OpenStack LLC.
+# 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.from sqlalchemy import *
+
+from sqlalchemy import *
+from migrate import *
+
+from nova import log as logging
+
+
+meta = MetaData()
+
+# Just for the ForeignKey and column creation to succeed, these are not the
+# actual definitions of instances or services.
+instances = Table('instances', meta,
+        Column('id', Integer(),  primary_key=True, nullable=False),
+        )
+
+#
+# New Tables
+#
+
+migrations = Table('migrations', meta,
+            Column('created_at', DateTime(timezone=False)),
+            Column('updated_at', DateTime(timezone=False)),
+            Column('deleted_at', DateTime(timezone=False)),
+            Column('deleted', Boolean(create_constraint=True, name=None)),
+            Column('id', Integer(), primary_key=True, nullable=False),
+            Column('source_compute', String(255)),
+            Column('dest_compute', String(255)),
+            Column('dest_host', String(255)),
+            Column('instance_id', Integer, ForeignKey('instances.id'),
+                nullable=True),
+            Column('status', String(255)),
+      )
+
+
+def upgrade(migrate_engine):
+    # Upgrade operations go here. Don't create your own engine;
+    # bind migrate_engine to your metadata
+    meta.bind = migrate_engine
+    for table in (migrations, ):
+        try:
+            table.create()
+        except Exception:
+            logging.info(repr(table))
+            logging.exception('Exception while creating table')
+            raise

=== modified file 'nova/db/sqlalchemy/migration.py'
--- nova/db/sqlalchemy/migration.py	2011-02-17 21:39:03 +0000
+++ nova/db/sqlalchemy/migration.py	2011-03-07 17:23:15 +0000
@@ -60,7 +60,7 @@
                           'key_pairs', 'networks', 'projects', 'quotas',
                           'security_group_instance_association',
                           'security_group_rules', 'security_groups',
-                          'services',
+                          'services', 'migrations',
                           'users', 'user_project_association',
                           'user_project_role_association',
                           'user_role_association',

=== modified file 'nova/db/sqlalchemy/models.py'
--- nova/db/sqlalchemy/models.py	2011-03-03 01:50:48 +0000
+++ nova/db/sqlalchemy/models.py	2011-03-07 17:23:15 +0000
@@ -389,6 +389,18 @@
     public_key = Column(Text)
 
 
+class Migration(BASE, NovaBase):
+    """Represents a running host-to-host migration."""
+    __tablename__ = 'migrations'
+    id = Column(Integer, primary_key=True, nullable=False)
+    source_compute = Column(String(255))
+    dest_compute = Column(String(255))
+    dest_host = Column(String(255))
+    instance_id = Column(Integer, ForeignKey('instances.id'), nullable=True)
+    #TODO(_cerberus_): enum
+    status = Column(String(255))
+
+
 class Network(BASE, NovaBase):
     """Represents a network."""
     __tablename__ = 'networks'
@@ -598,7 +610,7 @@
               Network, SecurityGroup, SecurityGroupIngressRule,
               SecurityGroupInstanceAssociation, AuthToken, User,
               Project, Certificate, ConsolePool, Console, Zone,
-              InstanceMetadata)
+              InstanceMetadata, Migration)
     engine = create_engine(FLAGS.sql_connection, echo=False)
     for model in models:
         model.metadata.create_all(engine)

=== modified file 'nova/rpc.py'
--- nova/rpc.py	2011-03-25 05:30:36 +0000
+++ nova/rpc.py	2011-03-07 17:23:15 +0000
@@ -123,7 +123,7 @@
                 LOG.error(_("Reconnected to queue"))
                 self.failed_connection = False
         # NOTE(vish): This is catching all errors because we really don't
-        #             exceptions to be logged 10 times a second if some
+        #             want exceptions to be logged 10 times a second if some
         #             persistent failure occurs.
         except Exception:  # pylint: disable-msg=W0703
             if not self.failed_connection:

=== added file 'nova/tests/api/openstack/common.py'
--- nova/tests/api/openstack/common.py	1970-01-01 00:00:00 +0000
+++ nova/tests/api/openstack/common.py	2011-03-07 17:23:15 +0000
@@ -0,0 +1,35 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2011 OpenStack LLC.
+# 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 json
+
+import webob
+
+
+def webob_factory(url):
+    """Factory for removing duplicate webob code from tests"""
+
+    base_url = url
+
+    def web_request(url, method=None, body=None):
+        req = webob.Request.blank("%s%s" % (base_url, url))
+        if method:
+            req.method = method
+        if body:
+            req.body = json.dumps(body)
+        return req
+    return web_request

=== modified file 'nova/tests/api/openstack/test_servers.py'
--- nova/tests/api/openstack/test_servers.py	2011-02-28 17:03:07 +0000
+++ nova/tests/api/openstack/test_servers.py	2011-03-07 17:23:15 +0000
@@ -1,6 +1,6 @@
 # vim: tabstop=4 shiftwidth=4 softtabstop=4
 
-# Copyright 2010 OpenStack LLC.
+# Copyright 2010-2011 OpenStack LLC.
 # All Rights Reserved.
 #
 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
@@ -26,10 +26,12 @@
 from nova import test
 import nova.api.openstack
 from nova.api.openstack import servers
+import nova.compute.api
 import nova.db.api
 from nova.db.sqlalchemy.models import Instance
 from nova.db.sqlalchemy.models import InstanceMetadata
 import nova.rpc
+from nova.tests.api.openstack import common
 from nova.tests.api.openstack import fakes
 
 
@@ -144,6 +146,8 @@
         self.stubs.Set(nova.compute.API, "get_actions", fake_compute_api)
         self.allow_admin = FLAGS.allow_admin_api
 
+        self.webreq = common.webob_factory('/v1.0/servers')
+
     def tearDown(self):
         self.stubs.UnsetAll()
         FLAGS.allow_admin_api = self.allow_admin
@@ -465,3 +469,99 @@
         res = req.get_response(fakes.wsgi_app())
         self.assertEqual(res.status, '202 Accepted')
         self.assertEqual(self.server_delete_called, True)
+
+    def test_resize_server(self):
+        req = self.webreq('/1/action', 'POST', dict(resize=dict(flavorId=3)))
+
+        self.resize_called = False
+
+        def resize_mock(*args):
+            self.resize_called = True
+
+        self.stubs.Set(nova.compute.api.API, 'resize', resize_mock)
+
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 202)
+        self.assertEqual(self.resize_called, True)
+
+    def test_resize_bad_flavor_fails(self):
+        req = self.webreq('/1/action', 'POST', dict(resize=dict(derp=3)))
+
+        self.resize_called = False
+
+        def resize_mock(*args):
+            self.resize_called = True
+
+        self.stubs.Set(nova.compute.api.API, 'resize', resize_mock)
+
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 422)
+        self.assertEqual(self.resize_called, False)
+
+    def test_resize_raises_fails(self):
+        req = self.webreq('/1/action', 'POST', dict(resize=dict(flavorId=3)))
+
+        def resize_mock(*args):
+            raise Exception('hurr durr')
+
+        self.stubs.Set(nova.compute.api.API, 'resize', resize_mock)
+
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 400)
+
+    def test_confirm_resize_server(self):
+        req = self.webreq('/1/action', 'POST', dict(confirmResize=None))
+
+        self.resize_called = False
+
+        def confirm_resize_mock(*args):
+            self.resize_called = True
+
+        self.stubs.Set(nova.compute.api.API, 'confirm_resize',
+                confirm_resize_mock)
+
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 204)
+        self.assertEqual(self.resize_called, True)
+
+    def test_confirm_resize_server_fails(self):
+        req = self.webreq('/1/action', 'POST', dict(confirmResize=None))
+
+        def confirm_resize_mock(*args):
+            raise Exception('hurr durr')
+
+        self.stubs.Set(nova.compute.api.API, 'confirm_resize',
+                confirm_resize_mock)
+
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 400)
+
+    def test_revert_resize_server(self):
+        req = self.webreq('/1/action', 'POST', dict(revertResize=None))
+
+        self.resize_called = False
+
+        def revert_resize_mock(*args):
+            self.resize_called = True
+
+        self.stubs.Set(nova.compute.api.API, 'revert_resize',
+                revert_resize_mock)
+
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 202)
+        self.assertEqual(self.resize_called, True)
+
+    def test_revert_resize_server_fails(self):
+        req = self.webreq('/1/action', 'POST', dict(revertResize=None))
+
+        def revert_resize_mock(*args):
+            raise Exception('hurr durr')
+
+        self.stubs.Set(nova.compute.api.API, 'revert_resize',
+                revert_resize_mock)
+
+        res = req.get_response(fakes.wsgi_app())
+        self.assertEqual(res.status_int, 400)
+
+if __name__ == "__main__":
+    unittest.main()

=== modified file 'nova/tests/test_compute.py'
--- nova/tests/test_compute.py	2011-02-28 17:25:14 +0000
+++ nova/tests/test_compute.py	2011-03-07 17:23:15 +0000
@@ -57,7 +57,7 @@
         self.manager.delete_project(self.project)
         super(ComputeTestCase, self).tearDown()
 
-    def _create_instance(self):
+    def _create_instance(self, params={}):
         """Create a test instance"""
         inst = {}
         inst['image_id'] = 'ami-test'
@@ -68,6 +68,7 @@
         inst['instance_type'] = 'm1.tiny'
         inst['mac_address'] = utils.generate_mac()
         inst['ami_launch_index'] = 0
+        inst.update(params)
         return db.instance_create(self.context, inst)['id']
 
     def _create_group(self):
@@ -268,9 +269,30 @@
 
         self.compute.terminate_instance(self.context, instance_id)
 
+    def test_resize_instance(self):
+        """Ensure instance can be migrated/resized"""
+        instance_id = self._create_instance()
+        context = self.context.elevated()
+        self.compute.run_instance(self.context, instance_id)
+        db.instance_update(self.context, instance_id, {'host': 'foo'})
+        self.compute.prep_resize(context, instance_id)
+        migration_ref = db.migration_get_by_instance_and_status(context,
+                instance_id, 'pre-migrating')
+        self.compute.resize_instance(context, instance_id,
+                migration_ref['id'])
+        self.compute.terminate_instance(context, instance_id)
+
     def test_get_by_flavor_id(self):
         type = instance_types.get_by_flavor_id(1)
         self.assertEqual(type, 'm1.tiny')
 
+    def test_resize_same_source_fails(self):
+        """Ensure instance fails to migrate when source and destination are
+        the same host"""
+        instance_id = self._create_instance()
+        self.compute.run_instance(self.context, instance_id)
+        self.assertRaises(exception.Error, self.compute.prep_resize,
+                self.context, instance_id)
+        self.compute.terminate_instance(self.context, instance_id)
         type = instance_types.get_by_flavor_id("1")
         self.assertEqual(type, 'm1.tiny')

=== modified file 'nova/tests/test_xenapi.py'
--- nova/tests/test_xenapi.py	2011-03-01 00:28:46 +0000
+++ nova/tests/test_xenapi.py	2011-03-07 17:23:15 +0000
@@ -346,6 +346,44 @@
         super(XenAPIDiffieHellmanTestCase, self).tearDown()
 
 
+class XenAPIMigrateInstance(test.TestCase):
+    """
+    Unit test for verifying migration-related actions
+    """
+
+    def setUp(self):
+        super(XenAPIMigrateInstance, self).setUp()
+        self.stubs = stubout.StubOutForTesting()
+        FLAGS.target_host = '127.0.0.1'
+        FLAGS.xenapi_connection_url = 'test_url'
+        FLAGS.xenapi_connection_password = 'test_pass'
+        db_fakes.stub_out_db_instance_api(self.stubs)
+        stubs.stub_out_get_target(self.stubs)
+        xenapi_fake.reset()
+        self.values = {'name': 1, 'id': 1,
+                  'project_id': 'fake',
+                  'user_id': 'fake',
+                  'image_id': 1,
+                  'kernel_id': 2,
+                  'ramdisk_id': 3,
+                  'instance_type': 'm1.large',
+                  'mac_address': 'aa:bb:cc:dd:ee:ff',
+                  }
+        stubs.stub_out_migration_methods(self.stubs)
+
+    def test_migrate_disk_and_power_off(self):
+        instance = db.instance_create(self.values)
+        stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests)
+        conn = xenapi_conn.get_connection(False)
+        conn.migrate_disk_and_power_off(instance, '127.0.0.1')
+
+    def test_attach_disk(self):
+        instance = db.instance_create(self.values)
+        stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests)
+        conn = xenapi_conn.get_connection(False)
+        conn.attach_disk(instance, {'base_copy': 'hurr', 'cow': 'durr'})
+
+
 class XenAPIDetermineDiskImageTestCase(test.TestCase):
     """
     Unit tests for code that detects the ImageType

=== modified file 'nova/tests/xenapi/stubs.py'
--- nova/tests/xenapi/stubs.py	2011-03-01 16:53:19 +0000
+++ nova/tests/xenapi/stubs.py	2011-03-07 17:23:15 +0000
@@ -20,6 +20,7 @@
 from nova.virt.xenapi import fake
 from nova.virt.xenapi import volume_utils
 from nova.virt.xenapi import vm_utils
+from nova.virt.xenapi import vmops
 
 
 def stubout_instance_snapshot(stubs):
@@ -217,3 +218,44 @@
 
     def SR_forget(self, _1, ref):
         pass
+
+
+class FakeSessionForMigrationTests(fake.SessionBase):
+    """Stubs out a XenAPISession for Migration tests"""
+    def __init__(self, uri):
+        super(FakeSessionForMigrationTests, self).__init__(uri)
+
+
+def stub_out_migration_methods(stubs):
+    def fake_get_snapshot(self, instance):
+        return 'foo', 'bar'
+
+    @classmethod
+    def fake_get_vdi(cls, session, vm_ref):
+        vdi_ref = fake.create_vdi(name_label='derp', read_only=False,
+                             sr_ref='herp', sharable=False)
+        vdi_rec = session.get_xenapi().VDI.get_record(vdi_ref)
+        return vdi_ref, {'uuid': vdi_rec['uuid'], }
+
+    def fake_shutdown(self, inst, vm, method='clean'):
+        pass
+
+    @classmethod
+    def fake_sr(cls, session, *args):
+        pass
+
+    @classmethod
+    def fake_get_sr_path(cls, *args):
+        return "fake"
+
+    def fake_destroy(*args, **kwargs):
+        pass
+
+    stubs.Set(vmops.VMOps, '_destroy', fake_destroy)
+    stubs.Set(vm_utils.VMHelper, 'scan_default_sr', fake_sr)
+    stubs.Set(vm_utils.VMHelper, 'scan_sr', fake_sr)
+    stubs.Set(vmops.VMOps, '_get_snapshot', fake_get_snapshot)
+    stubs.Set(vm_utils.VMHelper, 'get_vdi_for_vm_safely', fake_get_vdi)
+    stubs.Set(xenapi_conn.XenAPISession, 'wait_for_task', lambda x, y, z: None)
+    stubs.Set(vm_utils.VMHelper, 'get_sr_path', fake_get_sr_path)
+    stubs.Set(vmops.VMOps, '_shutdown', fake_shutdown)

=== modified file 'nova/virt/fake.py'
--- nova/virt/fake.py	2011-02-23 07:21:01 +0000
+++ nova/virt/fake.py	2011-03-07 17:23:15 +0000
@@ -139,6 +139,24 @@
         """
         pass
 
+    def get_host_ip_addr(self):
+        """
+        Retrieves the IP address of the dom0
+        """
+        pass
+
+    def resize(self, instance, flavor):
+        """
+        Resizes/Migrates the specified instance.
+
+        The flavor parameter determines whether or not the instance RAM and
+        disk space are modified, and if so, to what size.
+
+        The work will be done asynchronously. This function returns a task
+        that allows the caller to detect when it is complete.
+        """
+        pass
+
     def set_admin_password(self, instance, new_pass):
         """
         Set the root password on the specified instance.
@@ -179,6 +197,19 @@
         """
         pass
 
+    def migrate_disk_and_power_off(self, instance, dest):
+        """
+        Transfers the disk of a running instance in multiple phases, turning
+        off the instance before the end.
+        """
+        pass
+
+    def attach_disk(self, instance, disk_info):
+        """
+        Attaches the disk to an instance given the metadata disk_info
+        """
+        pass
+
     def pause(self, instance, callback):
         """
         Pause the specified instance.

=== modified file 'nova/virt/xenapi/fake.py'
--- nova/virt/xenapi/fake.py	2011-02-21 17:24:26 +0000
+++ nova/virt/xenapi/fake.py	2011-03-07 17:23:15 +0000
@@ -290,6 +290,9 @@
         #Always return 12GB available
         return 12 * 1024 * 1024 * 1024
 
+    def host_call_plugin(*args):
+        return 'herp'
+
     def xenapi_request(self, methodname, params):
         if methodname.startswith('login'):
             self._login(methodname, params)

=== modified file 'nova/virt/xenapi/vm_utils.py'
--- nova/virt/xenapi/vm_utils.py	2011-03-03 01:50:48 +0000
+++ nova/virt/xenapi/vm_utils.py	2011-03-07 17:23:15 +0000
@@ -253,16 +253,31 @@
         return vdi_ref
 
     @classmethod
+    def get_vdi_for_vm_safely(cls, session, vm_ref):
+        vdi_refs = VMHelper.lookup_vm_vdis(session, vm_ref)
+        if vdi_refs is None:
+            raise Exception(_("No VDIs found for VM %s") % vm_ref)
+        else:
+            num_vdis = len(vdi_refs)
+            if num_vdis != 1:
+                raise Exception(
+                        _("Unexpected number of VDIs (%(num_vdis)s) found"
+                        " for VM %(vm_ref)s") % locals())
+
+        vdi_ref = vdi_refs[0]
+        vdi_rec = session.get_xenapi().VDI.get_record(vdi_ref)
+        return vdi_ref, vdi_rec
+
+    @classmethod
     def create_snapshot(cls, session, instance_id, vm_ref, label):
-        """ Creates Snapshot (Template) VM, Snapshot VBD, Snapshot VDI,
-        Snapshot VHD
-        """
+        """Creates Snapshot (Template) VM, Snapshot VBD, Snapshot VDI,
+        Snapshot VHD"""
         #TODO(sirp): Add quiesce and VSS locking support when Windows support
         # is added
         LOG.debug(_("Snapshotting VM %(vm_ref)s with label '%(label)s'...")
                 % locals())
 
-        vm_vdi_ref, vm_vdi_rec = get_vdi_for_vm_safely(session, vm_ref)
+        vm_vdi_ref, vm_vdi_rec = cls.get_vdi_for_vm_safely(session, vm_ref)
         vm_vdi_uuid = vm_vdi_rec["uuid"]
         sr_ref = vm_vdi_rec["SR"]
 
@@ -270,7 +285,8 @@
 
         task = session.call_xenapi('Async.VM.snapshot', vm_ref, label)
         template_vm_ref = session.wait_for_task(task, instance_id)
-        template_vdi_rec = get_vdi_for_vm_safely(session, template_vm_ref)[1]
+        template_vdi_rec = cls.get_vdi_for_vm_safely(session,
+                template_vm_ref)[1]
         template_vdi_uuid = template_vdi_rec["uuid"]
 
         LOG.debug(_('Created snapshot %(template_vm_ref)s from'
@@ -285,6 +301,24 @@
         return template_vm_ref, template_vdi_uuids
 
     @classmethod
+    def get_sr(cls, session, sr_label='slices'):
+        """Finds the SR named by the given name label and returns
+        the UUID"""
+        return session.call_xenapi('SR.get_by_name_label', sr_label)[0]
+
+    @classmethod
+    def get_sr_path(cls, session):
+        """Return the path to our storage repository
+
+        This is used when we're dealing with VHDs directly, either by taking
+        snapshots or by restoring an image in the DISK_VHD format.
+        """
+        sr_ref = safe_find_sr(session)
+        sr_rec = session.get_xenapi().SR.get_record(sr_ref)
+        sr_uuid = sr_rec["uuid"]
+        return os.path.join(FLAGS.xenapi_sr_base_path, sr_uuid)
+
+    @classmethod
     def upload_image(cls, session, instance_id, vdi_uuids, image_id):
         """ Requests that the Glance plugin bundle the specified VDIs and
         push them into Glance using the specified human-friendly name.
@@ -298,7 +332,7 @@
                   'image_id': image_id,
                   'glance_host': FLAGS.glance_host,
                   'glance_port': FLAGS.glance_port,
-                  'sr_path': get_sr_path(session)}
+                  'sr_path': cls.get_sr_path(session)}
 
         kwargs = {'params': pickle.dumps(params)}
         task = session.async_call_plugin('glance', 'upload_vhd', kwargs)
@@ -341,13 +375,13 @@
                   'glance_host': FLAGS.glance_host,
                   'glance_port': FLAGS.glance_port,
                   'uuid_stack': uuid_stack,
-                  'sr_path': get_sr_path(session)}
+                  'sr_path': cls.get_sr_path(session)}
 
         kwargs = {'params': pickle.dumps(params)}
         task = session.async_call_plugin('glance', 'download_vhd', kwargs)
         vdi_uuid = session.wait_for_task(task, instance_id)
 
-        scan_sr(session, instance_id, sr_ref)
+        cls.scan_sr(session, instance_id, sr_ref)
 
         # Set the name-label to ease debugging
         vdi_ref = session.get_xenapi().VDI.get_by_uuid(vdi_uuid)
@@ -609,6 +643,21 @@
         except cls.XenAPI.Failure as e:
             return {"Unable to retrieve diagnostics": e}
 
+    @classmethod
+    def scan_sr(cls, session, instance_id=None, sr_ref=None):
+        """Scans the SR specified by sr_ref"""
+        if sr_ref:
+            LOG.debug(_("Re-scanning SR %s"), sr_ref)
+            task = session.call_xenapi('Async.SR.scan', sr_ref)
+            session.wait_for_task(task, instance_id)
+
+    @classmethod
+    def scan_default_sr(cls, session):
+        """Looks for the system default SR and triggers a re-scan"""
+        #FIXME(sirp/mdietz): refactor scan_default_sr in there
+        sr_ref = cls.get_sr(session)
+        session.call_xenapi('SR.scan', sr_ref)
+
 
 def get_rrd(host, uuid):
     """Return the VM RRD XML as a string"""
@@ -651,12 +700,6 @@
         return None
 
 
-def scan_sr(session, instance_id, sr_ref):
-    LOG.debug(_("Re-scanning SR %s"), sr_ref)
-    task = session.call_xenapi('Async.SR.scan', sr_ref)
-    session.wait_for_task(task, instance_id)
-
-
 def wait_for_vhd_coalesce(session, instance_id, sr_ref, vdi_ref,
                           original_parent_uuid):
     """ Spin until the parent VHD is coalesced into its parent VHD
@@ -681,7 +724,7 @@
                     " %(max_attempts)d), giving up...") % locals())
             raise exception.Error(msg)
 
-        scan_sr(session, instance_id, sr_ref)
+        VMHelper.scan_sr(session, instance_id, sr_ref)
         parent_uuid = get_vhd_parent_uuid(session, vdi_ref)
         if original_parent_uuid and (parent_uuid != original_parent_uuid):
             LOG.debug(_("Parent %(parent_uuid)s doesn't match original parent"
@@ -738,18 +781,6 @@
     return None
 
 
-def get_sr_path(session):
-    """Return the path to our storage repository
-
-    This is used when we're dealing with VHDs directly, either by taking
-    snapshots or by restoring an image in the DISK_VHD format.
-    """
-    sr_ref = safe_find_sr(session)
-    sr_rec = session.get_xenapi().SR.get_record(sr_ref)
-    sr_uuid = sr_rec["uuid"]
-    return os.path.join(FLAGS.xenapi_sr_base_path, sr_uuid)
-
-
 def remap_vbd_dev(dev):
     """Return the appropriate location for a plugged-in VBD device
 

=== modified file 'nova/virt/xenapi/vmops.py'
--- nova/virt/xenapi/vmops.py	2011-03-02 23:37:31 +0000
+++ nova/virt/xenapi/vmops.py	2011-03-07 17:23:15 +0000
@@ -22,6 +22,7 @@
 import json
 import M2Crypto
 import os
+import pickle
 import subprocess
 import tempfile
 import uuid
@@ -61,7 +62,17 @@
                 vms.append(rec["name_label"])
         return vms
 
-    def spawn(self, instance):
+    def _start(self, instance, vm_ref=None):
+        """Power on a VM instance"""
+        if not vm_ref:
+            vm_ref = VMHelper.lookup(self._session, instance.name)
+        if vm_ref is None:
+            raise exception(_('Attempted to power on non-existent instance'
+            ' bad instance id %s') % instance.id)
+        LOG.debug(_("Starting instance %s"), instance.name)
+        self._session.call_xenapi('VM.start', vm_ref, False, False)
+
+    def spawn(self, instance, disk):
         """Create VM instance"""
         instance_name = instance.name
         vm = VMHelper.lookup(self._session, instance_name)
@@ -81,16 +92,22 @@
         user = AuthManager().get_user(instance.user_id)
         project = AuthManager().get_project(instance.project_id)
 
-        disk_image_type = VMHelper.determine_disk_image_type(instance)
-
-        vdi_uuid = VMHelper.fetch_image(self._session, instance.id,
-            instance.image_id, user, project, disk_image_type)
-
-        vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', vdi_uuid)
-
-        pv_kernel = False
+        vdi_ref = kernel = ramdisk = pv_kernel = None
+
+        # Are we building from a pre-existing disk?
+        if not disk:
+            #if kernel is not present we must download a raw disk
+
+            disk_image_type = VMHelper.determine_disk_image_type(instance)
+            vdi_uuid = VMHelper.fetch_image(self._session, instance.id,
+                    instance.image_id, user, project, disk_image_type)
+            vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', vdi_uuid)
+
+        else:
+            vdi_ref = self._session.call_xenapi('VDI.get_by_uuid', disk)
+
         if disk_image_type == ImageType.DISK_RAW:
-            #Have a look at the VDI and see if it has a PV kernel
+            # Have a look at the VDI and see if it has a PV kernel
             pv_kernel = VMHelper.lookup_image(self._session, instance.id,
                                               vdi_ref)
         elif disk_image_type == ImageType.DISK_VHD:
@@ -98,19 +115,18 @@
             # configurable as Windows will use HVM.
             pv_kernel = True
 
-        kernel = None
         if instance.kernel_id:
             kernel = VMHelper.fetch_image(self._session, instance.id,
                 instance.kernel_id, user, project, ImageType.KERNEL_RAMDISK)
 
-        ramdisk = None
         if instance.ramdisk_id:
             ramdisk = VMHelper.fetch_image(self._session, instance.id,
                 instance.ramdisk_id, user, project, ImageType.KERNEL_RAMDISK)
 
         vm_ref = VMHelper.create_vm(self._session,
                                           instance, kernel, ramdisk, pv_kernel)
-        VMHelper.create_vbd(self._session, vm_ref, vdi_ref, 0, True)
+        VMHelper.create_vbd(session=self._session, vm_ref=vm_ref,
+                vdi_ref=vdi_ref, userdevice=0, bootable=True)
 
         # inject_network_info and create vifs
         networks = self.inject_network_info(instance)
@@ -217,7 +233,7 @@
             "start")
 
     def snapshot(self, instance, image_id):
-        """ Create snapshot from a running VM instance
+        """Create snapshot from a running VM instance
 
         :param instance: instance to be snapshotted
         :param image_id: id of image to upload to
@@ -238,7 +254,20 @@
             that will bundle the VHDs together and then push the bundle into
             Glance.
         """
-
+        template_vm_ref = None
+        try:
+            template_vm_ref, template_vdi_uuids = self._get_snapshot(instance)
+            # call plugin to ship snapshot off to glance
+            VMHelper.upload_image(
+                    self._session, instance.id, template_vdi_uuids, image_id)
+        finally:
+            if template_vm_ref:
+                self._destroy(instance, template_vm_ref,
+                        shutdown=False, destroy_kernel_ramdisk=False)
+
+        logging.debug(_("Finished snapshot and upload for VM %s"), instance)
+
+    def _get_snapshot(self, instance):
         #TODO(sirp): Add quiesce and VSS locking support when Windows support
         # is added
 
@@ -249,20 +278,89 @@
         try:
             template_vm_ref, template_vdi_uuids = VMHelper.create_snapshot(
                 self._session, instance.id, vm_ref, label)
+            return template_vm_ref, template_vdi_uuids
         except self.XenAPI.Failure, exc:
             logging.error(_("Unable to Snapshot %(vm_ref)s: %(exc)s")
                     % locals())
             return
 
+    def migrate_disk_and_power_off(self, instance, dest):
+        """Copies a VHD from one host machine to another
+
+        :param instance: the instance that owns the VHD in question
+        :param dest: the destination host machine
+        :param disk_type: values are 'primary' or 'cow'
+        """
+        vm_ref = VMHelper.lookup(self._session, instance.name)
+
+        # The primary VDI becomes the COW after the snapshot, and we can
+        # identify it via the VBD. The base copy is the parent_uuid returned
+        # from the snapshot creation
+
+        base_copy_uuid = cow_uuid = None
+        template_vdi_uuids = template_vm_ref = None
         try:
-            # call plugin to ship snapshot off to glance
-            VMHelper.upload_image(
-                self._session, instance.id, template_vdi_uuids, image_id)
+            # transfer the base copy
+            template_vm_ref, template_vdi_uuids = self._get_snapshot(instance)
+            base_copy_uuid = template_vdi_uuids[1]
+            vdi_ref, vm_vdi_rec = \
+                    VMHelper.get_vdi_for_vm_safely(self._session, vm_ref)
+            cow_uuid = vm_vdi_rec['uuid']
+
+            params = {'host': dest,
+                      'vdi_uuid': base_copy_uuid,
+                      'instance_id': instance.id,
+                      'sr_path': VMHelper.get_sr_path(self._session)}
+
+            task = self._session.async_call_plugin('migration', 'transfer_vhd',
+                    {'params': pickle.dumps(params)})
+            self._session.wait_for_task(task, instance.id)
+
+            # Now power down the instance and transfer the COW VHD
+            self._shutdown(instance, vm_ref, method='clean')
+
+            params = {'host': dest,
+                      'vdi_uuid': cow_uuid,
+                      'instance_id': instance.id,
+                      'sr_path': VMHelper.get_sr_path(self._session), }
+
+            task = self._session.async_call_plugin('migration', 'transfer_vhd',
+                    {'params': pickle.dumps(params)})
+            self._session.wait_for_task(task, instance.id)
+
         finally:
-            self._destroy(instance, template_vm_ref, shutdown=False,
-                          destroy_kernel_ramdisk=False)
-
-        logging.debug(_("Finished snapshot and upload for VM %s"), instance)
+            if template_vm_ref:
+                self._destroy(instance, template_vm_ref,
+                        shutdown=False, destroy_kernel_ramdisk=False)
+
+        # TODO(mdietz): we could also consider renaming these to something
+        # sensible so we don't need to blindly pass around dictionaries
+        return {'base_copy': base_copy_uuid, 'cow': cow_uuid}
+
+    def attach_disk(self, instance, disk_info):
+        """Links the base copy VHD to the COW via the XAPI plugin"""
+        vm_ref = VMHelper.lookup(self._session, instance.name)
+        new_base_copy_uuid = str(uuid.uuid4())
+        new_cow_uuid = str(uuid.uuid4())
+        params = {'instance_id': instance.id,
+                  'old_base_copy_uuid': disk_info['base_copy'],
+                  'old_cow_uuid': disk_info['cow'],
+                  'new_base_copy_uuid': new_base_copy_uuid,
+                  'new_cow_uuid': new_cow_uuid,
+                  'sr_path': VMHelper.get_sr_path(self._session), }
+
+        task = self._session.async_call_plugin('migration',
+                'move_vhds_into_sr', {'params': pickle.dumps(params)})
+        self._session.wait_for_task(task, instance.id)
+
+        # Now we rescan the SR so we find the VHDs
+        VMHelper.scan_default_sr(self._session)
+
+        return new_cow_uuid
+
+    def resize(self, instance, flavor):
+        """Resize a running instance by changing it's RAM and disk size """
+        raise NotImplementedError()
 
     def reboot(self, instance):
         """Reboot VM instance"""
@@ -308,11 +406,6 @@
             raise RuntimeError(resp_dict['message'])
         return resp_dict['message']
 
-    def _start(self, instance, vm):
-        """Start an instance"""
-        task = self._session.call_xenapi("Async.VM.start", vm, False, False)
-        self._session.wait_for_task(task, instance.id)
-
     def inject_file(self, instance, b64_path, b64_contents):
         """Write a file to the VM instance. The path to which it is to be
         written and the contents of the file need to be supplied; both should
@@ -355,8 +448,7 @@
             if hard:
                 task = self._session.call_xenapi("Async.VM.hard_shutdown", vm)
             else:
-                task = self._session.call_xenapi("Async.VM.clean_shutdown", vm)
-
+                task = self._session.call_xenapi('Async.VM.clean_shutdown', vm)
             self._session.wait_for_task(task, instance.id)
         except self.XenAPI.Failure, exc:
             LOG.exception(exc)

=== modified file 'nova/virt/xenapi_conn.py'
--- nova/virt/xenapi_conn.py	2011-03-01 16:53:19 +0000
+++ nova/virt/xenapi_conn.py	2011-03-07 17:23:15 +0000
@@ -154,14 +154,18 @@
         """List VM instances"""
         return self._vmops.list_instances()
 
-    def spawn(self, instance):
+    def spawn(self, instance, disk=None):
         """Create VM instance"""
-        self._vmops.spawn(instance)
+        self._vmops.spawn(instance, disk)
 
     def snapshot(self, instance, image_id):
         """ Create snapshot from a running VM instance """
         self._vmops.snapshot(instance, image_id)
 
+    def resize(self, instance, flavor):
+        """Resize a VM instance"""
+        raise NotImplementedError()
+
     def reboot(self, instance):
         """Reboot VM instance"""
         self._vmops.reboot(instance)
@@ -188,6 +192,15 @@
         """Unpause paused VM instance"""
         self._vmops.unpause(instance, callback)
 
+    def migrate_disk_and_power_off(self, instance, dest):
+        """Transfers the VHD of a running instance to another host, then shuts
+        off the instance copies over the COW disk"""
+        return self._vmops.migrate_disk_and_power_off(instance, dest)
+
+    def attach_disk(self, instance, disk_info):
+        """Moves the copied VDIs into the SR"""
+        return self._vmops.attach_disk(instance, disk_info)
+
     def suspend(self, instance, callback):
         """suspend the specified instance"""
         self._vmops.suspend(instance, callback)
@@ -228,6 +241,10 @@
         """Return link to instance's ajax console"""
         return self._vmops.get_ajax_console(instance)
 
+    def get_host_ip_addr(self):
+        xs_url = urlparse.urlparse(FLAGS.xenapi_connection_url)
+        return xs_url.netloc
+
     def attach_volume(self, instance_name, device_path, mountpoint):
         """Attach volume storage to VM instance"""
         return self._volumeops.attach_volume(instance_name,

=== added file 'plugins/xenserver/xenapi/etc/xapi.d/plugins/migration'
--- plugins/xenserver/xenapi/etc/xapi.d/plugins/migration	1970-01-01 00:00:00 +0000
+++ plugins/xenserver/xenapi/etc/xapi.d/plugins/migration	2011-03-07 17:23:15 +0000
@@ -0,0 +1,117 @@
+#!/usr/bin/env python
+
+# Copyright 2010 OpenStack LLC.
+# 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.
+
+"""
+XenAPI Plugin for transfering data between host nodes
+"""
+
+import os
+import os.path
+import pickle
+import shutil
+import subprocess
+
+import XenAPIPlugin
+
+from pluginlib_nova import *
+configure_logging('migration')
+
+
+def move_vhds_into_sr(session, args):
+    """Moves the VHDs from their copied location to the SR"""
+    params = pickle.loads(exists(args, 'params'))
+    instance_id = params['instance_id']
+
+    old_base_copy_uuid = params['old_base_copy_uuid']
+    old_cow_uuid = params['old_cow_uuid']
+
+    new_base_copy_uuid = params['new_base_copy_uuid']
+    new_cow_uuid = params['new_cow_uuid']
+
+    sr_path = params['sr_path']
+    sr_temp_path = "%s/images/" % sr_path
+
+    # Discover the copied VHDs locally, and then set up paths to copy
+    # them to under the SR
+    source_image_path = "%s/instance%d" % ('/images/', instance_id)
+    source_base_copy_path = "%s/%s.vhd" % (source_image_path,
+            old_base_copy_uuid)
+    source_cow_path = "%s/%s.vhd" % (source_image_path, old_cow_uuid)
+
+    temp_vhd_path = "%s/instance%d/" % (sr_temp_path, instance_id)
+    new_base_copy_path = "%s/%s.vhd" % (temp_vhd_path, new_base_copy_uuid)
+    new_cow_path = "%s/%s.vhd" % (temp_vhd_path, new_cow_uuid)
+
+    logging.debug('Creating temporary SR path %s' % temp_vhd_path)
+    os.makedirs(temp_vhd_path)
+
+    logging.debug('Moving %s into %s' % (source_base_copy_path, temp_vhd_path))
+    shutil.move(source_base_copy_path, new_base_copy_path)
+
+    logging.debug('Moving %s into %s' % (source_cow_path, temp_vhd_path))
+    shutil.move(source_cow_path, new_cow_path)
+
+    logging.debug('Cleaning up %s' % source_image_path)
+    os.rmdir(source_image_path)
+
+    # Link the COW to the base copy
+    logging.debug('Attaching COW to the base copy %s -> %s' %
+            (new_cow_path, new_base_copy_path))
+    subprocess.call(shlex.split('/usr/sbin/vhd-util modify -n %s -p %s' %
+            (new_cow_path, new_base_copy_path)))
+    logging.debug('Moving VHDs into SR %s' % sr_path)
+    shutil.move("%s/%s.vhd" % (temp_vhd_path, new_base_copy_uuid), sr_path)
+    shutil.move("%s/%s.vhd" % (temp_vhd_path, new_cow_uuid), sr_path)
+
+    logging.debug('Cleaning up temporary SR path %s' % temp_vhd_path)
+    os.rmdir(temp_vhd_path)
+    return ""
+
+
+def transfer_vhd(session, args):
+    """Rsyncs a VHD to an adjacent host"""
+    params = pickle.loads(exists(args, 'params'))
+    instance_id = params['instance_id']
+    host = params['host']
+    vdi_uuid = params['vdi_uuid']
+    sr_path = params['sr_path']
+    vhd_path = "%s.vhd" % vdi_uuid
+
+    source_path = "%s/%s" % (sr_path, vhd_path)
+    dest_path = '%s:%sinstance%d/' % (host, '/images/', instance_id)
+
+    logging.debug("Preparing to transmit %s to %s" % (source_path,
+            dest_path))
+
+    ssh_cmd = 'ssh -o StrictHostKeyChecking=no'
+
+    rsync_args = shlex.split('nohup /usr/bin/rsync -av --progress -e %s %s %s'
+            % (ssh_cmd, source_path, dest_path))
+
+    logging.debug('rsync %s' % (' '.join(rsync_args, )))
+
+    rsync_proc = subprocess.Popen(rsync_args, stdout=subprocess.PIPE)
+    logging.debug('Rsync output: \n %s' % rsync_proc.communicate()[0])
+    logging.debug('Rsync return: %d' % rsync_proc.returncode)
+    if rsync_proc.returncode != 0:
+        raise Exception("Unexpected VHD transfer failure")
+    return ""
+
+
+if __name__ == '__main__':
+    XenAPIPlugin.dispatch({'transfer_vhd': transfer_vhd,
+            'move_vhds_into_sr': move_vhds_into_sr, })