Merge lp:~justin-fathomdb/nova/virtualbox-support into lp:~hudson-openstack/nova/trunk

[justinsb]

Adds initial experimental support for VirtualBox. Currently doesn't work, because VirtualBox doesn't expose an async API. Will work once the Twisted removal is complete. Won't break anything unless users launch with the (undocumented) connection type 'virtualbox'.

[lp:~justin-fathomdb/nova/virtualbox-support]

166. By justinsb on 2010-08-20

Merged with trunk & lp:~justin-fathomdb/nova/better-error-messages-on-process-fail

167. By justinsb on 2010-08-20

Merged with updated kiss-backend (twisted for some services)

168. By justinsb on 2010-08-20

Merged with latest kiss-backend

169. By justinsb on 2010-08-20

Merged with latest kiss-backend

170. By justinsb on 2010-08-20

Merged with latest kiss-backend

171. By justinsb on 2010-08-20

Merged with latest lp:~justin-fathomdb/nova/better-error-messages-on-process-fail

172. By justinsb on 2010-08-21

Merged latest kiss-backend

173. By justinsb on 2010-08-21

Merged latest improved error message

174. By justinsb on 2010-08-21

Merged latest kiss-backend

175. By justinsb on 2010-08-21

Adding dev-scripts to make development easier

176. By justinsb on 2010-08-21

Added network to up/down helper scripts

177. By justinsb on 2010-08-21

Merged with latest ~justin-fathomdb/nova/kiss-backend/

178. By justinsb on 2010-08-21

Built disk and launched VM

179. By justinsb on 2010-08-21

Don't partition the disk if using raw images

180. By justinsb on 2010-08-21

Merged with latest lp:~justin-fathomdb/nova/raw-disk-images (though I'm fighting a mini merge-battle with myself because the code actually depends on the twisted removal)

181. By justinsb on 2010-08-21

Actually, it looks like the twisted removal & virtualbox support may be inextricably linked...

[Soren Hansen (soren)]

Twisted has no particular problem talking to a synchronous API. It can simply defer the call to a thread and get a response back as a callback once the call returns. What /is/ difficult is calling Twisted from a strictly synchronous application, but I don't see that as an issue here.

With that in mind, how is it that Twisted removal and VirtualBox support are related?

[justinsb (justin-fathomdb)]

The relational is the 'viral' nature of Twisted's async calls... when implementing an interface method that is currently synchronous (get_info, as I recall), that method needs to be made asynchronous if the implementation is now async (in this case, all VirtualBox methods must spawn the VirtualBoxManage command.) But then all its callers need to be made async, and then their callers etc. When I asked about this on the mailing list, it seems that the only thing to do was essentially to make _every_ method async, in case it ever called in future an interface method which had an async implementation. That's a pretty big lock-in to Twisted, and it seems that the momentum was moving against Twisted.

If this patch continued to work with Twisted, I think it counted it would need a dozen additional changes propagating new deferreds up the call chains.

Whatever the decision on Twisted, I'll work with it, but I do think we need _a_ decision. If it's to keep Twisted then I'll have to rework this patch, but that's OK... My vote is to go with Threads/Eventlet, because then the whole code isn't covered in Twisted-specifics and we have more options in future (Python Threads, Jython, Eventlet, source code translators) vs just Twisted.

Unmerged revisions

181. By justinsb on 2010-08-21

Actually, it looks like the twisted removal & virtualbox support may be inextricably linked...

180. By justinsb on 2010-08-21

Merged with latest lp:~justin-fathomdb/nova/raw-disk-images (though I'm fighting a mini merge-battle with myself because the code actually depends on the twisted removal)

179. By justinsb on 2010-08-21

Don't partition the disk if using raw images

178. By justinsb on 2010-08-21

Built disk and launched VM

177. By justinsb on 2010-08-21

Merged with latest ~justin-fathomdb/nova/kiss-backend/

176. By justinsb on 2010-08-21

Added network to up/down helper scripts

175. By justinsb on 2010-08-21

Adding dev-scripts to make development easier

174. By justinsb on 2010-08-21

Merged latest kiss-backend

173. By justinsb on 2010-08-21

Merged latest improved error message

172. By justinsb on 2010-08-21

Merged latest kiss-backend

=== modified file 'bin/nova-compute'
--- bin/nova-compute	2010-08-19 00:39:12 +0000
+++ bin/nova-compute	2010-08-21 14:09:42 +0000
@@ -24,9 +24,8 @@
 from nova import twistd
 from nova.compute import service
 
-
-if __name__ == '__main__':
-    twistd.serve(__file__)
-
-if __name__ == '__builtin__':
-    application = service.ComputeService.create() # pylint: disable-msg=C0103
+twistd.serve(__file__, twisted=False)
+
+service.ComputeService.create()
+
+service.ComputeService.main_loop()

=== added directory 'dev-scripts'
=== added file 'dev-scripts/api.sh'
--- dev-scripts/api.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/api.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,9 @@
+#!/bin/bash
+
+. ~/openstack/conf
+cd ${NOVA_SRC_BASE}
+
+NOVA_API_ARGS="--ca_path=${CA_DIR} --keys_path=${KEYS_PATH}  ${NETWORK_ARGS}" # --datastore_path=/var/lib/nova/keeper"
+echo NOVA_API_ARGS=${NOVA_API_ARGS}
+bin/nova-api ${NOVA_API_ARGS} --nodaemonize --verbose start
+

=== added file 'dev-scripts/compute.sh'
--- dev-scripts/compute.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/compute.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,19 @@
+#!/bin/bash
+
+. ~/openstack/conf
+cd ${NOVA_SRC_BASE}
+
+mkdir -p ${INSTANCES_PATH}
+mkdir -p ${NETWORKS_PATH}
+NOVA_COMPUTE_ARGS="--ca_path=${CA_DIR} --keys_path=${KEYS_PATH} ${NETWORK_ARGS} --images_path=${IMAGES_PATH} --use_s3=false --instances_path=${INSTANCES_PATH} --networks_path=${NETWORKS_PATH}" # --libvirt_xml_template=nova/compute/libvirt.xml.template --vpn_client_template=nova/cloudpipe/client.ovpn.template --credentials_template=nova/auth/novarc.template"
+# --datastore_path=/var/lib/nova/keeper"
+
+if [ -n $CONNECTION_TYPE ]; then
+    NOVA_COMPUTE_ARGS="${NOVA_COMPUTE_ARGS} --connection_type=${CONNECTION_TYPE}"
+fi
+
+echo NOVA_COMPUTE_ARGS=${NOVA_COMPUTE_ARGS}
+
+bin/nova-compute ${NOVA_COMPUTE_ARGS} --nodaemon --verbose start
+
+

=== added file 'dev-scripts/create-image.sh'
--- dev-scripts/create-image.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/create-image.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,17 @@
+#!/bin/bash
+
+apt-get --yes install apt-proxy ubuntu-vm-builder
+
+mkdir -p ~/openstack/images
+cd ~/openstack/images
+# If you're on 32 bit CPU, or if you don't have HVM (?), use 32 bit...
+# If you're not running a big-memory machine, you should probably be using 32 bit anyway...
+ARCH=i386
+sudo vmbuilder kvm ubuntu --addpkg openssh-server --install-mirror=http://127.0.0.1:9999/ubuntu --install-security-mirror=http://127.0.0.1:9999/ubuntu-security --debug  --arch ${ARCH}
+cd ubuntu-kvm
+# Sneaky link to get around random naming
+ln -s *.qcow2  root.qcow2 
+qemu-img info root.qcow2
+qemu-img convert -O vmdk root.qcow2 root.vmdk
+qemu-img info root.vmdk
+

=== added file 'dev-scripts/down.sh'
--- dev-scripts/down.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/down.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,8 @@
+#!/bin/bash
+
+pkill -f nova-api
+pkill -f nova-compute
+pkill -f nova-volume
+pkill -f nova-objectstore
+pkill -f nova-network
+pkill -f redis-server

=== added file 'dev-scripts/network.sh'
--- dev-scripts/network.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/network.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,7 @@
+#!/bin/bash
+
+. ~/openstack/conf
+cd ${NOVA_SRC_BASE}
+
+bin/nova-network --nodaemon --verbose
+

=== added file 'dev-scripts/objectstore.sh'
--- dev-scripts/objectstore.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/objectstore.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,15 @@
+#!/bin/bash
+
+. ~/openstack/conf
+cd ${NOVA_SRC_BASE}
+
+mkdir -p ${BUCKETS_PATH}
+mkdir -p ${IMAGES_PATH}
+NOVA_OBJECTSTORE_ARGS="--ca_path=${CA_DIR} --keys_path=${KEYS_PATH} ${NETWORK_ARGS} --images_path=${IMAGES_PATH} --buckets_path=${BUCKETS_PATH}" # --datastore_path=/var/lib/nova/keeper"
+# This hack is because otherwise the compute deamon also writes to twistd.log
+mkdir -p objectstorelogs
+cd objectstorelogs
+# PYTHONPATH needs to be set to '..' to find the modules when running this way.. (overriding the 'conf' file at the top) -- comstud
+PYTHONPATH=".."
+../bin/nova-objectstore ${NOVA_OBJECTSTORE_ARGS} --nodaemon --verbose start
+

=== added file 'dev-scripts/redis.sh'
--- dev-scripts/redis.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/redis.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,3 @@
+#!/bin/bash
+
+~/openstack/src/redis-latest/redis-server

=== added file 'dev-scripts/restart.sh'
--- dev-scripts/restart.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/restart.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,8 @@
+#!/bin/bash
+
+BASE=`dirname $0`
+
+${BASE}/down.sh
+sleep 1
+${BASE}/up.sh
+

=== added file 'dev-scripts/up.sh'
--- dev-scripts/up.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/up.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,10 @@
+#!/bin/bash
+
+BASE=`dirname $0`
+screen -X screen -t redis ${BASE}/redis.sh
+screen -X screen -t api ${BASE}/api.sh
+screen -X screen -t object ${BASE}/objectstore.sh
+screen -X screen -t compute ${BASE}/compute.sh
+screen -X screen -t volume ${BASE}/volume.sh
+screen -X screen -t network ${BASE}/network.sh
+

=== added file 'dev-scripts/upload-image.sh'
--- dev-scripts/upload-image.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/upload-image.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,13 @@
+#!/bin/bash
+
+cd ~/openstack/images
+cd ubuntu-kvm
+
+mkdir -p image 
+. ${NOVA_SRC_BASE}/me/novarc
+euca-bundle-image --image root.vmdk --destination ./image 
+euca-upload-bundle --bucket image --manifest ./image/root.vmdk.manifest.xml
+EMI=`euca-register image/root.img.manifest.xml | awk '{print $2}'`
+echo $EMI
+
+

=== added file 'dev-scripts/volume.sh'
--- dev-scripts/volume.sh	1970-01-01 00:00:00 +0000
+++ dev-scripts/volume.sh	2010-08-21 14:09:42 +0000
@@ -0,0 +1,11 @@
+#!/bin/bash
+
+. ~/openstack/conf
+cd ${NOVA_SRC_BASE}
+
+NOVA_VOLUME_ARGS="--storage_dev=${LOOPBACK_MOUNT}"
+
+#are you running from iSCSI? if not, comment out 2nd line and run first line:
+#bin/nova-volume ${NOVA_VOLUME_ARGS} --nodaemon --verbose start
+bin/nova-volume ${NOVA_VOLUME_ARGS} --nodaemon --iscsi_storage=true --verbose start
+

=== modified file 'nova/compute/disk.py'
--- nova/compute/disk.py	2010-08-21 14:09:42 +0000
+++ nova/compute/disk.py	2010-08-21 14:09:42 +0000
@@ -25,12 +25,9 @@
 import os
 import tempfile
 
-from twisted.internet import defer
-
 from nova import exception
 
 
[e-mail]
 def partition(infile, outfile, local_bytes=0, local_type='ext2', execute=None):
     """Takes a single partition represented by infile and writes a bootable
     drive image into outfile.
@@ -66,27 +63,26 @@
     last_sector = local_last # e
 
     # create an empty file
-    yield execute('dd if=/dev/zero of=%s count=1 seek=%d bs=%d'
+    execute('dd if=/dev/zero of=%s count=1 seek=%d bs=%d'
                   % (outfile, last_sector, sector_size))
 
     # make mbr partition
-    yield execute('parted --script %s mklabel msdos' % outfile)
+    execute('parted --script %s mklabel msdos' % outfile)
 
     # make primary partition
-    yield execute('parted --script %s mkpart primary %ds %ds'
+    execute('parted --script %s mkpart primary %ds %ds'
                   % (outfile, primary_first, primary_last))
 
     # make local partition
     if local_bytes > 0:
-        yield execute('parted --script %s mkpartfs primary %s %ds %ds'
+        execute('parted --script %s mkpartfs primary %s %ds %ds'
                       % (outfile, local_type, local_first, local_last))
 
     # copy file into partition
-    yield execute('dd if=%s of=%s bs=%d seek=%d conv=notrunc,fsync'
+    execute('dd if=%s of=%s bs=%d seek=%d conv=notrunc,fsync'
                   % (infile, outfile, sector_size, primary_first))
 
 
[e-mail]
 def inject_data(    image, key=None, net=None, dns=None, 
                     remove_network_udev=False, 
                     partition=None, execute=None):
@@ -98,14 +94,14 @@
     If partition is not specified it mounts the image as a single partition.
 
     """
-    out, err = yield execute('sudo losetup --find --show %s' % image)
+    out, err = execute('sudo losetup --find --show %s' % image)
     if err:
         raise exception.Error('Could not attach image to loopback: %s' % err)
     device = out.strip()
     try:
         if not partition is None:
             # create partition
-            out, err = yield execute('sudo kpartx -a %s' % device)
+            out, err = execute('sudo kpartx -a %s' % device)
             if err:
                 raise exception.Error('Failed to load partition: %s' % err)
             mapped_device = '/dev/mapper/%sp%s' % (device.split('/')[-1],
@@ -121,12 +117,12 @@
                 % mapped_device)
 
         # Configure ext2fs so that it doesn't auto-check every N boots
-        out, err = yield execute('sudo tune2fs -c 0 -i 0 %s' % mapped_device)
+        out, err = execute('sudo tune2fs -c 0 -i 0 %s' % mapped_device)
 
         tmpdir = tempfile.mkdtemp()
         try:
             # mount loopback to dir
-            out, err = yield execute(
+            out, err = execute(
                     'sudo mount %s %s' % (mapped_device, tmpdir))
             if err:
                 raise exception.Error('Failed to mount filesystem: %s' % err)
@@ -134,53 +130,48 @@
             try:
                 if key:
                     # inject key file
-                    yield _inject_key_into_fs(key, tmpdir, execute=execute)
+                    _inject_key_into_fs(key, tmpdir, execute=execute)
                 if net:
-                    yield _inject_net_into_fs(net, tmpdir, execute=execute)
+                    _inject_net_into_fs(net, tmpdir, execute=execute)
                 if dns:
-                    yield _inject_dns_into_fs(dns, tmpdir, execute=execute)
+                    _inject_dns_into_fs(dns, tmpdir, execute=execute)
                 if remove_network_udev:
-                    yield _remove_network_udev(tmpdir, execute=execute)
+                    _remove_network_udev(tmpdir, execute=execute)
 
             finally:
                 # unmount device
-                yield execute('sudo umount %s' % mapped_device)
+                execute('sudo umount %s' % mapped_device)
         finally:
             # remove temporary directory
-            yield execute('rmdir %s' % tmpdir)
+            execute('rmdir %s' % tmpdir)
             if not partition is None:
                 # remove partitions
-                yield execute('sudo kpartx -d %s' % device)
+                execute('sudo kpartx -d %s' % device)
     finally:
         # remove loopback
-        yield execute('sudo losetup --detach %s' % device)
-
-
[e-mail]
+        execute('sudo losetup --detach %s' % device)
+
 def _inject_key_into_fs(key, fs, execute=None):
     sshdir = os.path.join(fs, 'root', '.ssh')
-    yield execute('sudo mkdir -p %s' % sshdir) # existing dir doesn't matter
-    yield execute('sudo chown root %s' % sshdir)
-    yield execute('sudo chmod 700 %s' % sshdir)
+    execute('sudo mkdir -p %s' % sshdir) # existing dir doesn't matter
+    execute('sudo chown root %s' % sshdir)
+    execute('sudo chmod 700 %s' % sshdir)
     keyfile = os.path.join(sshdir, 'authorized_keys')
-    yield execute('sudo tee -a %s' % keyfile, '\n' + key.strip() + '\n')
-
-
[e-mail]
+    execute('sudo tee -a %s' % keyfile, '\n' + key.strip() + '\n')
+
+
 def _inject_net_into_fs(net, fs, execute=None):
     netfile = os.path.join(fs, 'etc', 'network', 'interfaces')
-    yield execute('sudo tee %s' % netfile, net)
+    execute('sudo tee %s' % netfile, net)
 
[e-mail]
 def _inject_dns_into_fs(dns, fs, execute=None):
     dnsfile = os.path.join(fs, 'etc', 'resolv.conf')
-    yield execute('sudo tee %s' % dnsfile, dns)
+    execute('sudo tee %s' % dnsfile, dns)
 
[e-mail]
 def _remove_network_udev(fs, execute=None):
     # TODO(justinsb): This is correct for Ubuntu, but might not be right for
     #  other distros.  There is a much bigger discussion to be had about what
     #  we inject and how we inject it.
     rulesfile = os.path.join(fs, 'etc', 'udev', 'rules.d', '70-persistent-net.rules')
-    yield execute('rm -f %s' % rulesfile)
+    execute('rm -f %s' % rulesfile)
 

=== modified file 'nova/compute/monitor.py'
--- nova/compute/monitor.py	2010-08-16 12:16:21 +0000
+++ nova/compute/monitor.py	2010-08-21 14:09:42 +0000
@@ -389,7 +389,7 @@
         return '%d:%d' % (rx, tx)
 
 
-class InstanceMonitor(object, service.Service):
+class InstanceMonitor(object, service.ThreadedService):
     """
     Monitors the running instances of the current machine.
     """

=== modified file 'nova/compute/service.py'
--- nova/compute/service.py	2010-08-16 12:16:21 +0000
+++ nova/compute/service.py	2010-08-21 14:09:42 +0000
@@ -30,12 +30,8 @@
 import os
 import sys
 
-from twisted.internet import defer
-from twisted.internet import task
-
 from nova import exception
 from nova import flags
-from nova import process
 from nova import service
 from nova import utils
 from nova.compute import disk
@@ -53,7 +49,7 @@
                         'where instances are stored on disk')
 
 
-class ComputeService(service.Service):
+class ComputeService(service.ThreadedService):
     """
     Manages the running instances.
     """
@@ -67,7 +63,7 @@
 
     def noop(self):
         """ simple test of an AMQP message call """
-        return defer.succeed('PONG')
+        return 'PONG'
 
     def get_instance(self, instance_id):
         # inst = self.instdir.get(instance_id)
@@ -79,7 +75,7 @@
     @exception.wrap_exception
     def adopt_instances(self):
         """ if there are instances already running, adopt them """
-        return defer.succeed(0)
+        return 0
         instance_names = self._conn.list_instances()
         for name in instance_names:
             try:
@@ -87,7 +83,7 @@
                 new_inst.update_state()
             except:
                 pass
-        return defer.succeed(len(self._instances))
+        return len(self._instances)
 
     @exception.wrap_exception
     def describe_instances(self):
@@ -97,7 +93,6 @@
                     Instance.fromName(self._conn, inst['instance_id']))
         return retval
 
-    @defer.inlineCallbacks
     def report_state(self, nodename, daemon):
         # TODO(termie): make this pattern be more elegant. -todd
         try:
@@ -111,7 +106,7 @@
             if not getattr(self, "model_disconnected", False):
                 self.model_disconnected = True
                 logging.exception("model server went away")
-        yield
+        
 
     @exception.wrap_exception
     def run_instance(self, instance_id, **_kwargs):
@@ -161,7 +156,6 @@
                     'trying to reboot unknown instance: %s' % instance_id)
         return instance.reboot()
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def get_console_output(self, instance_id):
         """ send the console output for an instance """
@@ -171,34 +165,31 @@
         if not instance:
             raise exception.Error(
                     'trying to get console log for unknown: %s' % instance_id)
-        rv = yield instance.console_output()
+        rv = instance.console_output()
         # TODO(termie): this stuff belongs in the API layer, no need to
         #               munge the data we send to ourselves
         output = {"InstanceId" : instance_id,
                   "Timestamp" : "2",
                   "output" : base64.b64encode(rv)}
-        defer.returnValue(output)
+        return output
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def attach_volume(self, instance_id = None,
                       volume_id = None, mountpoint = None):
         volume = volume_service.get_volume(volume_id)
-        yield self._init_aoe()
-        yield process.simple_execute(
+        self._init_aoe()
+        utils.execute(
                 "sudo virsh attach-disk %s /dev/etherd/%s %s" %
                 (instance_id,
                  volume['aoe_device'],
                  mountpoint.rpartition('/dev/')[2]))
         volume.finish_attach()
-        defer.returnValue(True)
+        return True
 
-    @defer.inlineCallbacks
     def _init_aoe(self):
-        yield process.simple_execute("sudo aoe-discover")
-        yield process.simple_execute("sudo aoe-stat")
+        utils.execute("sudo aoe-discover")
+        utils.execute("sudo aoe-stat")
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def detach_volume(self, instance_id, volume_id):
         """ detach a volume from an instance """
@@ -206,10 +197,10 @@
         # name without the leading /dev/
         volume = volume_service.get_volume(volume_id)
         target = volume['mountpoint'].rpartition('/dev/')[2]
-        yield process.simple_execute(
+        utils.execute(
                 "sudo virsh detach-disk %s %s " % (instance_id, target))
         volume.finish_detach()
-        defer.returnValue(True)
+        return True
 
 
 class Group(object):
@@ -317,7 +308,6 @@
         self.set_state(self.state)
         self.datamodel.save() # Extra, but harmless
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def destroy(self):
         if self.is_destroyed():
@@ -326,10 +316,9 @@
                                   ' instance: %s' % self.name)
 
         self.set_state(power_state.NOSTATE, 'shutting_down')
-        yield self._conn.destroy(self)
+        self._conn.destroy(self)
         self.datamodel.destroy()
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def reboot(self):
         if not self.is_running():
@@ -339,18 +328,17 @@
 
         logging.debug('rebooting instance %s' % self.name)
         self.set_state(power_state.NOSTATE, 'rebooting')
-        yield self._conn.reboot(self)
+        self._conn.reboot(self)
         self.update_state()
 
-    @defer.inlineCallbacks
     @exception.wrap_exception
     def spawn(self):
         self.set_state(power_state.NOSTATE, 'spawning')
         logging.debug("Starting spawn in Instance")
         try:
-            yield self._conn.spawn(self)
+            self._conn.spawn(self)
         except Exception, ex:
-            logging.debug(ex)
+            logging.warn(ex)
             self.set_state(power_state.SHUTDOWN)
         self.update_state()
 
@@ -364,4 +352,4 @@
                 console = f.read()
         else:
             console = 'FAKE CONSOLE OUTPUT'
-        return defer.succeed(console)
+        return console

=== modified file 'nova/network/service.py'
--- nova/network/service.py	2010-08-18 15:44:24 +0000
+++ nova/network/service.py	2010-08-21 14:09:42 +0000
@@ -81,7 +81,7 @@
     return "networkhost:%s" % project_id
 
 
-class BaseNetworkService(service.Service):
+class BaseNetworkService(service.TwistedService):
     """Implements common network service functionality
 
     This class must be subclassed.

=== modified file 'nova/process.py'
--- nova/process.py	2010-08-18 21:14:24 +0000
+++ nova/process.py	2010-08-21 14:09:42 +0000
@@ -29,28 +29,12 @@
 from twisted.internet import reactor
 
 from nova import flags
+from nova.utils import ProcessExecutionError
 
 FLAGS = flags.FLAGS
 flags.DEFINE_integer('process_pool_size', 4,
                      'Number of processes to use in the process pool')
 
-
-# NOTE(termie): this is copied from twisted.internet.utils but since
-#               they don't export it I've copied and modified
-class UnexpectedErrorOutput(IOError):
-    """
-    Standard error data was received where it was not expected.  This is a
-    subclass of L{IOError} to preserve backward compatibility with the previous
-    error behavior of L{getProcessOutput}.
-
-    @ivar processEnded: A L{Deferred} which will fire when the process which
-        produced the data on stderr has ended (exited and all file descriptors
-        closed).
-    """
-    def __init__(self, stdout=None, stderr=None):
-        IOError.__init__(self, "got stdout: %r\nstderr: %r" % (stdout, stderr))
-
-
 # This is based on _BackRelay from twister.internal.utils, but modified to 
 #  capture both stdout and stderr, without odd stderr handling, and also to 
 #  handle stdin
@@ -62,22 +46,23 @@
     @ivar deferred: A L{Deferred} which will be called back with all of stdout
         and all of stderr as well (as a tuple).  C{terminate_on_stderr} is true
         and any bytes are received over stderr, this will fire with an
-        L{_UnexpectedErrorOutput} instance and the attribute will be set to 
+        L{_ProcessExecutionError} instance and the attribute will be set to 
         C{None}.
 
     @ivar onProcessEnded: If C{terminate_on_stderr} is false and bytes are 
         received over stderr, this attribute will refer to a L{Deferred} which 
         will be called back when the process ends.  This C{Deferred} is also 
-        associated with the L{_UnexpectedErrorOutput} which C{deferred} fires 
+        associated with the L{_ProcessExecutionError} which C{deferred} fires 
         with earlier in this case so that users can determine when the process 
         has actually ended, in addition to knowing when bytes have been received
         via stderr.
     """
 
-    def __init__(self, deferred, started_deferred=None, 
+    def __init__(self, deferred, cmd, started_deferred=None, 
                     terminate_on_stderr=False, check_exit_code=True, 
                     process_input=None):
         self.deferred = deferred
+        self.cmd = cmd
         self.stdout = StringIO.StringIO()
         self.stderr = StringIO.StringIO()
         self.started_deferred = started_deferred
@@ -85,14 +70,18 @@
         self.check_exit_code = check_exit_code
         self.process_input = process_input
         self.on_process_ended = None
-    
+   
+    def _build_execution_error(self, exit_code=None):
+        return ProcessExecutionError(   cmd=self.cmd,
+                                        exit_code=exit_code,
+                                        stdout=self.stdout.getvalue(),
+                                        stderr=self.stderr.getvalue())
+
     def errReceived(self, text):
         self.stderr.write(text)
         if self.terminate_on_stderr and (self.deferred is not None):
             self.on_process_ended = defer.Deferred()
-            self.deferred.errback(UnexpectedErrorOutput(
-                                        stdout=self.stdout.getvalue(),
-                                        stderr=self.stderr.getvalue()))
+            self.deferred.errback(self._build_execution_error())
             self.deferred = None
             self.transport.loseConnection()
 
@@ -102,15 +91,19 @@
     def processEnded(self, reason):
         if self.deferred is not None:
             stdout, stderr = self.stdout.getvalue(), self.stderr.getvalue()
-            try:
-                if self.check_exit_code:
-                    reason.trap(error.ProcessDone)
-                self.deferred.callback((stdout, stderr))
-            except:
-                # NOTE(justinsb): This logic is a little suspicious to me...
-                # If the callback throws an exception, then errback will be 
-                # called also. However, this is what the unit tests test for...
-                self.deferred.errback(UnexpectedErrorOutput(stdout, stderr))
+            exit_code = reason.value.exitCode
+            if self.check_exit_code and exit_code <> 0:
+                self.deferred.errback(self._build_execution_error(exit_code))
+            else:
+                try:
+                    if self.check_exit_code:
+                        reason.trap(error.ProcessDone)
+                    self.deferred.callback((stdout, stderr))
+                except:
+                    # NOTE(justinsb): This logic is a little suspicious to me...
+                    # If the callback throws an exception, then errback will be 
+                    # called also. However, this is what the unit tests test for...
+                    self.deferred.errback(self._build_execution_error(exit_code))
         elif self.on_process_ended is not None:
             self.on_process_ended.errback(reason)
 
@@ -131,8 +124,12 @@
     args = args and args or ()
     env = env and env and {}
     deferred = defer.Deferred()
+    cmd = executable
+    if args:
+        cmd = cmd + " " + ' '.join(args)
     process_handler = BackRelayWithInput(
-            deferred, 
+            deferred,
+            cmd,
             started_deferred=started_deferred, 
             check_exit_code=check_exit_code, 
             process_input=process_input,

=== modified file 'nova/rpc.py'
--- nova/rpc.py	2010-08-18 15:44:24 +0000
+++ nova/rpc.py	2010-08-21 14:09:42 +0000
@@ -1,8 +1,12 @@
 # vim: tabstop=4 shiftwidth=4 softtabstop=4
+from nova.threads import PollingThread
+from threading import Thread
+import threading
 
 # Copyright 2010 United States Government as represented by the
 # Administrator of the National Aeronautics and Space Administration.
 # All Rights Reserved.
+# Copyright 2010 FathomDB 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
@@ -71,6 +75,14 @@
         del cls._instance
         return cls.instance()
 
+# Carrot is seriously FUBAR when it comes to threads
+#  (1990 called and wants its libraries back)
+# It looks like it we simultaneously issue fetches from two threads, carrot crashes.
+# Also, we can't open a queue connection on one thread and then call from another.
+# We'll probably have to introduce a message queueing library that works
+#  (or just wrap carrot)
+# Finding all the problems first, then going to decide...
+queue_lock = threading.Lock()
 
 class Consumer(messaging.Consumer):
     """Consumer base class
@@ -100,6 +112,7 @@
         """Wraps the parent fetch with some logic for failed connections"""
         # TODO(vish): the logic for failed connections and logging should be
         #             refactored into some sort of connection manager object
+        LOG.debug("Doing fetch from queue")
         try:
             if self.failed_connection:
                 # NOTE(vish): conn is defined in the parent class, we can
@@ -107,16 +120,17 @@
                 # pylint: disable-msg=W0201
                 self.conn = Connection.recreate()
                 self.backend = self.conn.create_backend()
-            super(Consumer, self).fetch(no_ack, auto_ack, enable_callbacks)
+            with queue_lock:
+                super(Consumer, self).fetch(no_ack, auto_ack, enable_callbacks)
             if self.failed_connection:
-                logging.error("Reconnected to queue")
+                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
         #             persistent failure occurs.
         except Exception:  # pylint: disable-msg=W0703
             if not self.failed_connection:
-                logging.exception("Failed to fetch message from queue")
+                LOG.exception("Failed to fetch message from queue")
                 self.failed_connection = True
 
     def attach_to_twisted(self):
@@ -124,7 +138,44 @@
         loop = task.LoopingCall(self.fetch, enable_callbacks=True)
         loop.start(interval=0.1)
 
-
+    def start_background_thread(self):
+        consumer_thread = ConsumeQueueThread()
+        consumer_thread.setName('Queue consumer thread')
+        consumer_thread.task_interval = 0.1
+        consumer_thread.consumer = self
+        consumer_thread.setDaemon(True)
+        consumer_thread.start()
+        return consumer_thread
+
+class ConsumeQueueThread(PollingThread):
+    """ Thread that waits for queue messages """
+    # It would be very easy to convert this to be non-polling
+    #  (i.e. just call consumer.wait(), but we're cloning existing behaviour before improving it
+    def _do_task(self):
+        self._fetch_from_queue()
+        
+    def _fetch_from_queue(self):
+        self.consumer.fetch(enable_callbacks=True)
+
+class ProcessMessageThread(Thread):
+    """ Thread that processes a queue message """
+    def run(self):
+        LOG.warning("Invoking %s %s", self.handler_function, self.handler_args)
+        
+        try:
+            rval = self.handler_function(**self.handler_args)
+        except Exception as e:
+            LOG.exception("Error invoking message handler")
+            if self.msg_id:
+                msg_reply(self.msg_id, str(e))
+            else:
+                LOG.warning("No msg_id, cannot send error message")
+                
+        LOG.warning("retval = %s", rval)
+        
+        if self.msg_id:
+            msg_reply(self.msg_id, rval)
+   
 class Publisher(messaging.Publisher):
     """Publisher base class"""
     pass
@@ -144,9 +195,11 @@
 
 class AdapterConsumer(TopicConsumer):
     """Calls methods on a proxy object based on method and args"""
-    def __init__(self, connection=None, topic="broadcast", proxy=None):
+    def __init__(self, connection=None, topic="broadcast", proxy=None, 
+                    twisted=True):
         LOG.debug('Initing the Adapter Consumer for %s' % (topic))
         self.proxy = proxy
+        self.twisted = twisted
         super(AdapterConsumer, self).__init__(connection=connection,
                                               topic=topic)
 
@@ -177,12 +230,21 @@
 
         node_func = getattr(self.proxy, str(method))
         node_args = dict((str(k), v) for k, v in args.iteritems())
-        # NOTE(vish): magic is fun!
-        # pylint: disable-msg=W0142
-        d = defer.maybeDeferred(node_func, **node_args)
-        if msg_id:
-            d.addCallback(lambda rval: msg_reply(msg_id, rval, None))
-            d.addErrback(lambda e: msg_reply(msg_id, None, e))
+        
+        if self.twisted:
+            # NOTE(vish): magic is fun!
+            # pylint: disable-msg=W0142
+            d = defer.maybeDeferred(node_func, **node_args)
+            if msg_id:
+                d.addCallback(lambda rval: msg_reply(msg_id, rval, None))
+                d.addErrback(lambda e: msg_reply(msg_id, None, e))
+        else:
+            handler_thread = ProcessMessageThread()
+            handler_thread.setName('Message handler: %s' % str(method))
+            handler_thread.handler_function = node_func
+            handler_thread.handler_args = node_args
+            handler_thread.msg_id = msg_id
+            handler_thread.start()
         return
 
 
@@ -227,14 +289,15 @@
     if failure:
         message = failure.getErrorMessage()
         traceback = failure.getTraceback()
-        logging.error("Returning exception %s to caller", message)
-        logging.error(traceback)
+        LOG.error("Returning exception %s to caller", message)
+        LOG.error(traceback)
         failure = (failure.type.__name__, str(failure.value), traceback)
     conn = Connection.instance()
     publisher = DirectPublisher(connection=conn, msg_id=msg_id)
     try:
         publisher.send({'result': reply, 'failure': failure})
     except TypeError:
+        LOG.exception("Error sending reply")
         publisher.send(
                 {'result': dict((k, repr(v))
                                 for k, v in reply.__dict__.iteritems()),
@@ -338,3 +401,40 @@
     #             topic and a json sting representing a dictionary
     #             for the method
     send_message(sys.argv[1], json.loads(sys.argv[2]))
+
+
+
+
+class ConnectionManager(object):
+    def __init__(self):
+        self.connection_map_lock = threading.Lock()
+        self.connection_map = {}
+
+    def _build_key(self, config):
+        key = tuple(sorted(config.items()))
+        return key
+
+    def _build_connection(self, config):
+        LOG.debug('Building queue connection using %s', config)
+        c = Connection(**config)
+        c.queue_config = config
+        c.connection_manager = self
+        return c
+
+    def get_connection(self, config):
+        key = self._build_key(config)
+
+        with self.connection_map_lock:
+            c = self.connection_map.get(key)
+            if not c:
+                c = self._build_connection(config)
+                self.connection_map[key] = c
+            return c
+
+    def invalidate_connection(self, connection):
+        key = self._build_key(connection.queue_config)
+
+        with self.connection_map_lock:
+            c = self.connection_map.get(key)
+            if c == connection:
+                del self.connection_map[key]

=== modified file 'nova/server.py'
--- nova/server.py	2010-08-09 14:46:33 +0000
+++ nova/server.py	2010-08-21 14:09:42 +0000
@@ -22,7 +22,6 @@
 
 import daemon
 from daemon import pidlockfile
-import logging
 import logging.handlers
 import os
 import signal
@@ -65,6 +64,7 @@
             time.sleep(0.1)
     except OSError, err:
         err = str(err)
+        # Should this be >= 0 ?
         if err.find("No such process") > 0:
             if os.path.exists(pidfile):
                 os.remove(pidfile)

=== modified file 'nova/service.py'
--- nova/service.py	2010-07-23 22:27:18 +0000
+++ nova/service.py	2010-08-21 14:09:42 +0000
@@ -1,4 +1,5 @@
 # vim: tabstop=4 shiftwidth=4 softtabstop=4
+from nova.threads import PollingThread
 
 # Copyright 2010 United States Government as represented by the
 # Administrator of the National Aeronautics and Space Administration.
@@ -23,10 +24,13 @@
 import inspect
 import logging
 import os
+import signal
+import time
 
 from twisted.internet import defer
 from twisted.internet import task
 from twisted.application import service
+from threading import Thread
 
 from nova import datastore
 from nova import flags
@@ -40,64 +44,117 @@
                      'seconds between nodes reporting state to cloud',
                      lower_bound=1)
 
-class Service(object, service.Service):
+def handle_sig_int(signum, frame):
+    logging.info('Received SIGINT')
+    ThreadedService.keep_running = False
+
+class ThreadedService(object):
     """Base class for workers that run on hosts"""
-
+    keep_running = True
+    
     @classmethod
     def create(cls,
                report_interval=None, # defaults to flag
                bin_name=None, # defaults to basename of executable
                topic=None): # defaults to basename - "nova-" part
+        return _create_common(cls, report_interval=report_interval, 
+                                bin_name=bin_name, topic=topic, twisted=False)
+
+    @classmethod
+    def main_loop(cls):
+        signal.signal(signal.SIGINT, handle_sig_int)
+        while ThreadedService.keep_running:
+            time.sleep(1)
+
+def _create_common(cls,
+                   report_interval=None, # defaults to flag
+                   bin_name=None, # defaults to basename of executable
+                   topic=None, # defaults to basename - "nova-" part
+                   twisted=False):
+    if True: # Hack to minimize diff
         """Instantiates class and passes back application object"""
-        if not report_interval:
-            # NOTE(vish): set here because if it is set to flag in the
-            #             parameter list, it wrongly uses the default
-            report_interval = FLAGS.report_interval
         # NOTE(vish): magic to automatically determine bin_name and topic
         if not bin_name:
             bin_name = os.path.basename(inspect.stack()[-1][1])
         if not topic:
             topic = bin_name.rpartition("nova-")[2]
-        logging.warn("Starting %s node" % topic)
+        logging.warn("Starting %s node", topic)
         node_instance = cls()
 
         conn = rpc.Connection.instance()
         consumer_all = rpc.AdapterConsumer(
                 connection=conn,
                 topic='%s' % topic,
-                proxy=node_instance)
+                proxy=node_instance,
+                twisted=twisted)
 
         consumer_node = rpc.AdapterConsumer(
                 connection=conn,
                 topic='%s.%s' % (topic, FLAGS.node_name),
-                proxy=node_instance)
-
-        pulse = task.LoopingCall(node_instance.report_state,
-                                 FLAGS.node_name,
-                                 bin_name)
-        pulse.start(interval=report_interval, now=False)
-
-        consumer_all.attach_to_twisted()
-        consumer_node.attach_to_twisted()
-
-        # This is the parent service that twistd will be looking for when it
-        # parses this file, return it so that we can get it into globals below
-        application = service.Application(bin_name)
-        node_instance.setServiceParent(application)
-        return application
+                proxy=node_instance,
+                twisted=twisted)
+
+        if not report_interval:
+            # NOTE(vish): set here because if it is set to flag in the
+            #             parameter list, it wrongly uses the default
+            report_interval = FLAGS.report_interval
+        if twisted:
+            pulse = task.LoopingCall(node_instance._report_state_twisted,
+                                     FLAGS.node_name,
+                                     bin_name)
+            pulse.start(interval=report_interval, now=False)
+        
+            consumer_all.attach_to_twisted()
+            consumer_node.attach_to_twisted()
+        else:
+            report_state_thread = ReportStateThread()
+            report_state_thread.setName('Report state thread')
+            report_state_thread.task_interval = report_interval
+            report_state_thread.nodename = FLAGS.node_name
+            report_state_thread.daemon = bin_name
+            report_state_thread.setDaemon(True)
+            report_state_thread.start()
+
+            consumer_all.start_background_thread()
+            consumer_node.start_background_thread()
+       
+        if twisted:
+            # This is the parent service that twistd will be looking for when it
+            # parses this file, return it so that we can get it into globals below
+            application = service.Application(bin_name)
+            node_instance.setServiceParent(application)
+            return application
+
+class TwistedService(object, service.Service):
+    @classmethod
+    def create(cls,
+               report_interval=None, # defaults to flag
+               bin_name=None, # defaults to basename of executable
+               topic=None): # defaults to basename - "nova-" part
+        return _create_common(cls, report_interval=report_interval,
+                              bin_name=bin_name, topic=topic, twisted=True)
+                                                              
 
     @defer.inlineCallbacks
-    def report_state(self, nodename, daemon):
-        # TODO(termie): make this pattern be more elegant. -todd
-        try:
-            record = model.Daemon(nodename, daemon)
-            record.heartbeat()
-            if getattr(self, "model_disconnected", False):
-                self.model_disconnected = False
-                logging.error("Recovered model server connection!")
-
-        except datastore.ConnectionError, ex:
-            if not getattr(self, "model_disconnected", False):
-                self.model_disconnected = True
-                logging.exception("model server went away")
+    def _report_state_twisted(self, nodename, daemon):
+        _report_state_common(self, nodename, daemon)
         yield
+
+def _report_state_common(target, nodename, daemon):
+    # TODO(termie): make this pattern be more elegant. -todd
+    try:
+        record = model.Daemon(nodename, daemon)
+        record.heartbeat()
+        if getattr(target, "model_disconnected", False):
+            target.model_disconnected = False
+            logging.error("Recovered model server connection!")
+    except datastore.ConnectionError, ex:
+        if not getattr(target, "model_disconnected", False):
+            target.model_disconnected = True
+            logging.exception("model server went away")
+
+class ReportStateThread(PollingThread):
+    """ Thread that reports state back periodically (heartbeat) """
+    def _do_task(self):
+        _report_state_common(self, self.nodename, self.daemon)
+   

=== added file 'nova/threads.py'
--- nova/threads.py	1970-01-01 00:00:00 +0000
+++ nova/threads.py	2010-08-21 14:09:42 +0000
@@ -0,0 +1,40 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 FathomDB 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.
+
+from threading import Thread
+import logging
+import time
+
+"""
+Helper functions for dealing with (simple python) threads
+"""
+
+class PollingThread(Thread):
+    """ Thread that does something periodically """
+    def run(self):
+        while True:
+            try:
+                self._run()
+            except Exception as e:
+                logging.exception("Error on thread %s", self.name)
+    
+    def _run(self):
+        time.sleep(self.task_interval)
+        self._do_task()
+
+    def _do_task(self):
+        raise Exception("Must implement _do_task")

=== modified file 'nova/twistd.py'
--- nova/twistd.py	2010-08-12 18:52:32 +0000
+++ nova/twistd.py	2010-08-21 14:09:42 +0000
@@ -193,6 +193,7 @@
             time.sleep(0.1)
     except OSError, err:
         err = str(err)
+        # Should this be >= 0 ?
         if err.find("No such process") > 0:
             if os.path.exists(pidfile):
                 os.remove(pidfile)
@@ -201,7 +202,7 @@
             sys.exit(1)
 
 
-def serve(filename):
+def serve(filename, twisted=True):
     logging.debug("Serving %s" % filename)
     name = os.path.basename(filename)
     OptionsClass = WrapTwistedOptions(TwistdServerOptions)
@@ -256,4 +257,5 @@
     for flag in FLAGS:
         logging.debug("%s : %s" % (flag, FLAGS.get(flag, None)))
 
-    twistd.runApp(options)
+    if twisted:
+        twistd.runApp(options)

=== modified file 'nova/utils.py'
--- nova/utils.py	2010-08-18 21:14:24 +0000
+++ nova/utils.py	2010-08-21 14:09:42 +0000
@@ -36,6 +36,16 @@
 FLAGS = flags.FLAGS
 TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
 
+class ProcessExecutionError(IOError):
+    def __init__(   self, stdout=None, stderr=None, exit_code=None, cmd=None,
+                    description=None):
+        if description is None:
+            description = "Unexpected error while running command."
+        if exit_code is None:
+            exit_code = '-'
+        message = "%s\nCommand: %s\nExit code: %s\nStdout: %r\nStderr: %r" % (
+                  description, cmd, exit_code, stdout, stderr)
+        IOError.__init__(self, message)
 
 def import_class(import_str):
     """Returns a class from a string including module and class"""
@@ -59,6 +69,7 @@
     execute("curl --fail %s -o %s" % (url, target))
 
 def execute(cmd, process_input=None, addl_env=None, check_exit_code=True):
+    logging.debug("Running command %s" % cmd)
     env = os.environ.copy()
     if addl_env:
         env.update(addl_env)
@@ -73,8 +84,11 @@
     if obj.returncode:
         logging.debug("Result was %s" % (obj.returncode))
         if check_exit_code and obj.returncode <> 0:
-            raise Exception(    "Unexpected exit code: %s.  result=%s" 
-                                % (obj.returncode, result))
+            (stdout, stderr) = result
+            raise ProcessExecutionError(exit_code=obj.returncode,
+                                        stdout=stdout,
+                                        stderr=stderr,
+                                        cmd=cmd)
     return result
 
 

=== modified file 'nova/virt/connection.py'
--- nova/virt/connection.py	2010-08-13 10:51:33 +0000
+++ nova/virt/connection.py	2010-08-21 14:09:42 +0000
@@ -17,11 +17,14 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import logging
+import sys
+
 from nova import flags
 from nova.virt import fake
 from nova.virt import libvirt_conn
 from nova.virt import xenapi
-
+from nova.virt import virtualbox
 
 FLAGS = flags.FLAGS
 
@@ -39,16 +42,21 @@
     # TODO(termie): maybe lazy load after initial check for permissions
     # TODO(termie): check whether we can be disconnected
     t = FLAGS.connection_type
+    print "connection_type %s" % t
     if t == 'fake':
         conn = fake.get_connection(read_only)
     elif t == 'libvirt':
         conn = libvirt_conn.get_connection(read_only)
     elif t == 'xenapi':
         conn = xenapi.get_connection(read_only)
+    elif t == 'virtualbox':
+        conn = virtualbox.get_connection(read_only)
     else:
-        raise Exception('Unknown connection type "%s"' % t)
+        logging.error('Unknown connection type "%s"', t)
+        sys.exit(1)
 
     if conn is None:
         logging.error('Failed to open connection to the hypervisor')
         sys.exit(1)
+        
     return conn

=== modified file 'nova/virt/images.py'
--- nova/virt/images.py	2010-08-18 21:14:24 +0000
+++ nova/virt/images.py	2010-08-21 14:09:42 +0000
@@ -29,6 +29,7 @@
 from nova import process
 from nova.auth import manager
 from nova.auth import signer
+from nova.auth import manager
 
 
 FLAGS = flags.FLAGS
@@ -43,7 +44,6 @@
         f = _fetch_local_image
     return f(image, path, user, project)
 
-
 def _fetch_s3_image(image, path, user, project):
     url = image_url(image)
 
@@ -53,11 +53,11 @@
     headers = {}
     headers['Date'] = time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime())
 
-    (_, _, url_path, _, _, _) = urlparse.urlparse(url)
+    uri = '/' + url.partition('/')[2]
     access = manager.AuthManager().get_access_key(user, project)
     signature = signer.Signer(user.secret.encode()).s3_authorization(headers,
                                                                      'GET',
-                                                                     url_path)
+                                                                     uri)
     headers['Authorization'] = 'AWS %s:%s' % (access, signature)
 
     cmd = ['/usr/bin/curl', '--fail', '--silent', url]
@@ -67,8 +67,7 @@
     cmd += ['-o', path]
     return process.SharedPool().execute(executable=cmd[0], args=cmd[1:])
 
-
-def _fetch_local_image(image, path, user, project):
+def _fetch_local_image(image, path, _user, _project):
     source = _image_path('%s/image' % image)
     return process.simple_execute('cp %s %s' % (source, path))
 

=== modified file 'nova/virt/libvirt_conn.py'
--- nova/virt/libvirt_conn.py	2010-08-21 14:09:42 +0000
+++ nova/virt/libvirt_conn.py	2010-08-21 14:09:42 +0000
@@ -244,6 +244,8 @@
         
         key = data['key_data']
         net = None
+        dns = None
+
         if data.get('inject_network', False):
             with open(FLAGS.injected_network_template) as f:
                 net = f.read() % {'address': data['private_dns_name'],
@@ -281,6 +283,7 @@
 
         # TODO(termie): lazy lazy hack because xml is annoying
         xml_info['nova'] = json.dumps(instance.datamodel.copy())
+        xml_info['type'] = FLAGS.libvirt_type
 
         if xml_info['kernel_id']:
             xml_info['kernel'] = xml_info['basepath'] + "/kernel"
@@ -311,6 +314,12 @@
                 'cpu_time': cpu_time}
 
     def get_disks(self, instance_id):
+        """
+        Note that this function takes an instance ID, not an Instance, so
+        that it can be called by monitor.
+
+        Returns a list of all block devices for this domain.
+        """
         domain = self._conn.lookupByName(instance_id)
         # TODO(devcamcar): Replace libxml2 with etree.
         xml = domain.XMLDesc(0)
@@ -347,6 +356,12 @@
         return disks
 
     def get_interfaces(self, instance_id):
+        """
+        Note that this function takes an instance ID, not an Instance, so
+        that it can be called by monitor.
+
+        Returns a list of all network interfaces for this instance.
+        """
         domain = self._conn.lookupByName(instance_id)
         # TODO(devcamcar): Replace libxml2 with etree.
         xml = domain.XMLDesc(0)
@@ -383,9 +398,17 @@
         return interfaces
 
     def block_stats(self, instance_id, disk):
+        """
+        Note that this function takes an instance ID, not an Instance, so
+        that it can be called by monitor.
+        """
         domain = self._conn.lookupByName(instance_id)
         return domain.blockStats(disk)
 
     def interface_stats(self, instance_id, interface):
+        """
+        Note that this function takes an instance ID, not an Instance, so
+        that it can be called by monitor.
+        """
         domain = self._conn.lookupByName(instance_id)
         return domain.interfaceStats(interface)

=== added file 'nova/virt/virtualbox.py'
--- nova/virt/virtualbox.py	1970-01-01 00:00:00 +0000
+++ nova/virt/virtualbox.py	2010-08-21 14:09:42 +0000
@@ -0,0 +1,332 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+import uuid
+import time
+
+# Copyright 2010 FathomDB 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.
+
+"""
+A connection to VirtualBox; this actually uses the VBoxManage command line tool
+because libvirt is problematic with Virtualbox and on a Mac
+"""
+
+import logging
+import os.path
+import shutil
+import time
+
+from nova import exception
+from nova import flags
+from nova import utils
+from nova.auth import manager
+from nova.compute import disk
+from nova.compute import instance_types
+from nova.compute import power_state
+from nova.virt import images
+
+from Cheetah.Template import Template
+
+FLAGS = flags.FLAGS
+flags.DEFINE_string('virtualbox_bridge_adapter',
+                    'eth0',
+                    'Network adapter to use for VirtualBox bridged networking')
+
+def get_connection(read_only):
+    return VirtualBoxConnection(read_only)
+
+
+class VirtualBoxConnection(object):
+    def __init__(self, read_only):
+        pass
+
+    def list_instances(self):
+        out, err = utils.execute('VBoxManage --nologo list vms')
+        if err:
+            raise exception.Error('Could not list virtualbox vms: %s' % err)
+
+        instance_ids = []
+        for line in out.splitlines():
+            tokens = line.split()
+            if len(tokens) == 2:
+                instance_id = tokens[0].strip('"')
+                instance_ids.append(instance_id)
+
+        return instance_ids
+
+    def destroy(self, instance):
+        _, err = utils.execute('VBoxManage --nologo controlvm %s poweroff' % instance.name)
+        if err:
+            # TODO(justinsb): If the instance was already stopped, we should accept that and continue...
+            raise exception.Error('Could not stop virtualbox vm: %s' % err)
+        
+        while True:
+            try:
+                instance.update_state()
+                if instance.state == power_state.SHUTOFF:
+                    return
+            except Exception:
+                logging.exception("Error waiting for machine to shut down")
+                instance.set_state(power_state.SHUTOFF)
+                return
+            time.sleep(0.5)
+
+    def _cleanup(self, instance):
+        target = os.path.abspath(instance.datamodel['basepath'])
+        logging.info("Deleting instance files at %s", target)
+        shutil.rmtree(target)
+
+    @exception.wrap_exception
+    def reboot(self, instance):
+        _, err = utils.execute('VBoxManage --nologo controlvm %s reset' % instance.name)
+        if err:
+            raise exception.Error('Could not reset virtualbox vm: %s' % err)
+        # TODO(justinsb): Should we wait for restart?
+
+    def _register_image(self, instance):
+        # See e.g. http://www.uhleeka.com/blog/2009/12/virtualbox-3-1-install-windows-xp-guest-on-ubuntu-9-04-host/
+        instance_name = instance.name
+        base_folder = self.basepath(instance)
+        image_path = self._image_path(instance)
+        image_uuid = uuid.uuid4()
+        memory_size = 1024
+        net_bridge_adapter = str(FLAGS.virtualbox_bridge_adapter)
+        
+        _, err = utils.execute('VBoxManage --nologo createvm --name %s --register --basefolder %s' % (instance_name, base_folder))
+        if err:
+            raise exception.Error('Could not create virtualbox vm: %s' % err)
+
+        _, err = utils.execute('VBoxManage --nologo storagectl %s --name "SATA Controller" --add sata --controller IntelAhci' % (instance_name, ))
+        if err:
+            raise exception.Error('Could not create SATA controller for virtualbox vm: %s' % err)
+
+        # VirtualBox seems to pull a UUID from the disk.  
+        # Thus we have to specify the UUID, or else VirtualBox refuses to
+        # register duplicate images because they have duplicate UUIDs
+        _, err = utils.execute('VBoxManage --nologo openmedium disk %s --uuid=%s' % (image_path, image_uuid))
+        if err:
+            raise exception.Error('Could not register disk with virtualbox vm: %s' % err)
+
+        _, err = utils.execute('VBoxManage --nologo storageattach %s --storagectl "SATA Controller" --port 0 --device 0 --type hdd --medium %s' % (instance_name, image_uuid))
+        if err:
+            raise exception.Error('Could not attach disk to virtualbox vm: %s' % err)
+
+        # TODO(justinsb): Do we need --sata on --sataportcount 30 ??
+        cmd = 'VBoxManage --nologo modifyvm %s --memory %s --vram 64 --nic1 bridged --bridgeadapter1 %s' % (instance_name, memory_size, net_bridge_adapter)
+        _, err = utils.execute(cmd)
+        if err:
+            raise exception.Error('Could not configure virtualbox vm: %s' % err)
+
+    def _start_image(self, instance):
+        instance_name = instance.name
+        _, err = utils.execute('VBoxManage --nologo startvm %s' % instance_name)
+        if err:
+            raise exception.Error('Could not start virtualbox instance %s: %s' % (instance_name, err))
+
+    @exception.wrap_exception
+    def spawn(self, instance):
+        instance.set_state(power_state.NOSTATE, 'launching')
+
+        self._create_image(instance)
+
+        self._register_image(instance)
+
+        self._start_image(instance)
+
+        # TODO(termie): this should actually register
+        # a callback to check for successful boot
+        logging.debug("Instance is running")
+
+        while True:
+            try:
+                instance.update_state()
+                if instance.is_running():
+                    logging.debug('booted instance %s', instance.name)
+                    return
+            except Exception as exn:
+                logging.error("_wait_for_boot exception %s", exn)
+                instance.set_state(power_state.SHUTDOWN)
+                logging.error('Failed to boot instance %s', instance.name)
+                return
+            time.sleep(0.5)
+
+
+    def _create_image(self, instance):
+        # syntactic nicety
+        data = instance.datamodel
+        basepath = lambda x = '': self.basepath(instance, x)
+
+        # ensure directories exist and are writable
+        utils.execute('mkdir -p %s' % basepath())
+        utils.execute('chmod 0777 %s' % basepath())
+
+
+        user = manager.AuthManager().get_user(data['user_id'])
+        project = manager.AuthManager().get_project(data['project_id'])
+        
+        if not os.path.exists(basepath('disk')):
+            images.fetch(data['image_id'], basepath('disk-raw'), user, project)
+
+        using_kernel = data['kernel_id'] and True
+
+        if using_kernel:
+            if not os.path.exists(basepath('kernel')):
+                images.fetch(data['kernel_id'], basepath('kernel'), user, project)
+            if not os.path.exists(basepath('ramdisk')):
+                images.fetch(data['ramdisk_id'], basepath('ramdisk'), user, project)
+
+        execute = lambda cmd, process_input = None: \
+                  utils.execute(    cmd=cmd,
+                                    process_input=process_input,
+                                    check_exit_code=True)
+
+        # For now, we assume that if we're not using a kernel, we're using a partitioned disk image
+        # where the target partition is the first partition
+        target_partition = None
+        if not using_kernel:
+            target_partition = "1"
+
+        key = data['key_data']
+        net = None
+        dns = None
+
+        if data.get('inject_network', False):
+            network_info = {    'address': data['private_dns_name'],
+                                'network': data['network_network'],
+                                'netmask': data['network_netmask'],
+                                'gateway': data['network_gateway'],
+                                'broadcast': data['network_broadcast'],
+                                'dns': data['network_dns']}
+
+            with open(FLAGS.injected_network_template) as f:
+                net = f.read() % network_info
+
+            #with open(FLAGS.simple_network_dns_template) as f:
+            #    dns = str(Template(f.read(), searchList=[ network_info ]))
+
+
+        if key or net or dns:
+            logging.info('Injecting data into image %s', data['image_id'])
+            try:
+                disk.inject_data(basepath('disk-raw'), key=key, net=net, dns=dns, remove_network_udev=True, partition=target_partition, execute=execute)
+            except Exception as e:
+                # This could be a windows image, or a vmdk format disk
+                logging.warn('Could not inject data; ignoring.  (%s)', e)
+
+        if using_kernel:
+            if os.path.exists(basepath('disk')):
+                utils.execute('rm -f %s' % basepath('disk'))
+
+            disk_bytes = (instance_types.INSTANCE_TYPES[data['instance_type']]['local_gb']
+                 * 1024 * 1024 * 1024)
+            disk.partition(
+                basepath('disk-raw'), basepath('disk'), disk_bytes, execute=execute)
+
+
+    def _uses_kernel(self, instance):
+        if instance.datamodel['ramdisk_id'] or instance.datamodel['kernel_id']:
+            return True
+        return False
+
+    def basepath(self, instance, path=''):
+        return os.path.abspath(os.path.join(instance.datamodel['basepath'], path))
+
+    def _image_path(self, instance):
+        if self._uses_kernel(instance):
+            return instance.datamodel['basepath'] + '/disk'
+        else:
+            return instance.datamodel['basepath'] + '/disk-raw'
+
+    def _get_virtualbox_info(self, instance_id):
+        out, err = utils.execute('VBoxManage --nologo showvminfo %s --machinereadable' % instance_id, check_exit_code=False)
+        if out.find("VBOX_E_OBJECT_NOT_FOUND") != -1:
+            return {}
+
+        if err:
+            raise exception.Error('Could not get virtualbox info: %s' % err)
+        # The output is a series of lines, each containing a key="value" pair 
+        info_map = {}
+        for line in out.splitlines():
+            line = line.strip()
+            if line == "":
+                continue
+            key, sep, value = line.partition('=')
+            if sep != '=':
+                if line.find('E_ACCESSDENIED') != -1:
+                    if 'poweroff' == info_map.get('VMState', None):
+                        # When the machine is being destroyed, we get some status and then E_ACCESSDENED; return what we've got
+                        break
+                logging.info("Could not parse virtualbox line: %s", line)
+                raise exception.Error('Could not parse virtualbox info: %s' % out)
+            info_map[key] = value.strip('"')
+
+        logging.debug("Virtual box info %s", info_map)
+
+        return info_map
+
+    def get_info(self, instance_id):
+        vb_info = self._get_virtualbox_info(instance_id)
+
+        state = self.instance_state_from_virtualbox_state[vb_info.get('VMState', None)]
+        max_mem = vb_info.get('memory', 0)
+        mem = max_mem
+        num_cpu = vb_info.get('cpus', 0)
+        cpu_time = 0
+
+        return {'state': state, 'max_mem': max_mem, 'mem': mem, 'num_cpu': num_cpu, 'cpu_time': cpu_time}
+
+    def get_disks(self, instance_id):
+        """
+        Note that this function takes an instance ID, not an Instance, so
+        that it can be called by monitor.
+        
+        Returns a list of all block devices for this domain.
+        """
+        raise Exception('Not yet implemented')
+
+
+    def get_interfaces(self, instance_id):
+        """
+        Note that this function takes an instance ID, not an Instance, so
+        that it can be called by monitor.
+        
+        Returns a list of all network interfaces for this instance.
+        """
+        raise Exception('Not yet implemented')
+
+
+    def block_stats(self, instance_id, disk):
+        """
+        Note that this function takes an instance ID, not an Instance, so
+        that it can be called by monitor.
+        """
+        raise Exception('Not yet implemented')
+
+
+    def interface_stats(self, instance_id, interface):
+        """
+        Note that this function takes an instance ID, not an Instance, so
+        that it can be called by monitor.
+        """
+        raise Exception('Not yet implemented')
+
+    instance_state_from_virtualbox_state = { 
+        None : power_state.NOSTATE,
+        'poweroff': power_state.SHUTOFF,
+        'starting': power_state.RUNNING,
+        'running' : power_state.RUNNING,
+        #'Running'  : power_state.RUNNING,
+        #'Paused'   : power_state.PAUSED,
+        #'Crashed'  : power_state.CRASHED
+    }

=== modified file 'nova/volume/service.py'
--- nova/volume/service.py	2010-08-18 21:19:39 +0000
+++ nova/volume/service.py	2010-08-21 14:09:42 +0000
@@ -77,7 +77,7 @@
     raise exception.Error("Volume does not exist")
 
 
-class VolumeService(service.Service):
+class VolumeService(service.TwistedService):
     """
     There is one VolumeNode running on each host.
     However, each VolumeNode can report on the state of