openstack/nova
Code Issues Proposed changes

Implements the blueprint for enabling the setting of the root/admin password on an instance.

Browse Source 
It uses a new xenapi plugin 'agent' that handles communication to/from the agent running on the guest.
This commit is contained in:
Ed Leafe
2011-01-14 07:49:41 +00:00
committed by Tarmac
parent 8a4eb03ec3 01a1ad3d2c
commit 9fb1e7b1f6
10 changed files with 431 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
nova/api/openstack/servers.py
View File

@@ -165,15 +165,18 @@ class Controller(wsgi.Controller):
if not inst_dict:
return faults.Fault(exc.HTTPUnprocessableEntity())
ctxt = req.environ['nova.context']
update_dict = {}
if 'adminPass' in inst_dict['server']:
update_dict['admin_pass'] = inst_dict['server']['adminPass']
try:
self.compute_api.set_admin_password(ctxt, id)
except exception.TimeoutException, e:
return exc.HTTPRequestTimeout()
if 'name' in inst_dict['server']:
update_dict['display_name'] = inst_dict['server']['name']
try:
self.compute_api.update(req.environ['nova.context'], id,
**update_dict)
self.compute_api.update(ctxt, id, **update_dict)
except exception.NotFound:
return faults.Fault(exc.HTTPNotFound())
return exc.HTTPNoContent()

115
nova/compute/api.py
View File

@@ -280,7 +280,7 @@ class API(base.Base):
return self.db.instance_update(context, instance_id, kwargs)
def delete(self, context, instance_id):
LOG.debug(_("Going to try and terminate %s"), instance_id)
LOG.debug(_("Going to try to terminate %s"), instance_id)
try:
instance = self.get(context, instance_id)
except exception.NotFound as e:
@@ -301,10 +301,8 @@ class API(base.Base):
host = instance['host']
if host:
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "terminate_instance",
"args": {"instance_id": instance_id}})
self._cast_compute_message('terminate_instance', context,
instance_id, host)
else:
self.db.instance_destroy(context, instance_id)
@@ -332,50 +330,34 @@ class API(base.Base):
project_id)
return self.db.instance_get_all(context)
def _cast_compute_message(self, method, context, instance_id, host=None):
"""Generic handler for RPC calls to compute."""
if not host:
instance = self.get(context, instance_id)
host = instance['host']
queue = self.db.queue_get_for(context, FLAGS.compute_topic, host)
kwargs = {'method': method, 'args': {'instance_id': instance_id}}
rpc.cast(context, queue, kwargs)
def snapshot(self, context, instance_id, name):
"""Snapshot the given instance."""
instance = self.get(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "snapshot_instance",
"args": {"instance_id": instance_id, "name": name}})
self._cast_compute_message('snapshot_instance', context, instance_id)
def reboot(self, context, instance_id):
"""Reboot the given instance."""
instance = self.get(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "reboot_instance",
"args": {"instance_id": instance_id}})
self._cast_compute_message('reboot_instance', context, instance_id)
def pause(self, context, instance_id):
"""Pause the given instance."""
instance = self.get(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "pause_instance",
"args": {"instance_id": instance_id}})
self._cast_compute_message('pause_instance', context, instance_id)
def unpause(self, context, instance_id):
"""Unpause the given instance."""
instance = self.get(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "unpause_instance",
"args": {"instance_id": instance_id}})
self._cast_compute_message('unpause_instance', context, instance_id)
def get_diagnostics(self, context, instance_id):
"""Retrieve diagnostics for the given instance."""
instance = self.get(context, instance_id)
host = instance["host"]
return rpc.call(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "get_diagnostics",
"args": {"instance_id": instance_id}})
self._cast_compute_message('get_diagnostics', context, instance_id)
def get_actions(self, context, instance_id):
"""Retrieve actions for the given instance."""
@@ -383,39 +365,23 @@ class API(base.Base):
def suspend(self, context, instance_id):
"""suspend the instance with instance_id"""
instance = self.get(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "suspend_instance",
"args": {"instance_id": instance_id}})
self._cast_compute_message('suspend_instance', context, instance_id)
def resume(self, context, instance_id):
"""resume the instance with instance_id"""
instance = self.get(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "resume_instance",
"args": {"instance_id": instance_id}})
self._cast_compute_message('resume_instance', context, instance_id)
def rescue(self, context, instance_id):
"""Rescue the given instance."""
instance = self.get(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "rescue_instance",
"args": {"instance_id": instance_id}})
self._cast_compute_message('rescue_instance', context, instance_id)
def unrescue(self, context, instance_id):
"""Unrescue the given instance."""
instance = self.get(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "unrescue_instance",
"args": {"instance_id": instance['id']}})
self._cast_compute_message('unrescue_instance', context, instance_id)
def set_admin_password(self, context, instance_id):
"""Set the root/admin password for the given instance."""
self._cast_compute_message('set_admin_password', context, instance_id)
def get_ajax_console(self, context, instance_id):
"""Get a url to an AJAX Console"""
@@ -437,35 +403,16 @@ class API(base.Base):
output['token'])}
def lock(self, context, instance_id):
"""
lock the instance with instance_id
"""
instance = self.get_instance(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "lock_instance",
"args": {"instance_id": instance['id']}})
"""lock the instance with instance_id"""
self._cast_compute_message('lock_instance', context, instance_id)
def unlock(self, context, instance_id):
"""
unlock the instance with instance_id
"""
instance = self.get_instance(context, instance_id)
host = instance['host']
rpc.cast(context,
self.db.queue_get_for(context, FLAGS.compute_topic, host),
{"method": "unlock_instance",
"args": {"instance_id": instance['id']}})
"""unlock the instance with instance_id"""
self._cast_compute_message('unlock_instance', context, instance_id)
def get_lock(self, context, instance_id):
"""
return the boolean state of (instance with instance_id)'s lock
"""
instance = self.get_instance(context, instance_id)
"""return the boolean state of (instance with instance_id)'s lock"""
instance = self.get(context, instance_id)
return instance['locked']
def attach_volume(self, context, instance_id, volume_id, device):

33
nova/compute/manager.py
View File

@@ -35,6 +35,8 @@ terminating it.
"""
import datetime
import random
import string
import logging
import socket
import functools
@@ -54,6 +56,8 @@ flags.DEFINE_string('compute_driver', 'nova.virt.connection.get_connection',
'Driver to use for controlling virtualization')
flags.DEFINE_string('stub_network', False,
'Stub network related code')
flags.DEFINE_integer('password_length', 12,
'Length of generated admin passwords')
flags.DEFINE_string('console_host', socket.gethostname(),
'Console proxy host to use to connect to instances on'
'this host.')
@@ -309,6 +313,35 @@ class ComputeManager(manager.Manager):
self.driver.snapshot(instance_ref, name)
@exception.wrap_exception
@checks_instance_lock
def set_admin_password(self, context, instance_id, new_pass=None):
"""Set the root/admin password for an instance on this server."""
context = context.elevated()
instance_ref = self.db.instance_get(context, instance_id)
if instance_ref['state'] != power_state.RUNNING:
logging.warn('trying to reset the password on a non-running '
'instance: %s (state: %s expected: %s)',
instance_ref['id'],
instance_ref['state'],
power_state.RUNNING)
logging.debug('instance %s: setting admin password',
instance_ref['name'])
if new_pass is None:
# Generate a random password
new_pass = self._generate_password(FLAGS.password_length)
self.driver.set_admin_password(instance_ref, new_pass)
self._update_state(context, instance_id)
def _generate_password(self, length=20):
"""Generate a random sequence of letters and digits
to be used as a password.
"""
chrs = string.letters + string.digits
return "".join([random.choice(chrs) for i in xrange(length)])
@exception.wrap_exception
@checks_instance_lock
def rescue_instance(self, context, instance_id):

4
nova/exception.py
View File

@@ -76,6 +76,10 @@ class InvalidInputException(Error):
pass
class TimeoutException(Error):
pass
def wrap_exception(f):
def _wrap(*args, **kw):
try:

7
nova/tests/test_compute.py
View File

@@ -151,6 +151,13 @@ class ComputeTestCase(test.TestCase):
self.compute.reboot_instance(self.context, instance_id)
self.compute.terminate_instance(self.context, instance_id)
def test_set_admin_password(self):
"""Ensure instance can have its admin password set"""
instance_id = self._create_instance()
self.compute.run_instance(self.context, instance_id)
self.compute.set_admin_password(self.context, instance_id)
self.compute.terminate_instance(self.context, instance_id)
def test_snapshot(self):
"""Ensure instance can be snapshotted"""
instance_id = self._create_instance()

27
nova/tests/test_xenapi.py
View File

@@ -31,6 +31,7 @@ from nova.compute import power_state
from nova.virt import xenapi_conn
from nova.virt.xenapi import fake as xenapi_fake
from nova.virt.xenapi import volume_utils
from nova.virt.xenapi.vmops import SimpleDH
from nova.tests.db import fakes as db_fakes
from nova.tests.xenapi import stubs
@@ -262,3 +263,29 @@ class XenAPIVMTestCase(test.TestCase):
instance = db.instance_create(values)
self.conn.spawn(instance)
return instance
class XenAPIDiffieHellmanTestCase(test.TestCase):
"""
Unit tests for Diffie-Hellman code
"""
def setUp(self):
super(XenAPIDiffieHellmanTestCase, self).setUp()
self.alice = SimpleDH()
self.bob = SimpleDH()
def test_shared(self):
alice_pub = self.alice.get_public()
bob_pub = self.bob.get_public()
alice_shared = self.alice.compute_shared(bob_pub)
bob_shared = self.bob.compute_shared(alice_pub)
self.assertEquals(alice_shared, bob_shared)
def test_encryption(self):
msg = "This is a top-secret message"
enc = self.alice.encrypt(msg)
dec = self.bob.decrypt(enc)
self.assertEquals(dec, msg)
def tearDown(self):
super(XenAPIDiffieHellmanTestCase, self).tearDown()

21
nova/virt/fake.py
View File

@@ -98,7 +98,7 @@ class FakeConnection(object):
the new instance.
The work will be done asynchronously. This function returns a
Deferred that allows the caller to detect when it is complete.
task that allows the caller to detect when it is complete.
Once this successfully completes, the instance should be
running (power_state.RUNNING).
@@ -122,7 +122,7 @@ class FakeConnection(object):
The second parameter is the name of the snapshot.
The work will be done asynchronously. This function returns a
Deferred that allows the caller to detect when it is complete.
task that allows the caller to detect when it is complete.
"""
pass
@@ -134,7 +134,20 @@ class FakeConnection(object):
and so the instance is being specified as instance.name.
The work will be done asynchronously. This function returns a
Deferred that allows the caller to detect when it is complete.
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.
The first parameter is an instance of nova.compute.service.Instance,
and so the instance is being specified as instance.name. The second
parameter is the value of the new password.
The work will be done asynchronously. This function returns a
task that allows the caller to detect when it is complete.
"""
pass
@@ -182,7 +195,7 @@ class FakeConnection(object):
and so the instance is being specified as instance.name.
The work will be done asynchronously. This function returns a
Deferred that allows the caller to detect when it is complete.
task that allows the caller to detect when it is complete.
"""
del self.instances[instance.name]

199
nova/virt/xenapi/vmops.py
View File

@@ -20,6 +20,11 @@ Management class for VM-related functions (spawn, reboot, etc).
"""
import json
import M2Crypto
import os
import subprocess
import tempfile
import uuid
from nova import db
from nova import context
@@ -127,12 +132,31 @@ class VMOps(object):
"""Refactored out the common code of many methods that receive either
a vm name or a vm instance, and want a vm instance in return.
"""
vm = None
try:
instance_name = instance_or_vm.name
vm = VMHelper.lookup(self._session, instance_name)
except AttributeError:
# A vm opaque ref was passed
vm = instance_or_vm
if instance_or_vm.startswith("OpaqueRef:"):
# Got passed an opaque ref; return it
return instance_or_vm
else:
# Must be the instance name
instance_name = instance_or_vm
except (AttributeError, KeyError):
# Note the the KeyError will only happen with fakes.py
# Not a string; must be an ID or a vm instance
if isinstance(instance_or_vm, (int, long)):
ctx = context.get_admin_context()
try:
instance_obj = db.instance_get_by_id(ctx, instance_or_vm)
instance_name = instance_obj.name
except exception.NotFound:
# The unit tests screw this up, as they use an integer for
# the vm name. I'd fix that up, but that's a matter for
# another bug report. So for now, just try with the passed
# value
instance_name = instance_or_vm
else:
instance_name = instance_or_vm.name
vm = VMHelper.lookup(self._session, instance_name)
if vm is None:
raise Exception(_('Instance not present %s') % instance_name)
return vm
@@ -189,6 +213,44 @@ class VMOps(object):
task = self._session.call_xenapi('Async.VM.clean_reboot', vm)
self._session.wait_for_task(instance.id, task)
def set_admin_password(self, instance, new_pass):
"""Set the root/admin password on the VM instance. This is done via
an agent running on the VM. Communication between nova and the agent
is done via writing xenstore records. Since communication is done over
the XenAPI RPC calls, we need to encrypt the password. We're using a
simple Diffie-Hellman class instead of the more advanced one in
M2Crypto for compatibility with the agent code.
"""
# Need to uniquely identify this request.
transaction_id = str(uuid.uuid4())
# The simple Diffie-Hellman class is used to manage key exchange.
dh = SimpleDH()
args = {'id': transaction_id, 'pub': str(dh.get_public())}
resp = self._make_agent_call('key_init', instance, '', args)
if resp is None:
# No response from the agent
return
resp_dict = json.loads(resp)
# Successful return code from key_init is 'D0'
if resp_dict['returncode'] != 'D0':
# There was some sort of error; the message will contain
# a description of the error.
raise RuntimeError(resp_dict['message'])
agent_pub = int(resp_dict['message'])
dh.compute_shared(agent_pub)
enc_pass = dh.encrypt(new_pass)
# Send the encrypted password
args['enc_pass'] = enc_pass
resp = self._make_agent_call('password', instance, '', args)
if resp is None:
# No response from the agent
return
resp_dict = json.loads(resp)
# Successful return code from password is '0'
if resp_dict['returncode'] != '0':
raise RuntimeError(resp_dict['message'])
return resp_dict['message']
def destroy(self, instance):
"""Destroy VM instance"""
vm = VMHelper.lookup(self._session, instance.name)
@@ -246,30 +308,19 @@ class VMOps(object):
def suspend(self, instance, callback):
"""suspend the specified instance"""
instance_name = instance.name
vm = VMHelper.lookup(self._session, instance_name)
if vm is None:
raise Exception(_("suspend: instance not present %s") %
instance_name)
vm = self._get_vm_opaque_ref(instance)
task = self._session.call_xenapi('Async.VM.suspend', vm)
self._wait_with_callback(instance.id, task, callback)
def resume(self, instance, callback):
"""resume the specified instance"""
instance_name = instance.name
vm = VMHelper.lookup(self._session, instance_name)
if vm is None:
raise Exception(_("resume: instance not present %s") %
instance_name)
vm = self._get_vm_opaque_ref(instance)
task = self._session.call_xenapi('Async.VM.resume', vm, False, True)
self._wait_with_callback(instance.id, task, callback)
def get_info(self, instance_id):
def get_info(self, instance):
"""Return data about VM instance"""
vm = VMHelper.lookup(self._session, instance_id)
if vm is None:
raise exception.NotFound(_('Instance not'
' found %s') % instance_id)
vm = self._get_vm_opaque_ref(instance)
rec = self._session.get_xenapi().VM.get_record(vm)
return VMHelper.compile_info(rec)
@@ -333,22 +384,34 @@ class VMOps(object):
return self._make_plugin_call('xenstore.py', method=method, vm=vm,
path=path, addl_args=addl_args)
def _make_agent_call(self, method, vm, path, addl_args={}):
"""Abstracts out the interaction with the agent xenapi plugin."""
return self._make_plugin_call('agent', method=method, vm=vm,
path=path, addl_args=addl_args)
def _make_plugin_call(self, plugin, method, vm, path, addl_args={}):
"""Abstracts out the process of calling a method of a xenapi plugin.
Any errors raised by the plugin will in turn raise a RuntimeError here.
"""
instance_id = vm.id
vm = self._get_vm_opaque_ref(vm)
rec = self._session.get_xenapi().VM.get_record(vm)
args = {'dom_id': rec['domid'], 'path': path}
args.update(addl_args)
# If the 'testing_mode' attribute is set, add that to the args.
if getattr(self, 'testing_mode', False):
args['testing_mode'] = 'true'
try:
task = self._session.async_call_plugin(plugin, method, args)
ret = self._session.wait_for_task(0, task)
ret = self._session.wait_for_task(instance_id, task)
except self.XenAPI.Failure, e:
raise RuntimeError("%s" % e.details[-1])
ret = None
err_trace = e.details[-1]
err_msg = err_trace.splitlines()[-1]
strargs = str(args)
if 'TIMEOUT:' in err_msg:
LOG.error(_('TIMEOUT: The call to %(method)s timed out. '
'VM id=%(instance_id)s; args=%(strargs)s') % locals())
else:
LOG.error(_('The call to %(method)s returned an error: %(e)s. '
'VM id=%(instance_id)s; args=%(strargs)s') % locals())
return ret
def add_to_xenstore(self, vm, path, key, value):
@@ -460,3 +523,89 @@ class VMOps(object):
"""Removes all data from the xenstore parameter record for this VM."""
self.write_to_param_xenstore(instance_or_vm, {})
########################################################################
def _runproc(cmd):
pipe = subprocess.PIPE
return subprocess.Popen([cmd], shell=True, stdin=pipe, stdout=pipe,
stderr=pipe, close_fds=True)
class SimpleDH(object):
"""This class wraps all the functionality needed to implement
basic Diffie-Hellman-Merkle key exchange in Python. It features
intelligent defaults for the prime and base numbers needed for the
calculation, while allowing you to supply your own. It requires that
the openssl binary be installed on the system on which this is run,
as it uses that to handle the encryption and decryption. If openssl
is not available, a RuntimeError will be raised.
"""
def __init__(self, prime=None, base=None, secret=None):
"""You can specify the values for prime and base if you wish;
otherwise, reasonable default values will be used.
"""
if prime is None:
self._prime = 162259276829213363391578010288127
else:
self._prime = prime
if base is None:
self._base = 5
else:
self._base = base
self._shared = self._public = None
self._dh = M2Crypto.DH.set_params(
self.dec_to_mpi(self._prime),
self.dec_to_mpi(self._base))
self._dh.gen_key()
self._public = self.mpi_to_dec(self._dh.pub)
def get_public(self):
return self._public
def compute_shared(self, other):
self._shared = self.bin_to_dec(
self._dh.compute_key(self.dec_to_mpi(other)))
return self._shared
def mpi_to_dec(self, mpi):
bn = M2Crypto.m2.mpi_to_bn(mpi)
hexval = M2Crypto.m2.bn_to_hex(bn)
dec = int(hexval, 16)
return dec
def bin_to_dec(self, binval):
bn = M2Crypto.m2.bin_to_bn(binval)
hexval = M2Crypto.m2.bn_to_hex(bn)
dec = int(hexval, 16)
return dec
def dec_to_mpi(self, dec):
bn = M2Crypto.m2.dec_to_bn('%s' % dec)
mpi = M2Crypto.m2.bn_to_mpi(bn)
return mpi
def _run_ssl(self, text, which):
base_cmd = ('cat %(tmpfile)s | openssl enc -aes-128-cbc '
'-a -pass pass:%(shared)s -nosalt %(dec_flag)s')
if which.lower()[0] == 'd':
dec_flag = ' -d'
else:
dec_flag = ''
fd, tmpfile = tempfile.mkstemp()
os.close(fd)
file(tmpfile, 'w').write(text)
shared = self._shared
cmd = base_cmd % locals()
proc = _runproc(cmd)
proc.wait()
err = proc.stderr.read()
if err:
raise RuntimeError(_('OpenSSL error: %s') % err)
return proc.stdout.read()
def encrypt(self, text):
return self._run_ssl(text, 'enc')
def decrypt(self, text):
return self._run_ssl(text, 'dec')

7
nova/virt/xenapi_conn.py
View File

@@ -149,6 +149,10 @@ class XenAPIConnection(object):
"""Reboot VM instance"""
self._vmops.reboot(instance)
def set_admin_password(self, instance, new_pass):
"""Set the root/admin password on the VM instance"""
self._vmops.set_admin_password(instance, new_pass)
def destroy(self, instance):
"""Destroy VM instance"""
self._vmops.destroy(instance)
@@ -266,7 +270,8 @@ class XenAPISession(object):
def _poll_task(self, id, task, done):
"""Poll the given XenAPI task, and fire the given action if we
get a result."""
get a result.
"""
try:
name = self._session.xenapi.task.get_name_label(task)
status = self._session.xenapi.task.get_status(task)

126
plugins/xenserver/xenapi/etc/xapi.d/plugins/agent Executable file
View File

@@ -0,0 +1,126 @@
#!/usr/bin/env python
# Copyright (c) 2011 Citrix Systems, Inc.
# Copyright 2011 OpenStack LLC.
# Copyright 2011 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# 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 reading/writing information to xenstore
#
try:
import json
except ImportError:
import simplejson as json
import os
import random
import subprocess
import tempfile
import time
import XenAPIPlugin
from pluginlib_nova import *
configure_logging("xenstore")
import xenstore
AGENT_TIMEOUT = 30
def jsonify(fnc):
def wrapper(*args, **kwargs):
return json.dumps(fnc(*args, **kwargs))
return wrapper
class TimeoutError(StandardError):
pass
@jsonify
def key_init(self, arg_dict):
"""Handles the Diffie-Hellman key exchange with the agent to
establish the shared secret key used to encrypt/decrypt sensitive
info to be passed, such as passwords. Returns the shared
secret key value.
"""
pub = int(arg_dict["pub"])
arg_dict["value"] = json.dumps({"name": "keyinit", "value": pub})
request_id = arg_dict["id"]
arg_dict["path"] = "data/host/%s" % request_id
xenstore.write_record(self, arg_dict)
try:
resp = _wait_for_agent(self, request_id, arg_dict)
except TimeoutError, e:
raise PluginError("%s" % e)
return resp
@jsonify
def password(self, arg_dict):
"""Writes a request to xenstore that tells the agent to set
the root password for the given VM. The password should be
encrypted using the shared secret key that was returned by a
previous call to key_init. The encrypted password value should
be passed as the value for the 'enc_pass' key in arg_dict.
"""
pub = int(arg_dict["pub"])
enc_pass = arg_dict["enc_pass"]
arg_dict["value"] = json.dumps({"name": "password", "value": enc_pass})
request_id = arg_dict["id"]
arg_dict["path"] = "data/host/%s" % request_id
xenstore.write_record(self, arg_dict)
try:
resp = _wait_for_agent(self, request_id, arg_dict)
except TimeoutError, e:
raise PluginError("%s" % e)
return resp
def _wait_for_agent(self, request_id, arg_dict):
"""Periodically checks xenstore for a response from the agent.
The request is always written to 'data/host/{id}', and
the agent's response for that request will be in 'data/guest/{id}'.
If no value appears from the agent within the time specified by
AGENT_TIMEOUT, the original request is deleted and a TimeoutError
is returned.
"""
arg_dict["path"] = "data/guest/%s" % request_id
arg_dict["ignore_missing_path"] = True
start = time.time()
while True:
if time.time() - start > AGENT_TIMEOUT:
# No response within the timeout period; bail out
# First, delete the request record
arg_dict["path"] = "data/host/%s" % request_id
xenstore.delete_record(self, arg_dict)
raise TimeoutError("TIMEOUT: No response from agent within %s seconds." %
AGENT_TIMEOUT)
ret = xenstore.read_record(self, arg_dict)
# Note: the response for None with be a string that includes
# double quotes.
if ret != '"None"':
# The agent responded
return ret
else:
time.sleep(3)
if __name__ == "__main__":
XenAPIPlugin.dispatch(
{"key_init": key_init,
"password": password})