X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/cb92e7a14b7ee510dcc26c2a14dddaa0fa8f10e4..3b7218422467e12322b6b628f3bd71d69eb9337a:/lib/client/gnt_node.py diff --git a/lib/client/gnt_node.py b/lib/client/gnt_node.py index 0fe37c7..e893b5e 100644 --- a/lib/client/gnt_node.py +++ b/lib/client/gnt_node.py @@ -1,7 +1,7 @@ # # -# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 Google Inc. +# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012 Google Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by @@ -27,6 +27,7 @@ # C0103: Invalid name gnt-node import itertools +import errno from ganeti.cli import * from ganeti import cli @@ -36,8 +37,13 @@ from ganeti import utils from ganeti import constants from ganeti import errors from ganeti import netutils +from ganeti import pathutils +from ganeti import ssh +from ganeti import compat from cStringIO import StringIO +from ganeti import confd +from ganeti.confd import client as confd_client #: default list of field for L{ListNodes} _LIST_DEF_FIELDS = [ @@ -101,10 +107,12 @@ _REPAIRABLE_STORAGE_TYPES = \ _MODIFIABLE_STORAGE_TYPES = constants.MODIFIABLE_STORAGE_FIELDS.keys() +_OOB_COMMAND_ASK = compat.UniqueFrozenset([ + constants.OOB_POWER_OFF, + constants.OOB_POWER_CYCLE, + ]) -_OOB_COMMAND_ASK = frozenset([constants.OOB_POWER_OFF, - constants.OOB_POWER_CYCLE]) - +_ENV_OVERRIDE = compat.UniqueFrozenset(["list"]) NONODE_SETUP_OPT = cli_option("--no-node-setup", default=True, action="store_false", dest="node_setup", @@ -128,34 +136,86 @@ def ConvertStorageType(user_storage_type): errors.ECODE_INVAL) -def _RunSetupSSH(options, nodes): - """Wrapper around utils.RunCmd to call setup-ssh +def _TryReadFile(path): + """Tries to read a file. + + If the file is not found, C{None} is returned. + + @type path: string + @param path: Filename + @rtype: None or string + @todo: Consider adding a generic ENOENT wrapper + + """ + try: + return utils.ReadFile(path) + except EnvironmentError, err: + if err.errno == errno.ENOENT: + return None + else: + raise + + +def _ReadSshKeys(keyfiles, _tostderr_fn=ToStderr): + """Reads SSH keys according to C{keyfiles}. + + @type keyfiles: dict + @param keyfiles: Dictionary with keys of L{constants.SSHK_ALL} and two-values + tuples (private and public key file) + @rtype: list + @return: List of three-values tuples (L{constants.SSHK_ALL}, private and + public key as strings) + + """ + result = [] + + for (kind, (private_file, public_file)) in keyfiles.items(): + private_key = _TryReadFile(private_file) + public_key = _TryReadFile(public_file) + + if public_key and private_key: + result.append((kind, private_key, public_key)) + elif public_key or private_key: + _tostderr_fn("Couldn't find a complete set of keys for kind '%s'; files" + " '%s' and '%s'", kind, private_file, public_file) + + return result + + +def _SetupSSH(options, cluster_name, node): + """Configures a destination node's SSH daemon. - @param options: The command line options - @param nodes: The nodes to setup + @param options: Command line options + @type cluster_name + @param cluster_name: Cluster name + @type node: string + @param node: Destination node name """ - cmd = [constants.SETUP_SSH] - - # Pass --debug|--verbose to the external script if set on our invocation - # --debug overrides --verbose - if options.debug: - cmd.append("--debug") - elif options.verbose: - cmd.append("--verbose") - if not options.ssh_key_check: - cmd.append("--no-ssh-key-check") if options.force_join: - cmd.append("--force-join") + ToStderr("The \"--force-join\" option is no longer supported and will be" + " ignored.") - cmd.extend(nodes) + host_keys = _ReadSshKeys(constants.SSH_DAEMON_KEYFILES) - result = utils.RunCmd(cmd, interactive=True) + (_, root_keyfiles) = \ + ssh.GetAllUserFiles(constants.SSH_LOGIN_USER, mkdir=False, dircheck=False) - if result.failed: - errmsg = ("Command '%s' failed with exit code %s; output %r" % - (result.cmd, result.exit_code, result.output)) - raise errors.OpExecError(errmsg) + root_keys = _ReadSshKeys(root_keyfiles) + + (_, cert_pem) = \ + utils.ExtractX509Certificate(utils.ReadFile(pathutils.NODED_CERT_FILE)) + + data = { + constants.SSHS_CLUSTER_NAME: cluster_name, + constants.SSHS_NODE_DAEMON_CERTIFICATE: cert_pem, + constants.SSHS_SSH_HOST_KEY: host_keys, + constants.SSHS_SSH_ROOT_KEY: root_keys, + } + + bootstrap.RunNodeSetupCmd(cluster_name, node, pathutils.PREPARE_NODE_JOIN, + options.debug, options.verbose, False, + options.ssh_key_check, options.ssh_key_check, data) @UsesRPC @@ -197,8 +257,7 @@ def AddNode(opts, args): sip = opts.secondary_ip # read the cluster name from the master - output = cl.QueryConfigValues(["cluster_name"]) - cluster_name = output[0] + (cluster_name, ) = cl.QueryConfigValues(["cluster_name"]) if not readd and opts.node_setup: ToStderr("-- WARNING -- \n" @@ -209,14 +268,23 @@ def AddNode(opts, args): "and grant full intra-cluster ssh root access to/from it\n", node) if opts.node_setup: - _RunSetupSSH(opts, [node]) + _SetupSSH(opts, cluster_name, node) - bootstrap.SetupNodeDaemon(cluster_name, node, opts.ssh_key_check) + bootstrap.SetupNodeDaemon(opts, cluster_name, node) + + if opts.disk_state: + disk_state = utils.FlatToDict(opts.disk_state) + else: + disk_state = {} + + hv_state = dict(opts.hv_state) op = opcodes.OpNodeAdd(node_name=args[0], secondary_ip=sip, readd=opts.readd, group=opts.nodegroup, vm_capable=opts.vm_capable, ndparams=opts.ndparams, - master_capable=opts.master_capable) + master_capable=opts.master_capable, + disk_state=disk_state, + hv_state=hv_state) SubmitOpCode(op, opts=opts) @@ -235,10 +303,12 @@ def ListNodes(opts, args): fmtoverride = dict.fromkeys(["pinst_list", "sinst_list", "tags"], (",".join, False)) + cl = GetClient(query=True) + return GenericList(constants.QR_NODE, selected_fields, args, opts.units, opts.separator, not opts.no_headers, format_override=fmtoverride, verbose=opts.verbose, - force_filter=opts.force_filter) + force_filter=opts.force_filter, cl=cl) def ListNodeFields(opts, args): @@ -251,8 +321,10 @@ def ListNodeFields(opts, args): @return: the desired exit code """ + cl = GetClient(query=True) + return GenericListFields(constants.QR_NODE, args, opts.separator, - not opts.no_headers) + not opts.no_headers, cl=cl) def EvacuateNode(opts, args): @@ -292,7 +364,10 @@ def EvacuateNode(opts, args): cl = GetClient() - result = cl.QueryNodes(names=args, fields=fields, use_locking=False) + qcl = GetClient(query=True) + result = qcl.QueryNodes(names=args, fields=fields, use_locking=False) + qcl.Close() + instances = set(itertools.chain(*itertools.chain(*itertools.chain(result)))) if not instances: @@ -312,7 +387,7 @@ def EvacuateNode(opts, args): remote_node=opts.dst_node, iallocator=opts.iallocator, early_release=opts.early_release) - result = SubmitOpCode(op, cl=cl, opts=opts) + result = SubmitOrSend(op, opts, cl=cl) # Keep track of submitted jobs jex = JobExecutor(cl=cl, opts=opts) @@ -348,8 +423,10 @@ def FailoverNode(opts, args): # these fields are static data anyway, so it doesn't matter, but # locking=True should be safer + qcl = GetClient(query=True) result = cl.QueryNodes(names=args, fields=selected_fields, use_locking=False) + qcl.Close() node, pinst = result[0] if not pinst: @@ -388,7 +465,9 @@ def MigrateNode(opts, args): force = opts.force selected_fields = ["name", "pinst_list"] + qcl = GetClient(query=True) result = cl.QueryNodes(names=args, fields=selected_fields, use_locking=False) + qcl.Close() ((node, pinst), ) = result if not pinst: @@ -414,9 +493,11 @@ def MigrateNode(opts, args): op = opcodes.OpNodeMigrate(node_name=args[0], mode=mode, iallocator=opts.iallocator, - target_node=opts.dst_node) + target_node=opts.dst_node, + allow_runtime_changes=opts.allow_runtime_chgs, + ignore_ipolicy=opts.ignore_ipolicy) - result = SubmitOpCode(op, cl=cl, opts=opts) + result = SubmitOrSend(op, opts, cl=cl) # Keep track of submitted jobs jex = JobExecutor(cl=cl, opts=opts) @@ -448,7 +529,7 @@ def ShowNodeConfig(opts, args): @return: the desired exit code """ - cl = GetClient() + cl = GetClient(query=True) result = cl.QueryNodes(fields=["name", "pip", "sip", "pinst_list", "sinst_list", "master_candidate", "drained", "offline", @@ -523,7 +604,7 @@ def PowercycleNode(opts, args): return 2 op = opcodes.OpNodePowercycle(node_name=node, force=opts.force) - result = SubmitOpCode(op, opts=opts) + result = SubmitOrSend(op, opts) if result: ToStderr(result) return 0 @@ -785,7 +866,7 @@ def ModifyStorage(opts, args): storage_type=storage_type, name=volume_name, changes=changes) - SubmitOpCode(op, opts=opts) + SubmitOrSend(op, opts) else: ToStderr("No changes to perform, exiting.") @@ -808,7 +889,7 @@ def RepairStorage(opts, args): storage_type=storage_type, name=volume_name, ignore_consistency=opts.ignore_consistency) - SubmitOpCode(op, opts=opts) + SubmitOrSend(op, opts) def SetNodeParams(opts, args): @@ -824,10 +905,18 @@ def SetNodeParams(opts, args): all_changes = [opts.master_candidate, opts.drained, opts.offline, opts.master_capable, opts.vm_capable, opts.secondary_ip, opts.ndparams] - if all_changes.count(None) == len(all_changes): + if (all_changes.count(None) == len(all_changes) and + not (opts.hv_state or opts.disk_state)): ToStderr("Please give at least one of the parameters.") return 1 + if opts.disk_state: + disk_state = utils.FlatToDict(opts.disk_state) + else: + disk_state = {} + + hv_state = dict(opts.hv_state) + op = opcodes.OpNodeSetParams(node_name=args[0], master_candidate=opts.master_candidate, offline=opts.offline, @@ -838,7 +927,9 @@ def SetNodeParams(opts, args): force=opts.force, ndparams=opts.ndparams, auto_promote=opts.auto_promote, - powered=opts.node_powered) + powered=opts.node_powered, + hv_state=hv_state, + disk_state=disk_state) # even if here we process the result, we allow submit only result = SubmitOrSend(op, opts) @@ -850,12 +941,145 @@ def SetNodeParams(opts, args): return 0 +def RestrictedCommand(opts, args): + """Runs a remote command on node(s). + + @param opts: Command line options selected by user + @type args: list + @param args: Command line arguments + @rtype: int + @return: Exit code + + """ + cl = GetClient() + + if len(args) > 1 or opts.nodegroup: + # Expand node names + nodes = GetOnlineNodes(nodes=args[1:], cl=cl, nodegroup=opts.nodegroup) + else: + raise errors.OpPrereqError("Node group or node names must be given", + errors.ECODE_INVAL) + + op = opcodes.OpRestrictedCommand(command=args[0], nodes=nodes, + use_locking=opts.do_locking) + result = SubmitOrSend(op, opts, cl=cl) + + exit_code = constants.EXIT_SUCCESS + + for (node, (status, text)) in zip(nodes, result): + ToStdout("------------------------------------------------") + if status: + if opts.show_machine_names: + for line in text.splitlines(): + ToStdout("%s: %s", node, line) + else: + ToStdout("Node: %s", node) + ToStdout(text) + else: + exit_code = constants.EXIT_FAILURE + ToStdout(text) + + return exit_code + + +class ReplyStatus(object): + """Class holding a reply status for synchronous confd clients. + + """ + def __init__(self): + self.failure = True + self.answer = False + + +def ListDrbd(opts, args): + """Modifies a node. + + @param opts: the command line options selected by the user + @type args: list + @param args: should contain only one element, the node name + @rtype: int + @return: the desired exit code + + """ + if len(args) != 1: + ToStderr("Please give one (and only one) node.") + return constants.EXIT_FAILURE + + if not constants.ENABLE_CONFD: + ToStderr("Error: this command requires confd support, but it has not" + " been enabled at build time.") + return constants.EXIT_FAILURE + + status = ReplyStatus() + + def ListDrbdConfdCallback(reply): + """Callback for confd queries""" + if reply.type == confd_client.UPCALL_REPLY: + answer = reply.server_reply.answer + reqtype = reply.orig_request.type + if reqtype == constants.CONFD_REQ_NODE_DRBD: + if reply.server_reply.status != constants.CONFD_REPL_STATUS_OK: + ToStderr("Query gave non-ok status '%s': %s" % + (reply.server_reply.status, + reply.server_reply.answer)) + status.failure = True + return + if not confd.HTNodeDrbd(answer): + ToStderr("Invalid response from server: expected %s, got %s", + confd.HTNodeDrbd, answer) + status.failure = True + else: + status.failure = False + status.answer = answer + else: + ToStderr("Unexpected reply %s!?", reqtype) + status.failure = True + + node = args[0] + hmac = utils.ReadFile(pathutils.CONFD_HMAC_KEY) + filter_callback = confd_client.ConfdFilterCallback(ListDrbdConfdCallback) + counting_callback = confd_client.ConfdCountingCallback(filter_callback) + cf_client = confd_client.ConfdClient(hmac, [constants.IP4_ADDRESS_LOCALHOST], + counting_callback) + req = confd_client.ConfdClientRequest(type=constants.CONFD_REQ_NODE_DRBD, + query=node) + + def DoConfdRequestReply(req): + counting_callback.RegisterQuery(req.rsalt) + cf_client.SendRequest(req, async=False) + while not counting_callback.AllAnswered(): + if not cf_client.ReceiveReply(): + ToStderr("Did not receive all expected confd replies") + break + + DoConfdRequestReply(req) + + if status.failure: + return constants.EXIT_FAILURE + + fields = ["node", "minor", "instance", "disk", "role", "peer"] + if opts.no_headers: + headers = None + else: + headers = {"node": "Node", "minor": "Minor", "instance": "Instance", + "disk": "Disk", "role": "Role", "peer": "PeerNode"} + + data = GenerateTable(separator=opts.separator, headers=headers, + fields=fields, data=sorted(status.answer), + numfields=["minor"]) + for line in data: + ToStdout(line) + + return constants.EXIT_SUCCESS + + commands = { "add": ( AddNode, [ArgHost(min=1, max=1)], [SECONDARY_IP_OPT, READD_OPT, NOSSH_KEYCHECK_OPT, NODE_FORCE_JOIN_OPT, NONODE_SETUP_OPT, VERBOSE_OPT, NODEGROUP_OPT, PRIORITY_OPT, - CAPAB_MASTER_OPT, CAPAB_VM_OPT, NODE_PARAMS_OPT], + CAPAB_MASTER_OPT, CAPAB_VM_OPT, NODE_PARAMS_OPT, HV_STATE_OPT, + DISK_STATE_OPT], "[-s ip] [--readd] [--no-ssh-key-check] [--force-join]" " [--no-node-setup] [--verbose]" " ", @@ -863,10 +1087,9 @@ commands = { "evacuate": ( EvacuateNode, ARGS_ONE_NODE, [FORCE_OPT, IALLOCATOR_OPT, NEW_SECONDARY_OPT, EARLY_RELEASE_OPT, - PRIORITY_OPT, PRIMARY_ONLY_OPT, SECONDARY_ONLY_OPT], - "[-f] {-I | -n } ", - "Relocate the secondary instances from a node" - " to other nodes"), + PRIORITY_OPT, PRIMARY_ONLY_OPT, SECONDARY_ONLY_OPT, SUBMIT_OPT], + "[-f] {-I | -n } [-p | -s] [options...] ", + "Relocate the primary and/or secondary instances from a node"), "failover": ( FailoverNode, ARGS_ONE_NODE, [FORCE_OPT, IGNORE_CONSIST_OPT, IALLOCATOR_OPT, PRIORITY_OPT], @@ -876,7 +1099,8 @@ commands = { "migrate": ( MigrateNode, ARGS_ONE_NODE, [FORCE_OPT, NONLIVE_OPT, MIGRATION_MODE_OPT, DST_NODE_OPT, - IALLOCATOR_OPT, PRIORITY_OPT], + IALLOCATOR_OPT, PRIORITY_OPT, IGNORE_IPOLICY_OPT, + NORUNTIME_CHGS_OPT, SUBMIT_OPT], "[-f] ", "Migrate all the primary instance on a node away from it" " (only for instances of type drbd)"), @@ -902,11 +1126,11 @@ commands = { [FORCE_OPT, SUBMIT_OPT, MC_OPT, DRAINED_OPT, OFFLINE_OPT, CAPAB_MASTER_OPT, CAPAB_VM_OPT, SECONDARY_IP_OPT, AUTO_PROMOTE_OPT, DRY_RUN_OPT, PRIORITY_OPT, NODE_PARAMS_OPT, - NODE_POWERED_OPT], + NODE_POWERED_OPT, HV_STATE_OPT, DISK_STATE_OPT], "", "Alters the parameters of a node"), "powercycle": ( PowercycleNode, ARGS_ONE_NODE, - [FORCE_OPT, CONFIRM_OPT, DRY_RUN_OPT, PRIORITY_OPT], + [FORCE_OPT, CONFIRM_OPT, DRY_RUN_OPT, PRIORITY_OPT, SUBMIT_OPT], "", "Tries to forcefully powercycle a node"), "power": ( PowerNode, @@ -935,32 +1159,49 @@ commands = { [ArgNode(min=1, max=1), ArgChoice(min=1, max=1, choices=_MODIFIABLE_STORAGE_TYPES), ArgFile(min=1, max=1)], - [ALLOCATABLE_OPT, DRY_RUN_OPT, PRIORITY_OPT], + [ALLOCATABLE_OPT, DRY_RUN_OPT, PRIORITY_OPT, SUBMIT_OPT], " ", "Modify storage volume on a node"), "repair-storage": ( RepairStorage, [ArgNode(min=1, max=1), ArgChoice(min=1, max=1, choices=_REPAIRABLE_STORAGE_TYPES), ArgFile(min=1, max=1)], - [IGNORE_CONSIST_OPT, DRY_RUN_OPT, PRIORITY_OPT], + [IGNORE_CONSIST_OPT, DRY_RUN_OPT, PRIORITY_OPT, SUBMIT_OPT], " ", "Repairs a storage volume on a node"), "list-tags": ( ListTags, ARGS_ONE_NODE, [], "", "List the tags of the given node"), "add-tags": ( - AddTags, [ArgNode(min=1, max=1), ArgUnknown()], [TAG_SRC_OPT, PRIORITY_OPT], + AddTags, [ArgNode(min=1, max=1), ArgUnknown()], + [TAG_SRC_OPT, PRIORITY_OPT, SUBMIT_OPT], " tag...", "Add tags to the given node"), "remove-tags": ( RemoveTags, [ArgNode(min=1, max=1), ArgUnknown()], - [TAG_SRC_OPT, PRIORITY_OPT], + [TAG_SRC_OPT, PRIORITY_OPT, SUBMIT_OPT], " tag...", "Remove tags from the given node"), "health": ( Health, ARGS_MANY_NODES, - [NOHDR_OPT, SEP_OPT, SUBMIT_OPT, PRIORITY_OPT, OOB_TIMEOUT_OPT], + [NOHDR_OPT, SEP_OPT, PRIORITY_OPT, OOB_TIMEOUT_OPT], "[...]", "List health of node(s) using out-of-band"), + "list-drbd": ( + ListDrbd, ARGS_ONE_NODE, + [NOHDR_OPT, SEP_OPT], + "[]", "Query the list of used DRBD minors on the given node"), + "restricted-command": ( + RestrictedCommand, [ArgUnknown(min=1, max=1)] + ARGS_MANY_NODES, + [SYNC_OPT, PRIORITY_OPT, SUBMIT_OPT, SHOW_MACHINE_OPT, NODEGROUP_OPT], + " [...]", + "Executes a restricted command on node(s)"), + } + +#: dictionary with aliases for commands +aliases = { + "show": "info", } def Main(): - return GenericMain(commands, override={"tag_type": constants.TAG_NODE}) + return GenericMain(commands, aliases=aliases, + override={"tag_type": constants.TAG_NODE}, + env_override=_ENV_OVERRIDE)