Newer
Older
# This file is part of CloudControl.
#
# CloudControl is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# CloudControl is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with CloudControl. If not, see <http://www.gnu.org/licenses/>.
from sjrpc.core import RpcError
from cloudcontrol.common.datastructures.orderedset import OrderedSet
from cloudcontrol.server.conf import CCConf
from cloudcontrol.server.exceptions import (ReservedTagError, BadRoleError,
NotConnectedAccountError, CloneError)
from cloudcontrol.server.election import Elector
from cloudcontrol.server.repository import RepositoryOperationError
from cloudcontrol.server.handlers import listed, Reporter
from cloudcontrol.server.clients import Client, RegisteredCCHandler
from cloudcontrol.server.jobs import (ColdMigrationJob, HotMigrationJob,
from cloudcontrol.common.tql.db.tag import StaticTag
MIGRATION_TYPES = {'cold': ColdMigrationJob,
'hot': HotMigrationJob,}
class CliHandler(RegisteredCCHandler):
""" Handler binded to the 'cli' role.
.. currentmodule:: cloudcontrol.server.clients.cli
.. autosummary::
CliHandler.list
CliHandler.start
CliHandler.stop
CliHandler.destroy
CliHandler.pause
CliHandler.resume
CliHandler.disablevirtiocache
CliHandler.autostart
CliHandler.undefine
CliHandler.passwd
CliHandler.addaccount
CliHandler.copyaccount
CliHandler.addtag
CliHandler.deltag
CliHandler.tags
CliHandler.delaccount
CliHandler.close
CliHandler.declose
CliHandler.kill
CliHandler.execute
CliHandler.shutdown
CliHandler.jobs
CliHandler.cancel
CliHandler.purge
CliHandler.attachment
CliHandler.loadscript
CliHandler.savescript
CliHandler.delscript
CliHandler.runscript
CliHandler.loadplugin
CliHandler.saveplugin
CliHandler.delplugin
CliHandler.installplugin
CliHandler.uninstallplugin
CliHandler.runplugin
CliHandler.electiontypes
CliHandler.election
CliHandler.migrate
CliHandler.clone
CliHandler.shell
CliHandler.resize
""" List all objects registered on this instance.
:param query: the query to select objects to show
"""
self.logger.debug('Executed list function with query %s', query)
objects = self.client.list(query, method=method)
order = OrderedSet(['id'])
#if tags is not None:
# order |= OrderedSet(tags)
return {'objects': objects, 'order': list(order)}
@listed
def wall(self, message):
""" Send a wall to all connected users.
"""
self.client.check('wall')
self.server.wall(self.client.login, message)
@listed
def loadmotd(self):
""" Load and return the message of the day.
"""
return self.server.load_motd()
@listed
def savemotd(self, motd):
""" Save a new message of the day.
"""
self.server.save_motd(motd)
#
# VM actions:
#
def _vm_action(self, query, method, *args, **kwargs):
""" Do an action on a virtual machine.
"""
errs = Reporter()
# Search all hypervisors of selected vms:
for vm in self.client.list(query, show=('r', 'h', 'p'),
method=method[3:]): # Strip the vm_ prefix
if vm['r'] != 'vm':
errs.error(vm['id'], 'not a vm')
else:
hvclient = self.server.get_client(vm['p'])
if hvclient is None:
errs.error(vm['id'], 'offline hypervisor')
else:
try:
hvclient.vm_action(method, vm['h'], *args, **kwargs)
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
except Exception as err:
errs.error(vm['id'], str(err))
else:
errs.success(vm['id'], 'ok')
return errs.get_dict()
@listed
def start(self, query):
""" Start a virtual machine.
"""
return self._vm_action(query, 'vm_start')
@listed
def stop(self, query):
""" Stop a virtual machine.
"""
return self._vm_action(query, 'vm_stop')
@listed
def destroy(self, query):
""" Destroy (hard shutdown) a virtual machine.
"""
return self._vm_action(query, 'vm_destroy')
@listed
def pause(self, query):
""" Pause a virtual machine.
"""
return self._vm_action(query, 'vm_suspend')
@listed
def resume(self, query):
""" Resume a virtual machine.
"""
return self._vm_action(query, 'vm_resume')
@listed
def disablevirtiocache(self, query):
""" Set virtio cache to none on VMs disk devices.
:param query: tql query
"""
return self._vm_action(query, 'vm_disable_virtio_cache')
@listed
def autostart(self, query, flag):
""" Set autostart flag on VMs.
:param query: tql query
:param bool flag: autostart value to set
"""
return self._vm_action(query, 'vm_set_autostart', flag)
@listed
def undefine(self, query, delete_storage=True):
""" Undefine selected virtual machines.
:param query: the tql query to select objects.
:param delete_storage: delete storage of vm.
:return: a dict where key is the id of a selected object, and the value
a tuple (errcode, message) where errcode is (success|error|warn) and
message an error message or the output of the command in case of
success.
"""
objects = self.client.list(query, show=('r', 'p', 'h', 'disk*',), method='undefine')
errs = Reporter()
for obj in objects:
if obj['r'] != 'vm':
errs.error(obj['id'], 'bad role')
continue
try:
hvcon = self.server.get_client(obj['p'])
except KeyError:
errs.error(obj['id'], 'hypervisor not connected')
else:
if delete_storage:
for disk in obj.get('disk', '').split():
pool = obj.get('disk%s_pool' % disk)
name = obj.get('disk%s_vol' % disk)
hvcon.proxy.vol_delete(pool, name)
hvcon.proxy.vm_undefine(obj['h'])
errs.success(obj['id'], 'vm undefined')
return errs.get_dict()
#
# Account management:
#
@listed
def passwd(self, query, password, method='ssha'):
""" Define a new password for selected users.
:param query: the query to select the objects to change
:param password: the password to set (None to remove password)
:param method: the hash method (sha, ssha, md5, smd5 or plain)
:return: a standard report output
"""
objects = self.client.list(query, show=('a',), method='passwd')
errs = Reporter()
with self.conf:
for obj in objects:
if 'a' not in obj:
errs.error(obj['id'], 'not an account')
continue
self.conf.set_password(obj['a'], password, method)
errs.success(obj['id'], 'password updated')
return errs.get_dict()
@listed
def addaccount(self, login, role, password=None):
""" Create a new account with specified login.
:param login: the login of the new account
:param role: the role of the new account
:param password: the password of the new account (None = not set)
"""
self.client.check('addaccount')
if role in Client.roles:
self.conf.create_account(login, role, password)
else:
raise BadRoleError('%r is not a legal role.' % role)
@listed
def copyaccount(self, copy_login, login, password=None):
""" Create a new account with specified login.
:param copy_login: the login of the account to copy
:param login: the login of the new account
:param password: the password of the new account (default None)
"""
self.client.check('addaccount')
self.conf.copy_account(copy_login, login, password)
@listed
def addtag(self, query, tag_name, tag_value):
""" Add a tag to the accounts which match the specified query.
:param query: the query to select objects
:param tag_name: the name of the tag to add
:param tag_value: the value of the tag
"""
if tag_name in self.server.RESERVED_TAGS:
raise ReservedTagError('Tag %r is read-only' % tag_name)
objects = self.client.list(query, show=('a', 'p', 'h'), method='addtag')
errs = Reporter()
with self.conf:
for obj in objects:
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
if 'a' in obj:
# Update the configuration for this account:
tags = self.conf.show(obj['a'])['tags']
if tag_name in tags:
errs.warn(obj['id'], 'tag already exists, changed from %s'
' to %s' % (tags[tag_name], tag_value))
# Update the object db (update the tag value):
dbobj = self.server.db.get(obj['id'])
dbobj[tag_name].value = tag_value
else:
errs.success(obj['id'], 'tag created')
# Update the object db (create the tag):
dbobj = self.server.db.get(obj['id'])
dbobj.register(StaticTag(tag_name, tag_value), override=True)
self.conf.add_tag(obj['a'], tag_name, tag_value)
elif 'p' in obj:
try:
hvcon = self.server.get_client(obj['p'])
except KeyError:
errs.error(obj['id'], 'hypervisor not connected')
else:
try:
hvcon.proxy.tag_add(obj['h'], tag_name, tag_value)
except NameError:
errs.error(obj['id'], 'hypervisor does not handle tags on VMs')
except Exception, err:
errs.error(obj['id'], str(err))
else:
errs.success(obj['id'], 'tag set on VM')
errs.error(obj['id'], 'this object cannot have static tags')
continue
return errs.get_dict()
@listed
def deltag(self, query, tag_name):
""" Remove a tag of the selected accounts.
:param query: the query to select objects
:param tag_name: the name of the tag to remove
"""
if tag_name in self.server.RESERVED_TAGS:
raise ReservedTagError('Tag %r is read-only' % tag_name)
objects = self.client.list(query, show=('a', 'p', 'h'), method='deltag')
errs = Reporter()
with self.conf:
for obj in objects:
if 'a' in obj:
tags = self.conf.show(obj['a'])['tags']
if tag_name in tags:
errs.success(obj['id'], 'tag deleted')
dbobj = self.server.db.get(obj['id'])
dbobj.unregister(tag_name, override=True)
else:
errs.warn(obj['id'], 'unknown tag')
self.server.conf.remove_tag(obj['a'], tag_name)
elif 'p' in obj:
try:
hvcon = self.server.get_client(obj['p'])
except KeyError:
errs.error(obj['id'], 'hypervisor not connected')
else:
try:
hvcon.proxy.tag_delete(obj['h'], tag_name)
except NameError:
errs.error(obj['id'], 'hypervisor does not handle tags on VMs')
except Exception, err:
errs.error(obj['id'], str(err))
else:
errs.success(obj['id'], 'tag deleted on VM')
errs.error(obj['id'], 'this object cannot have static tags')
continue
return errs.get_dict()
@listed
def tags(self, query):
""" Return all static tags attached to the selected accounts.
:param query: the query to select objects
"""
objects = self.client.list(query, show=('a', 'p', 'h'), method='tags')
tags = []
for obj in objects:
o = {'id': obj['id']}
if 'a' in obj:
otags = self.server.conf.show(obj['a'])['tags']
otags.update({'id': obj['id']})
o.update(otags)
elif 'p' in obj:
try:
hvcon = self.server.get_client(obj['p'])
except KeyError:
pass
else:
try:
otags = hvcon.proxy.tag_show(obj['h'])
except:
self.logger.exception('Unable to get tags, node error')
else:
otags.update({'id': obj['id']})
o.update(otags)
tags.append(o)
return {'objects': tags, 'order': ['id']}
@listed
def delaccount(self, query):
""" Delete the accounts selected by query.
:param query: the query to select objects
"""
objects = self.client.list(query, show=('a',), method='delaccount')
errs = Reporter()
with self.server.conf:
for obj in objects:
if 'a' not in obj:
errs.error(obj['id'], 'not an account')
continue
try:
self.server.conf.remove_account(obj['a'])
except CCConf.UnknownAccount:
errs.error(obj['id'], 'unknown account')
else:
errs.success(obj['id'], 'account deleted')
self.server.jobs.spawn(KillClientJob, self.client.login,
settings={'server': self.server,
'account': obj['a'],
'gracetime': 1})
return errs.get_dict()
@listed
def close(self, query):
""" Close selected account an account without deleting them.
:param query: the query to select objects
"""
objects = self.client.list(query, show=('a',), method='close')
errs = Reporter()
with self.server.conf:
for obj in objects:
if 'a' not in obj:
errs.error(obj['id'], 'not an account')
continue
tags = self.server.conf.show(obj['a'])['tags']
if 'close' in tags:
errs.warn(obj['id'], 'account already closed')
continue
errs.success(obj['id'], 'closed')
self.server.conf.add_tag(obj['a'], 'close', 'yes')
dbobj = self.server.db.get(obj['id'])
dbobj.register(StaticTag('close', 'yes'), override=True)
self.server.jobs.spawn(KillClientJob, self.client.login,
settings={'server': self.server,
'account': obj['a'],
'gracetime': 1})
return errs.get_dict()
@listed
def declose(self, query):
""" Re-open selected closed accounts.
:param query: the query to select objects
"""
objects = self.client.list(query, show=('a',), method='declose')
errs = Reporter()
with self.server.conf:
for obj in objects:
if 'a' not in obj:
errs.error(obj['id'], 'not an account')
continue
tags = self.conf.show(obj['a'])['tags']
if 'close' in tags:
errs.success(obj['id'], 'account declosed')
self.conf.remove_tag(obj['a'], 'close')
dbobj = self.server.db.get(obj['id'])
dbobj.unregister('close', override=True)
else:
errs.warn(obj['id'], 'account not closed')
return errs.get_dict()
@listed
def kill(self, query):
""" Disconnect all connected accounts selected by query.
:param query: the query to select objects
"""
objects = self.client.list(query, show=set(('a',)), method='kill')
errs = Reporter()
with self.server.conf:
for obj in objects:
if 'a' not in obj:
errs.error(obj['id'], 'not an account')
continue
try:
self.server.kill(obj['a'])
except NotConnectedAccountError:
errs.error(obj['id'], 'account is not connected')
else:
errs.success(obj['id'], 'account killed')
return errs.get_dict()
#
# Rights management:
#
def loadrights(self):
""" Get the current ruleset.
self.client.check('rights')
return self.server.rights.export()
def saverights(self, ruleset):
""" Set the current ruleset.
:param ruleset: the ruleset to load.
self.client.check('rights')
self.server.rights.load(ruleset)
#
# Nodes actions:
#
@listed
def execute(self, query, command):
""" Execute command on matched objects (must be roles hv or host).
:param query: the tql query to select objects.
:param command: the command to execute on each object
:return: a dict where key is the id of a selected object, and the value
a tuple (errcode, message) where errcode is (success|error|warn) and
message an error message or the output of the command in case of
success.
"""
objects = self.client.list(query, show=('r',), method='execute')
errs = Reporter()
for obj in objects:
if obj['r'] not in ('hv', 'host'):
errs.error(obj['id'], 'bad role')
continue
try:
client = self.server.get_client(obj['id'])
except KeyError:
errs.error(obj['id'], 'node not connected')
else:
try:
returned = client.execute(command)
except Exception as err:
errs.error(obj['id'], str(err))
else:
errs.success(obj['id'], 'command executed', output=returned)
return errs.get_dict()
@listed
def shutdown(self, query, reboot=True, gracefull=True):
""" Execute a shutdown on selected objects (must be roles hv or host).
:param query: the tql query to select objects.
:param reboot: reboot the host instead of just shut it off
:param gracefull: properly shutdown the host
:return: a dict where key is the id of a selected object, and the value
a tuple (errcode, message) where errcode is (success|error|warn) and
message an error message.
"""
objects = self.client.list(query, show=set(('r',)), method='execute')
errs = Reporter()
for obj in objects:
if obj['r'] not in ('hv', 'host'):
errs.error(obj['id'], 'bad role')
continue
try:
node = self.server.get_client(obj['id'])
except KeyError:
errs.error(obj['id'], 'node not connected')
else:
try:
except RpcError as err:
errs.error(obj['id'], '%s (exc: %s)' % (err.message,
err.exception))
else:
errs.success(obj['id'], 'ok')
return errs.get_dict()
#
# Jobs management:
#
def cancel(self, query):
""" Cancel a job.
:param query: the tql query used to select jobs to cancel
objects = self.client.list(query, show=('r', 'p'), method='cancel')
errs = Reporter()
for obj in objects:
if obj['r'] != 'job':
errs.error(obj['id'], 'not a job')
elif 'p' in obj:
sub_id = obj['id'][len(obj['p']) + 1:]
client = self.server.get_client(obj['p'])
try:
client.proxy.job_cancel(sub_id)
except Exception as err:
errs.error(obj['id'], '%s' % err)
else:
errs.success(obj['id'], 'job cancelled')
else:
try:
self.server.jobs.get(obj['id']).cancel()
except KeyError:
errs.error(obj['id'], 'unknown job')
else:
errs.success(obj['id'], 'job cancelled')
def purge(self, query):
""" Purge matching jobs from server.
:param query: the tql query used to select jobs to purge
.. note::
Purge only work for job with state done.
objects = self.client.list(query, show=('r', 'p', 'state'), method='purge')
errs = Reporter()
for obj in objects:
if obj['r'] != 'job':
errs.error(obj['id'], 'not a job')
elif obj['state'] != 'done':
errs.error(obj['id'], 'job must be done')
elif 'p' in obj:
sub_id = obj['id'][len(obj['p']) + 1:]
client = self.server.get_client(obj['p'])
try:
client.proxy.job_purge(sub_id)
except Exception as err:
errs.error(obj['id'], '%s' % err)
else:
errs.success(obj['id'], 'job purged')
else:
try:
self.server.jobs.purge(obj['id'])
except KeyError:
raise
errs.warn(obj['id'], 'job already purged')
else:
errs.success(obj['id'], 'job purged')
def attachment(self, query, name):
""" Get the specified attachment for jobs matching query.
:param query: the tql query used to select jobs
objects = self.client.list(query, show=('r', 'p'), method='attachment')
errs = Reporter()
for obj in objects:
if obj['r'] != 'job':
errs.error(obj['id'], 'not a job')
elif 'p' in obj:
sub_id = obj['id'][len(obj['p']) + 1:]
client = self.server.get_client(obj['p'])
try:
output = client.proxy.job_attachment(sub_id, name)
except Exception as err:
errs.error(obj['id'], '%s' % err)
else:
errs.success(obj['id'], 'ok', output=output)
job = self.server.jobs.get(obj['id'])
if job is None:
errs.error(obj['id'], 'unknown job')
attachment = job.read_attachment(name)
errs.success(obj['id'], 'ok', output=attachment)
#
# Script management:
#
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
@listed
def loadscript(self, name):
""" Get the content of a script stored ont he server.
:param name: name of the script to load
"""
self.client.check('script')
return self.server.scripts.load(name, empty_if_missing=True)[1]
@listed
def savescript(self, name, content):
""" Save a script on the server.
:param name: name of the script to save
:param content: content of the script to save
"""
self.client.check('script')
return self.server.scripts.save(name, content)
@listed
def delscript(self, query):
""" Delete a script on the server.
:param query: the tql query used to select scripts to delete
"""
objects = self.client.list(query, show=('r', 'name'), method='script')
errs = Reporter()
for obj in objects:
if obj['r'] != 'script':
errs.error(obj['id'], 'not a script')
else:
try:
self.server.scripts.delete(obj['name'])
except RepositoryOperationError as err:
errs.error(obj['id'], 'error: %s' % err)
else:
errs.success(obj['id'], 'script deleted')
return errs.get_dict()
@listed
def runscript(self, query, script, batch=None, *args):
""" Execute a script on matching nodes.
"""
# Load the script:
sha1_hash, _ = self.server.scripts.load(script)
objects = self.client.list(query, show=('r', ), method='script')
errs = Reporter()
for obj in objects:
if obj['r'] not in ('host', 'hv'):
errs.error(obj['id'], 'not a host')
else:
try:
node = self.server.get_client(obj['id'])
except KeyError:
errs.error(obj['id'], 'node not connected')
continue
try:
job_id = node.script_run(sha1_hash, script,
self.client.login, batch=batch,
*args)
except RpcError as err:
errs.error(obj['id'], '%s (exc: %s)' % (err.message,
err.exception))
else:
job_id = '.'.join((obj['id'], job_id))
errs.success(obj['id'], 'ok.', jobs=job_id)
return errs.get_dict()
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
#
# Plugins management:
#
@listed
def loadplugin(self, name):
""" Get the content of a plugin stored ont he server.
:param name: name of the plugin to load
"""
self.client.check('plugin')
return self.server.plugins.load(name, empty_if_missing=True)[1]
@listed
def saveplugin(self, name, content):
""" Save a plugin on the server.
:param name: name of the plugin to save
:param content: content of the plugin to save
"""
self.client.check('plugin')
return self.server.plugins.save(name, content)
@listed
def delplugin(self, query):
""" Delete a plugin on the server.
:param query: the tql query used to select plugins to delete
"""
objects = self.client.list(query, show=('r', 'name'), method='plugin')
errs = Reporter()
for obj in objects:
if obj['r'] != 'plugin':
errs.error(obj['id'], 'not a plugin')
else:
try:
self.server.plugins.delete(obj['name'])
except RepositoryOperationError as err:
errs.error(obj['id'], 'error: %s' % err)
else:
errs.success(obj['id'], 'plugin deleted')
return errs.get_dict()
@listed
def installplugin(self, query, plugin):
""" Install a plugin on matching nodes.
"""
# Load the plugin:
sha1_hash, _ = self.server.plugins.load(plugin)
objects = self.client.list(query, show=('r', ), method='plugin')
errs = Reporter()
for obj in objects:
if obj['r'] not in ('host', 'hv'):
errs.error(obj['id'], 'not a host')
else:
try:
node = self.server.get_client(obj['id'])
except KeyError:
errs.error(obj['id'], 'node not connected')
continue
try:
node.plugin_install(sha1_hash, plugin)
except RpcError as err:
errs.error(obj['id'], '%s (exc: %s)' % (err.message,
err.exception))
else:
errs.success(obj['id'], 'plugin installed')
return errs.get_dict()
@listed
def uninstallplugin(self, query, plugin):
""" Install a plugin on matching nodes.
"""
objects = self.client.list(query, show=('r', ), method='plugin')
errs = Reporter()
for obj in objects:
if obj['r'] not in ('host', 'hv'):
errs.error(obj['id'], 'not a host')
else:
try:
node = self.server.get_client(obj['id'])
except KeyError:
errs.error(obj['id'], 'node not connected')
continue
try:
node.plugin_uninstall(plugin)
except RpcError as err:
errs.error(obj['id'], '%s (exc: %s)' % (err.message,
err.exception))
else:
errs.success(obj['id'], 'plugin uninstalled')
return errs.get_dict()
@listed
def helpplugin(self, query, plugin, method):
""" Get help about a plugin installed on the matching host.
"""
objects = self.client.list(query, show=('r', ), method='plugin')
errs = Reporter()
for obj in objects:
if obj['r'] not in ('host', 'hw'):
errs.error(obj['id'], 'not a host')
else:
try:
node = self.server.get_client(obj['id'])
except KeyError:
errs.error(obj['id'], 'node not connected')
continue
try:
help = node.plugin_help(plugin, method)
except RpcError as err:
errs.error(obj['id'], '%s (exc: %s)' % (err.message,
err.exception))
else:
errs.success(obj['id'], 'ok.', output=help)
return errs.get_dict()
@listed
def runplugin(self, query, plugin, method, batch=None, **kwargs):
""" Execute a plugin method on matching nodes.
"""
# Load the plugin:
sha1_hash, _ = self.server.plugins.load(plugin)
objects = self.client.list(query, show=('r', ), method='plugin')
errs = Reporter()
for obj in objects:
if obj['r'] not in ('host', 'hv'):
errs.error(obj['id'], 'not a host')
else:
try:
node = self.server.get_client(obj['id'])
except KeyError:
errs.error(obj['id'], 'node not connected')
continue
try:
job_id = node.plugin_run(plugin, method, self.client.login,
batch=batch, **kwargs)
except RpcError as err:
errs.error(obj['id'], '%s (exc: %s)' % (err.message,
err.exception))
else:
job_id = '.'.join((obj['id'], job_id))
errs.success(obj['id'], 'ok.', jobs=job_id)
return errs.get_dict()
#
# Election / Migration / Cloning:
#
@listed
def electiontypes(self):
return Elector.ALGO_BY_TYPES
@listed
def election(self, query_vm, query_dest, mtype='cold', algo='fair', **kwargs):
""" Consult the server for the migration of specified vm on
an hypervisor pool.
:param query_vm: the tql query to select VMs to migrate
:param query_dest: the tql query to select destination hypervisors
candidates
:param mtype: type of migration
:param algo: algo used for distribution
"""
elector = Elector(self.server, query_vm, query_dest, self.client)
return elector.election(mtype, algo, **kwargs)
@listed
def migrate(self, migration_plan):
""" Launch the provided migration plan.
:param migration_plan: the plan of the migration.
:return: a standard error report
"""
errs = Reporter()
for migration in migration_plan:
# Check if the migration type is know:
if migration['type'] in MIGRATION_TYPES:
mtype = MIGRATION_TYPES[migration['type']]
else:
errmsg = '%r unknown migration type' % migration['type']
errs.error(migration['sid'], errmsg)
continue
vm = self.server.db.get_by_id(migration['sid'], ('h', 'hv', 'p'))
# Construct the migration properties:
migration_properties = {
'client': self.client,
'vm_name': vm['h'],
'hv_source': vm['p'],
'hv_dest': migration['did']
}
# Create the job:
self.client.spawn_job(mtype, settings=migration_properties)
errs.success(migration['sid'], 'migration launched')
return errs.get_dict()