+def JobQueueUpdate(file_name, content):
+ """Updates a file in the queue directory.
+
+ This is just a wrapper over L{utils.WriteFile}, with proper
+ checking.
+
+ @type file_name: str
+ @param file_name: the job file name
+ @type content: str
+ @param content: the new job contents
+ @rtype: boolean
+ @return: the success of the operation
+
+ """
+ if not _IsJobQueueFile(file_name):
+ return False
+
+ # Write and replace the file atomically
+ utils.WriteFile(file_name, data=_Decompress(content))
+
+ return True
+
+
+def JobQueueRename(old, new):
+ """Renames a job queue file.
+
+ This is just a wrapper over os.rename with proper checking.
+
+ @type old: str
+ @param old: the old (actual) file name
+ @type new: str
+ @param new: the desired file name
+ @rtype: boolean
+ @return: the success of the operation
+
+ """
+ if not (_IsJobQueueFile(old) and _IsJobQueueFile(new)):
+ return False
+
+ utils.RenameFile(old, new, mkdir=True)
+
+ return True
+
+
+def JobQueueSetDrainFlag(drain_flag):
+ """Set the drain flag for the queue.
+
+ This will set or unset the queue drain flag.
+
+ @type drain_flag: boolean
+ @param drain_flag: if True, will set the drain flag, otherwise reset it.
+ @rtype: boolean
+ @return: always True
+ @warning: the function always returns True
+
+ """
+ if drain_flag:
+ utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
+ else:
+ utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
+
+ return True
+
+
+def BlockdevClose(instance_name, disks):
+ """Closes the given block devices.
+
+ This means they will be switched to secondary mode (in case of
+ DRBD).
+
+ @param instance_name: if the argument is not empty, the symlinks
+ of this instance will be removed
+ @type disks: list of L{objects.Disk}
+ @param disks: the list of disks to be closed
+ @rtype: tuple (success, message)
+ @return: a tuple of success and message, where success
+ indicates the succes of the operation, and message
+ which will contain the error details in case we
+ failed
+
+ """
+ bdevs = []
+ for cf in disks:
+ rd = _RecursiveFindBD(cf)
+ if rd is None:
+ return (False, "Can't find device %s" % cf)
+ bdevs.append(rd)
+
+ msg = []
+ for rd in bdevs:
+ try:
+ rd.Close()
+ except errors.BlockDeviceError, err:
+ msg.append(str(err))
+ if msg:
+ return (False, "Can't make devices secondary: %s" % ",".join(msg))
+ else:
+ if instance_name:
+ _RemoveBlockDevLinks(instance_name, disks)
+ return (True, "All devices secondary")
+
+
+def ValidateHVParams(hvname, hvparams):
+ """Validates the given hypervisor parameters.
+
+ @type hvname: string
+ @param hvname: the hypervisor name
+ @type hvparams: dict
+ @param hvparams: the hypervisor parameters to be validated
+ @rtype: tuple (success, message)
+ @return: a tuple of success and message, where success
+ indicates the succes of the operation, and message
+ which will contain the error details in case we
+ failed
+
+ """
+ try:
+ hv_type = hypervisor.GetHypervisor(hvname)
+ hv_type.ValidateParameters(hvparams)
+ return (True, "Validation passed")
+ except errors.HypervisorError, err:
+ return (False, str(err))
+
+
+def DemoteFromMC():
+ """Demotes the current node from master candidate role.
+
+ """
+ # try to ensure we're not the master by mistake
+ master, myself = ssconf.GetMasterAndMyself()
+ if master == myself:
+ return (False, "ssconf status shows I'm the master node, will not demote")
+ pid_file = utils.DaemonPidFileName(constants.MASTERD_PID)
+ if utils.IsProcessAlive(utils.ReadPidFile(pid_file)):
+ return (False, "The master daemon is running, will not demote")
+ try:
+ utils.CreateBackup(constants.CLUSTER_CONF_FILE)
+ except EnvironmentError, err:
+ if err.errno != errno.ENOENT:
+ return (False, "Error while backing up cluster file: %s" % str(err))
+ utils.RemoveFile(constants.CLUSTER_CONF_FILE)
+ return (True, "Done")
+
+
+def _FindDisks(nodes_ip, disks):
+ """Sets the physical ID on disks and returns the block devices.
+
+ """
+ # set the correct physical ID
+ my_name = utils.HostInfo().name
+ for cf in disks:
+ cf.SetPhysicalID(my_name, nodes_ip)
+
+ bdevs = []
+
+ for cf in disks:
+ rd = _RecursiveFindBD(cf)
+ if rd is None:
+ return (False, "Can't find device %s" % cf)
+ bdevs.append(rd)
+ return (True, bdevs)
+
+
+def DrbdDisconnectNet(nodes_ip, disks):
+ """Disconnects the network on a list of drbd devices.
+
+ """
+ status, bdevs = _FindDisks(nodes_ip, disks)
+ if not status:
+ return status, bdevs
+
+ # disconnect disks
+ for rd in bdevs:
+ try:
+ rd.DisconnectNet()
+ except errors.BlockDeviceError, err:
+ logging.exception("Failed to go into standalone mode")
+ return (False, "Can't change network configuration: %s" % str(err))
+ return (True, "All disks are now disconnected")
+
+
+def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
+ """Attaches the network on a list of drbd devices.
+
+ """
+ status, bdevs = _FindDisks(nodes_ip, disks)
+ if not status:
+ return status, bdevs
+
+ if multimaster:
+ for idx, rd in enumerate(bdevs):
+ try:
+ _SymlinkBlockDev(instance_name, rd.dev_path, idx)
+ except EnvironmentError, err:
+ return (False, "Can't create symlink: %s" % str(err))
+ # reconnect disks, switch to new master configuration and if
+ # needed primary mode
+ for rd in bdevs:
+ try:
+ rd.AttachNet(multimaster)
+ except errors.BlockDeviceError, err:
+ return (False, "Can't change network configuration: %s" % str(err))
+ # wait until the disks are connected; we need to retry the re-attach
+ # if the device becomes standalone, as this might happen if the one
+ # node disconnects and reconnects in a different mode before the
+ # other node reconnects; in this case, one or both of the nodes will
+ # decide it has wrong configuration and switch to standalone
+ RECONNECT_TIMEOUT = 2 * 60
+ sleep_time = 0.100 # start with 100 miliseconds
+ timeout_limit = time.time() + RECONNECT_TIMEOUT
+ while time.time() < timeout_limit:
+ all_connected = True
+ for rd in bdevs:
+ stats = rd.GetProcStatus()
+ if not (stats.is_connected or stats.is_in_resync):
+ all_connected = False
+ if stats.is_standalone:
+ # peer had different config info and this node became
+ # standalone, even though this should not happen with the
+ # new staged way of changing disk configs
+ try:
+ rd.ReAttachNet(multimaster)
+ except errors.BlockDeviceError, err:
+ return (False, "Can't change network configuration: %s" % str(err))
+ if all_connected:
+ break
+ time.sleep(sleep_time)
+ sleep_time = min(5, sleep_time * 1.5)
+ if not all_connected:
+ return (False, "Timeout in disk reconnecting")
+ if multimaster:
+ # change to primary mode
+ for rd in bdevs:
+ try:
+ rd.Open()
+ except errors.BlockDeviceError, err:
+ return (False, "Can't change to primary mode: %s" % str(err))
+ if multimaster:
+ msg = "multi-master and primary"
+ else:
+ msg = "single-master"
+ return (True, "Disks are now configured as %s" % msg)
+
+
+def DrbdWaitSync(nodes_ip, disks):
+ """Wait until DRBDs have synchronized.
+
+ """
+ status, bdevs = _FindDisks(nodes_ip, disks)
+ if not status:
+ return status, bdevs
+
+ min_resync = 100
+ alldone = True
+ failure = False
+ for rd in bdevs:
+ stats = rd.GetProcStatus()
+ if not (stats.is_connected or stats.is_in_resync):
+ failure = True
+ break
+ alldone = alldone and (not stats.is_in_resync)
+ if stats.sync_percent is not None:
+ min_resync = min(min_resync, stats.sync_percent)
+ return (not failure, (alldone, min_resync))
+
+
+def PowercycleNode(hypervisor_type):
+ """Hard-powercycle the node.
+
+ Because we need to return first, and schedule the powercycle in the
+ background, we won't be able to report failures nicely.
+
+ """
+ hyper = hypervisor.GetHypervisor(hypervisor_type)
+ try:
+ pid = os.fork()
+ except OSError, err:
+ # if we can't fork, we'll pretend that we're in the child process
+ pid = 0
+ if pid > 0:
+ return (True, "Reboot scheduled in 5 seconds")
+ time.sleep(5)
+ hyper.PowercycleNode()
+
+