# 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 . from sjrpc.core import RpcError from cloudcontrol.common.datastructures.orderedset import OrderedSet from cloudcontrol.common.allocation.vmspec import expand_vmspec 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, CloneJob, KillClientJob, AllocationJob, MigrationJob, DiskCopyJob) from cloudcontrol.server.db import SObject from cloudcontrol.common.tql.db.tag import StaticTag MIGRATION_TYPES = {'cold': ColdMigrationJob, 'hot': HotMigrationJob,} RESCUE_SCRIPT = 'rescue' class CliHandler(RegisteredCCHandler): """ Handler binded to the 'cli' role. Summary of methods: .. currentmodule:: cloudcontrol.server.clients.cli .. autosummary:: CliHandler.list CliHandler.wall 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.loadrights CliHandler.saverights 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.console CliHandler.shell CliHandler.resize CliHandler.forward CliHandler.dbstats """ @listed def list(self, query, method='list'): """ 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) except Exception as err: errs.error(vm['id'], str(err)) else: if method == 'vm_start': hvclient.vm_action('vm_set_autostart', vm['h'], True) elif method in ('vm_stop', 'vm_destroy'): hvclient.vm_action('vm_set_autostart', vm['h'], False) errs.success(vm['id'], 'ok') return errs.get_dict() @listed def start(self, query, pause=False): """ Start a virtual machine. """ return self._vm_action(query, 'vm_start', pause) @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 reset(self, query): """ Reset a virtual machine. """ return self._vm_action(query, 'vm_reset') @listed def cycle(self, query): """ Cycle a virtual machine. """ return self._vm_action(query, 'vm_cycle') @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 attachdisk(self, query, pool, volume, driver='virtio', bps=0, iops=0): """ Attach a disk on a VM. :param query: tql query """ return self._vm_action(query, 'vm_attach_disk', pool, volume, driver, bps, iops) @listed def detachdisk(self, query, pool, volume): """ Detach a disk from a VM. :param query: tql query """ return self._vm_action(query, 'vm_detach_disk', pool, volume) @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(): if obj.get('disk%s_shared' % disk) == 'no': 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() @listed def rescue(self, query): """ Enable rescue mode on selected virtual machines. :param query: the tql query to select objects. """ objects = self.client.list(query, show=('r', 'p', 'h', 'n', 'status'), method='rescue') 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: try: hvcon.proxy.vm_rescue(obj['h']) except Exception as err: errs.error(obj['id'], 'Unexpected exception: %s' % str(err)) else: if obj['status'] == 'stopped': errs.success(obj['id'], 'vm now in rescue mode') else: errs.success(obj['id'], 'vm will be in rescue mode after a restart') return errs.get_dict() @listed def unrescue(self, query): """ Disable rescue mode on selected virtual machines. :param query: the tql query to select objects. """ objects = self.client.list(query, show=('r', 'p', 'h', 'n', 'status'), method='rescue') 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: try: hvcon.proxy.vm_unrescue(obj['h']) except Exception as err: errs.error(obj['id'], 'Unexpected exception: %s' % str(err)) else: if obj['status'] == 'stopped': errs.success(obj['id'], 'vm now in normal mode') else: errs.success(obj['id'], 'vm will be in normal mode after a restart') return errs.get_dict() @listed def title(self, query, new_title): """ Edit the title of a VM. :param query: the tql query to select objects. """ objects = self.client.list(query, show=('r', 'p', 'h'), method='title') if not objects: raise NotImplementedError('No objects matched by query') elif len(objects) != 1: raise NotImplementedError('Can change title on only one VM at once') 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: try: hvcon.proxy.vm_change_title(obj['h'], new_title) except Exception as err: errs.error(obj['id'], str(err)) else: errs.success(obj['id'], 'title has been edited') return errs.get_dict() @listed def vlans(self, query, vlan_update, mac_address=None): """ Edit VLANs propagated to a VM interface. :param query: the tql query to select objects. """ objects = self.client.list(query, show=('r', 'p', 'h'), method='vlans') if not objects: raise NotImplementedError('No objects matched by query') 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: try: hvcon.proxy.vm_set_vlans(obj['h'], vlan_update, mac_address=mac_address) except Exception as err: errs.error(obj['id'], str(err)) else: errs.success(obj['id'], 'vlans updated') 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: 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') else: 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') else: 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: # @listed def loadrights(self): """ Get the current ruleset. """ self.client.check('rights') return self.server.rights.export() @listed def saverights(self, ruleset): """ Set the current ruleset. :param ruleset: the ruleset to load. """ self.client.check('rights') self.server.rights.load(ruleset) # # Riskgroups management: # @listed def getriskgroups(self): """ Get existing groups. """ self.client.check('riskgroups') return self.server.riskgroups.groups @listed def getriskgroup(self, name): """ Get specified group. """ self.client.check('riskgroups') return self.server.riskgroups.get(name) @listed def setriskgroup(self, name, properties): """ Set a new or an existing group. """ self.client.check('riskgroups') return self.server.riskgroups.set(name, properties) @listed def delriskgroup(self, name): """ Delete an existing group. """ self.client.check('riskgroups') return self.server.riskgroups.delete(name) # # 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: node.shutdown_node(reboot, gracefull) 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: # @listed 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') return errs.get_dict() @listed 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') return errs.get_dict() @listed 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) else: job = self.server.jobs.get(obj['id']) if job is None: errs.error(obj['id'], 'unknown job') else: attachment = job.read_attachment(name) errs.success(obj['id'], 'ok', output=attachment) return errs.get_dict() # # Script management: # @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() # # 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 / Allocation # @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, 'server': self.server, 'vm_name': vm['h'], 'hv_source': vm['p'], 'hv_dest': migration['did'] } # Create the job: job = self.client.spawn_job(mtype, settings=migration_properties) errs.success(migration['sid'], 'migration launched, id:%s' % job.id) return errs.get_dict() @listed def migrate2(self, tql_vm, tql_target, batch=None, live=False, flags=None): """ Launch a migration. """ self.client.check('migrate') vms = self.client.list(tql_vm, show=('r', ), method='migrate') errs = Reporter() for vm in vms: if vm['r'] != 'vm': errs.error(vm['id'], 'not a vm') else: settings = {'server': self.server, 'client': self.client, 'vm_id': vm['id'], 'tql_target': tql_target, 'live': live, 'flags': flags} job = self.client.spawn_job(MigrationJob, batch=batch, settings=settings) errs.success(vm['id'], 'migration launched, id:%s' % job.id) return errs.get_dict() @listed def diskcopy(self, source_id, source_pool, source_vol, dest_id, dest_pool, dest_vol): """ Launch a migration. """ self.client.check('diskcopy') settings = {'server': self.server, 'client': self.client, 'source_id': source_id, 'source_pool': source_pool, 'source_vol': source_vol, 'dest_id': dest_id, 'dest_pool': dest_pool, 'dest_vol': dest_vol} job = self.client.spawn_job(DiskCopyJob, settings=settings) return job.id @listed def clone(self, tql_vm, tql_dest, name): """ Create and launch a clone job. :param tql_vm: a tql matching one vm object (the cloned vm) :param tql_dest: a tql matching one hypervisor object (the destination hypervisor) :param name: the new name of the VM """ vm = self.client.list(tql_vm, show=('r', 'h', 'p'), method='clone') if len(vm) != 1: raise CloneError('VM Tql must select ONE vm') elif vm[0]['r'] != 'vm': raise CloneError('Destination Tql must select a vm') else: vm = vm[0] dest = self.client.list(tql_dest, show=('r',), method='clone') if len(dest) != 1: raise CloneError('Destination Tql must select ONE hypervisor') elif dest[0]['r'] != 'hv': raise CloneError('Destination Tql must select an hypervisor') else: dest = dest[0] job = self.client.spawn_job(CloneJob, settings={'server': self.server, 'client': self.client, 'vm_name': vm['h'], 'new_vm_name': name, 'hv_source': vm['p'], 'hv_dest': dest['id']}) return job.id @listed def allocate(self, vmspec, tql_target): """ Allocate new VMs. :param vmspec: a vmspec structure :param tql_target: a TQL used as target restriction """ self.client.check('allocate') # Check and expand vmspec input: expanded_vmspec = expand_vmspec(vmspec) job = self.client.spawn_job(AllocationJob, settings={'server': self.server, 'client': self.client, 'expanded_vmspec': expanded_vmspec, 'tql_target': tql_target}) return job.id # # Remote console: # @listed def console(self, tql): """ Start a remote console on object matching the provided tql. :param tql: tql matching only one object on which start the console :return: the label of the created tunnel """ objects = self.client.list(tql, show=('r', 'p', 'h'), method='console') if not objects: raise NotImplementedError('No objects matched by query') elif len(objects) != 1: raise NotImplementedError('Console only support one tunnel at time for now') errs = Reporter() for obj in objects: if obj['r'] in ('vm',): client = self.server.get_client(obj['p']) try: srv_to_host_tun = client.console(obj['h']) except Exception as err: errs.error(obj['id'], str(err)) else: cli_tun = self.client.register_tunnel('console', client, srv_to_host_tun) errs.success(obj['id'], 'tunnel started.', output=cli_tun.label) else: errs.error(obj['id'], 'bad role') return errs.get_dict() # # Remote shell: # @listed def shell(self, tql): """ Start a remote shell on object matching the provided tql. :param tql: tql matching only one object on which start the shell :return: the label of the created tunnel """ objects = self.client.list(tql, show=('r', 'p'), method='shell') if not objects: raise NotImplementedError('No objects matched by query') elif len(objects) != 1: raise NotImplementedError('Shell only support one tunnel at time for now') errs = Reporter() for obj in objects: if obj['r'] in ('host', 'hv'): client = self.server.get_client(obj['id']) srv_to_host_tun = client.shell() cli_tun = self.client.register_tunnel('shell', client, srv_to_host_tun) errs.success(obj['id'], 'tunnel started.', output=cli_tun.label) else: errs.error(obj['id'], 'bad role') return errs.get_dict() @listed def resize(self, label, row, col, xpixel, ypixel): """ Send a resize event to the remote shell's tty. :param label: label of the tty tunnel to resize :param row: number of rows :param col: number of columns :param xpixel: unused :param ypixel: unused """ if label is None: tuns = [(c, st.label) for t, c, ct, st in self.client.tunnels.values() if t == 'shell'] else: ttype, client, ctun, stun = self.client.get_tunnel(label) if ttype != 'shell': raise ValueError('Label does not refers on a shell') tuns = [(client, stun.label)] for client, label in tuns: client.resize(label, row, col, xpixel, ypixel) # # Port forwarding: # @listed def forward(self, login, port, destination='127.0.0.1'): """ Forward a TCP port to the client. :param login: login of the remote client on which establish the tunnel :param port: port on which establish the tunnel on destination :param destination: tunnel destination (from the remote client side) """ self.client.check('forward', query='id=%s' % login) # Create the tunnel to the node: try: host_client = self.server.get_client(login) except KeyError: raise KeyError('Specified client is not connected') s2n_tun = host_client.forward(port, destination) # Create tunnel to the CLI c2s_tun = self.client.register_tunnel('forward', host_client, s2n_tun) return c2s_tun.label # # Debug: # @listed def dbstats(self): """ Get statistics about tql database. """ return self.server.db.stats() def forward_call(self, login, func, *args, **kwargs): """ Forward a call to a connected client and return result. :param login: login of the connected client :param func: function to execute on the client :param \*args, \*\*kwargs: arguments of the call """ self.client.check('forward_call') client = self.server.get_client(login) return client.conn.call(func, *args, **kwargs) class CliClient(Client): """ A cli client connected to the cc-server. """ ROLE = 'cli' RPC_HANDLER = CliHandler KILL_ALREADY_CONNECTED = True def __init__(self, *args, **kwargs): super(CliClient, self).__init__(*args, **kwargs) self._tunnels = {} # Running tunnels for this client (as client) @property def tunnels(self): """ Get active client tunnels by label (a dict). """ return self._tunnels def spawn_job(self, job_class, **kwargs): return self._server.jobs.spawn(job_class, self.account, **kwargs) def register_tunnel(self, ttype, client, tun, label=None): """ Create and register a tunnel for this client. :param ttype: type of tunnel :param client: client where the tunnel go :param tun: the tunnel of this client :param label: label of the tunnel to create """ def cb_on_shutdown(tun): # Call the default callback: tun.cb_default_on_shutdown(tun) # Delete the tunnel from the current running tunnels: self.unregister_tunnel(tun.label) ctun = self.conn.create_tunnel(label=label, endpoint=tun.socket, on_shutdown=cb_on_shutdown, close_endpoint_on_shutdown=False) self._tunnels[ctun.label] = (ttype, client, ctun, tun) return ctun def get_tunnel(self, label): """ Get the tunnel binded to the provided label. :return: a tuple (type, remote_client, tunnel, remote_client_tunnel) where: **type** is a string provided on tunnel creation, **remote_client** the client object of the remote client on which the tunnel is established, **tunnel** the cli-to-server tunnel object from the sjRpc, **remote_client_tunnel** the server-to-remote-client tunnel object from the sjRpc. """ return self._tunnels[label] def unregister_tunnel(self, label): try: del self._tunnels[label] except KeyError: pass def wall(self, sender, message): """ Send a wall to the client. """ self.conn.call('wall', sender, message) class MultiCliClient(CliClient): """ A bootstrap client connected to the cc-server. """ ROLE = 'mcli' KILL_ALREADY_CONNECTED = False def _get_tql_object(self): tql_object = SObject(self.login) tql_object.register(StaticTag('r', self.role)) tql_object.register(StaticTag('a', self._login)) self._server.db.register(tql_object) return tql_object @property def login(self): return '%s.%s' % (self._login, self.conn.get_fd()) @property def role(self): return 'cli' def shutdown(self): super(MultiCliClient, self).shutdown() # Also, remote the object from the db: self._server.db.unregister(self.login) Client.register_client_class(MultiCliClient) Client.register_client_class(CliClient)