from ganeti import utils
from ganeti import errors
from ganeti import hypervisor
+from ganeti import locking
from ganeti import config
from ganeti import constants
from ganeti import objects
"""Logical Unit base class.
Subclasses must follow these rules:
- - implement CheckPrereq which also fills in the opcode instance
- with all the fields (even if as None)
+ - implement ExpandNames
+ - implement CheckPrereq
- implement Exec
- implement BuildHooksEnv
- redefine HPATH and HTYPE
- optionally redefine their run requirements:
REQ_MASTER: the LU needs to run on the master node
REQ_WSSTORE: the LU needs a writable SimpleStore
+ REQ_BGL: the LU needs to hold the Big Ganeti Lock exclusively
Note that all commands require root permissions.
_OP_REQP = []
REQ_MASTER = True
REQ_WSSTORE = False
+ REQ_BGL = True
- def __init__(self, processor, op, cfg, sstore):
+ def __init__(self, processor, op, context, sstore):
"""Constructor for LogicalUnit.
This needs to be overriden in derived classes in order to check op
"""
self.proc = processor
self.op = op
- self.cfg = cfg
+ self.cfg = context.cfg
self.sstore = sstore
+ self.context = context
+ self.needed_locks = None
self.__ssh = None
for attr_name in self._OP_REQP:
raise errors.OpPrereqError("Required parameter '%s' missing" %
attr_name)
- if not cfg.IsCluster():
+ if not self.cfg.IsCluster():
raise errors.OpPrereqError("Cluster not initialized yet,"
" use 'gnt-cluster init' first.")
if self.REQ_MASTER:
ssh = property(fget=__GetSSH)
+ def ExpandNames(self):
+ """Expand names for this LU.
+
+ This method is called before starting to execute the opcode, and it should
+ update all the parameters of the opcode to their canonical form (e.g. a
+ short node name must be fully expanded after this method has successfully
+ completed). This way locking, hooks, logging, ecc. can work correctly.
+
+ LUs which implement this method must also populate the self.needed_locks
+ member, as a dict with lock levels as keys, and a list of needed lock names
+ as values. Rules:
+ - Use an empty dict if you don't need any lock
+ - If you don't need any lock at a particular level omit that level
+ - Don't put anything for the BGL level
+ - If you want all locks at a level use None as a value
+ (this reflects what LockSet does, and will be replaced before
+ CheckPrereq with the full list of nodes that have been locked)
+
+ Examples:
+ # Acquire all nodes and one instance
+ self.needed_locks = {
+ locking.LEVEL_NODE: None,
+ locking.LEVEL_INSTANCES: ['instance1.example.tld'],
+ }
+ # Acquire just two nodes
+ self.needed_locks = {
+ locking.LEVEL_NODE: ['node1.example.tld', 'node2.example.tld'],
+ }
+ # Acquire no locks
+ self.needed_locks = {} # No, you can't leave it to the default value None
+
+ """
+ # The implementation of this method is mandatory only if the new LU is
+ # concurrent, so that old LUs don't need to be changed all at the same
+ # time.
+ if self.REQ_BGL:
+ self.needed_locks = {} # Exclusive LUs don't need locks.
+ else:
+ raise NotImplementedError
+
def CheckPrereq(self):
"""Check prerequisites for this LU.
not fulfilled. Its return value is ignored.
This method should also update all the parameters of the opcode to
- their canonical form; e.g. a short node name must be fully
- expanded after this method has successfully completed (so that
- hooks, logging, etc. work correctly).
+ their canonical form if it hasn't been done by ExpandNames before.
"""
raise NotImplementedError
raise errors.OpPrereqError("Neither the name nor the IP address of the"
" cluster has changed")
if new_ip != old_ip:
- result = utils.RunCmd(["fping", "-q", new_ip])
- if not result.failed:
+ if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
raise errors.OpPrereqError("The given cluster IP address (%s) is"
" reachable on the network. Aborting." %
new_ip)
if done or oneshot:
break
- if unlock:
- #utils.Unlock('cmd')
- pass
- try:
- time.sleep(min(60, max_time))
- finally:
- if unlock:
- #utils.Lock('cmd')
- pass
+ time.sleep(min(60, max_time))
if done:
proc.LogInfo("Instance %s's disks are in sync." % instance.name)
"""Build hooks env.
This doesn't run on the target node in the pre phase as a failed
- node would not allows itself to run.
+ node would then be impossible to remove.
"""
env = {
rpc.call_node_leave_cluster(node.name)
- self.ssh.Run(node.name, 'root', "%s stop" % constants.NODE_INITD_SCRIPT)
-
logger.Info("Removing node %s from config" % node.name)
self.cfg.RemoveNode(node.name)
+ # Remove the node from the Ganeti Lock Manager
+ self.context.glm.remove(locking.LEVEL_NODE, node.name)
utils.RemoveHostFromEtcHosts(node.name)
primary_ip=primary_ip,
secondary_ip=secondary_ip)
- if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31:
- if not os.path.exists(constants.VNC_PASSWORD_FILE):
- raise errors.OpPrereqError("Cluster VNC password file %s missing" %
- constants.VNC_PASSWORD_FILE)
-
def Exec(self, feedback_fn):
"""Adds the new node to the cluster.
new_node = self.new_node
node = new_node.name
- # set up inter-node password and certificate and restarts the node daemon
- gntpass = self.sstore.GetNodeDaemonPassword()
- if not re.match('^[a-zA-Z0-9.]{1,64}$', gntpass):
- raise errors.OpExecError("ganeti password corruption detected")
- f = open(constants.SSL_CERT_FILE)
- try:
- gntpem = f.read(8192)
- finally:
- f.close()
- # in the base64 pem encoding, neither '!' nor '.' are valid chars,
- # so we use this to detect an invalid certificate; as long as the
- # cert doesn't contain this, the here-document will be correctly
- # parsed by the shell sequence below
- if re.search('^!EOF\.', gntpem, re.MULTILINE):
- raise errors.OpExecError("invalid PEM encoding in the SSL certificate")
- if not gntpem.endswith("\n"):
- raise errors.OpExecError("PEM must end with newline")
- logger.Info("copy cluster pass to %s and starting the node daemon" % node)
-
- # and then connect with ssh to set password and start ganeti-noded
- # note that all the below variables are sanitized at this point,
- # either by being constants or by the checks above
- ss = self.sstore
- mycommand = ("umask 077 && "
- "echo '%s' > '%s' && "
- "cat > '%s' << '!EOF.' && \n"
- "%s!EOF.\n%s restart" %
- (gntpass, ss.KeyToFilename(ss.SS_NODED_PASS),
- constants.SSL_CERT_FILE, gntpem,
- constants.NODE_INITD_SCRIPT))
-
- result = self.ssh.Run(node, 'root', mycommand, batch=False, ask_key=True)
- if result.failed:
- raise errors.OpExecError("Remote command on node %s, error: %s,"
- " output: %s" %
- (node, result.fail_reason, result.output))
-
# check connectivity
- time.sleep(4)
-
result = rpc.call_version([node])[node]
if result:
if constants.PROTOCOL_VERSION == result:
" you gave (%s). Please fix and re-run this"
" command." % new_node.secondary_ip)
- success, msg = self.ssh.VerifyNodeHostname(node)
- if not success:
- raise errors.OpExecError("Node '%s' claims it has a different hostname"
- " than the one the resolver gives: %s."
- " Please fix and re-run this command." %
- (node, msg))
+ node_verify_list = [self.sstore.GetMasterNode()]
+ node_verify_param = {
+ 'nodelist': [node],
+ # TODO: do a node-net-test as well?
+ }
+
+ result = rpc.call_node_verify(node_verify_list, node_verify_param)
+ for verifier in node_verify_list:
+ if not result[verifier]:
+ raise errors.OpExecError("Cannot communicate with %s's node daemon"
+ " for remote verification" % verifier)
+ if result[verifier]['nodelist']:
+ for failed in result[verifier]['nodelist']:
+ feedback_fn("ssh/hostname verification failed %s -> %s" %
+ (verifier, result[verifier]['nodelist'][failed]))
+ raise errors.OpExecError("ssh/hostname verification failed.")
# Distribute updated /etc/hosts and known_hosts to all nodes,
# including the node just added
logger.Error("copy of file %s to node %s failed" %
(fname, to_node))
- to_copy = ss.GetFileList()
+ to_copy = self.sstore.GetFileList()
if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31:
to_copy.append(constants.VNC_PASSWORD_FILE)
for fname in to_copy:
- if not self.ssh.CopyFileToNode(node, fname):
+ result = rpc.call_upload_file([node], fname)
+ if not result[node]:
logger.Error("could not copy file %s to node %s" % (fname, node))
if not self.op.readd:
logger.Info("adding node %s to cluster.conf" % node)
self.cfg.AddNode(new_node)
+ # Add the new node to the Ganeti Lock Manager
+ self.context.glm.add(locking.LEVEL_NODE, node)
class LUMasterFailover(LogicalUnit):
return result
-class LUClusterCopyFile(NoHooksLU):
- """Copy file to cluster.
-
- """
- _OP_REQP = ["nodes", "filename"]
-
- def CheckPrereq(self):
- """Check prerequisites.
-
- It should check that the named file exists and that the given list
- of nodes is valid.
-
- """
- if not os.path.exists(self.op.filename):
- raise errors.OpPrereqError("No such filename '%s'" % self.op.filename)
-
- self.nodes = _GetWantedNodes(self, self.op.nodes)
-
- def Exec(self, feedback_fn):
- """Copy a file from master to some nodes.
-
- Args:
- opts - class with options as members
- args - list containing a single element, the file name
- Opts used:
- nodes - list containing the name of target nodes; if empty, all nodes
-
- """
- filename = self.op.filename
-
- myname = utils.HostInfo().name
-
- for node in self.nodes:
- if node == myname:
- continue
- if not self.ssh.CopyFileToNode(node, filename):
- logger.Error("Copy of file %s to node %s failed" % (filename, node))
-
-
class LUDumpClusterConfig(NoHooksLU):
"""Return a text-representation of the cluster-config.
return self.cfg.DumpConfig()
-class LURunClusterCommand(NoHooksLU):
- """Run a command on some nodes.
-
- """
- _OP_REQP = ["command", "nodes"]
-
- def CheckPrereq(self):
- """Check prerequisites.
-
- It checks that the given list of nodes is valid.
-
- """
- self.nodes = _GetWantedNodes(self, self.op.nodes)
-
- def Exec(self, feedback_fn):
- """Run a command on some nodes.
-
- """
- # put the master at the end of the nodes list
- master_node = self.sstore.GetMasterNode()
- if master_node in self.nodes:
- self.nodes.remove(master_node)
- self.nodes.append(master_node)
-
- data = []
- for node in self.nodes:
- result = self.ssh.Run(node, "root", self.op.command)
- data.append((node, result.output, result.exit_code))
-
- return data
-
-
class LUActivateInstanceDisks(NoHooksLU):
"""Bring up an instance's disks.
new_name)
if not getattr(self.op, "ignore_ip", False):
- command = ["fping", "-q", name_info.ip]
- result = utils.RunCmd(command)
- if not result.failed:
+ if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
raise errors.OpPrereqError("IP %s of instance %s already in use" %
(name_info.ip, new_name))
logger.Info("removing instance %s out of cluster config" % instance.name)
self.cfg.RemoveInstance(instance.name)
+ # Remove the new instance from the Ganeti Lock Manager
+ self.context.glm.remove(locking.LEVEL_INSTANCE, instance.name)
class LUQueryInstances(NoHooksLU):
return results
-def _GenerateMDDRBDBranch(cfg, primary, secondary, size, names):
- """Generate a drbd device complete with its children.
-
- """
- port = cfg.AllocatePort()
- vgname = cfg.GetVGName()
- dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
- logical_id=(vgname, names[0]))
- dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
- logical_id=(vgname, names[1]))
- drbd_dev = objects.Disk(dev_type=constants.LD_DRBD7, size=size,
- logical_id = (primary, secondary, port),
- children = [dev_data, dev_meta])
- return drbd_dev
-
-
def _GenerateDRBD8Branch(cfg, primary, secondary, size, names, iv_name):
"""Generate a drbd8 device complete with its children.
feedback_fn("adding instance %s to cluster config" % instance)
self.cfg.AddInstance(iobj)
+ # Add the new instance to the Ganeti Lock Manager
+ self.context.glm.add(locking.LEVEL_INSTANCE, instance)
if self.op.wait_for_sync:
disk_abort = not _WaitForSync(self.cfg, iobj, self.proc)
if disk_abort:
_RemoveDisks(iobj, self.cfg)
self.cfg.RemoveInstance(iobj.name)
+ # Remove the new instance from the Ganeti Lock Manager
+ self.context.glm.remove(locking.LEVEL_INSTANCE, iobj.name)
raise errors.OpExecError("There are some degraded disks for"
" this instance")
class LUTestDelay(NoHooksLU):
"""Sleep for a specified amount of time.
- This LU sleeps on the master and/or nodes for a specified amoutn of
+ This LU sleeps on the master and/or nodes for a specified amount of
time.
"""
_OP_REQP = ["duration", "on_master", "on_nodes"]
+ REQ_BGL = False
- def CheckPrereq(self):
- """Check prerequisites.
+ def ExpandNames(self):
+ """Expand names and set required locks.
- This checks that we have a good list of nodes and/or the duration
- is valid.
+ This expands the node list, if any.
"""
-
+ self.needed_locks = {}
if self.op.on_nodes:
+ # _GetWantedNodes can be used here, but is not always appropriate to use
+ # this way in ExpandNames. Check LogicalUnit.ExpandNames docstring for
+ # more information.
self.op.on_nodes = _GetWantedNodes(self, self.op.on_nodes)
+ self.needed_locks[locking.LEVEL_NODE] = self.op.on_nodes
+
+ def CheckPrereq(self):
+ """Check prerequisites.
+
+ """
def Exec(self, feedback_fn):
"""Do the actual sleep.
projects / ganeti-local / blobdiff