| # |
| # |
| |
| # Copyright (C) 2006, 2007, 2011, 2012 Google Inc. |
| # All rights reserved. |
| # |
| # Redistribution and use in source and binary forms, with or without |
| # modification, are permitted provided that the following conditions are |
| # met: |
| # |
| # 1. Redistributions of source code must retain the above copyright notice, |
| # this list of conditions and the following disclaimer. |
| # |
| # 2. Redistributions in binary form must reproduce the above copyright |
| # notice, this list of conditions and the following disclaimer in the |
| # documentation and/or other materials provided with the distribution. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS |
| # IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED |
| # TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR |
| # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR |
| # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, |
| # EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, |
| # PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR |
| # PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF |
| # LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING |
| # NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS |
| # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| |
| |
| """Module implementing the logic behind the cluster operations |
| |
| This module implements the logic for doing operations in the cluster. There |
| are two kinds of classes defined: |
| - logical units, which know how to deal with their specific opcode only |
| - the processor, which dispatches the opcodes to their logical units |
| |
| """ |
| |
| import sys |
| import logging |
| import random |
| import time |
| import itertools |
| import traceback |
| |
| from ganeti import opcodes |
| from ganeti import opcodes_base |
| from ganeti import constants |
| from ganeti import errors |
| from ganeti import hooksmaster |
| from ganeti import cmdlib |
| from ganeti import locking |
| from ganeti import utils |
| from ganeti import wconfd |
| |
| |
| sighupReceived = [False] |
| lusExecuting = [0] |
| |
| _OP_PREFIX = "Op" |
| _LU_PREFIX = "LU" |
| |
| |
| class LockAcquireTimeout(Exception): |
| """Exception to report timeouts on acquiring locks. |
| |
| """ |
| |
| |
| def _CalculateLockAttemptTimeouts(): |
| """Calculate timeouts for lock attempts. |
| |
| """ |
| result = [constants.LOCK_ATTEMPTS_MINWAIT] |
| running_sum = result[0] |
| |
| # Wait for a total of at least LOCK_ATTEMPTS_TIMEOUT before doing a |
| # blocking acquire |
| while running_sum < constants.LOCK_ATTEMPTS_TIMEOUT: |
| timeout = (result[-1] * 1.05) ** 1.25 |
| |
| # Cap max timeout. This gives other jobs a chance to run even if |
| # we're still trying to get our locks, before finally moving to a |
| # blocking acquire. |
| timeout = min(timeout, constants.LOCK_ATTEMPTS_MAXWAIT) |
| # And also cap the lower boundary for safety |
| timeout = max(timeout, constants.LOCK_ATTEMPTS_MINWAIT) |
| |
| result.append(timeout) |
| running_sum += timeout |
| |
| return result |
| |
| |
| class LockAttemptTimeoutStrategy(object): |
| """Class with lock acquire timeout strategy. |
| |
| """ |
| __slots__ = [ |
| "_timeouts", |
| "_random_fn", |
| "_time_fn", |
| ] |
| |
| _TIMEOUT_PER_ATTEMPT = _CalculateLockAttemptTimeouts() |
| |
| def __init__(self, _time_fn=time.time, _random_fn=random.random): |
| """Initializes this class. |
| |
| @param _time_fn: Time function for unittests |
| @param _random_fn: Random number generator for unittests |
| |
| """ |
| object.__init__(self) |
| |
| self._timeouts = iter(self._TIMEOUT_PER_ATTEMPT) |
| self._time_fn = _time_fn |
| self._random_fn = _random_fn |
| |
| def NextAttempt(self): |
| """Returns the timeout for the next attempt. |
| |
| """ |
| try: |
| timeout = self._timeouts.next() |
| except StopIteration: |
| # No more timeouts, do blocking acquire |
| timeout = None |
| |
| if timeout is not None: |
| # Add a small variation (-/+ 5%) to timeout. This helps in situations |
| # where two or more jobs are fighting for the same lock(s). |
| variation_range = timeout * 0.1 |
| timeout += ((self._random_fn() * variation_range) - |
| (variation_range * 0.5)) |
| |
| return timeout |
| |
| |
| class OpExecCbBase(object): # pylint: disable=W0232 |
| """Base class for OpCode execution callbacks. |
| |
| """ |
| def NotifyStart(self): |
| """Called when we are about to execute the LU. |
| |
| This function is called when we're about to start the lu's Exec() method, |
| that is, after we have acquired all locks. |
| |
| """ |
| |
| def NotifyRetry(self): |
| """Called when we are about to reset an LU to retry again. |
| |
| This function is called after PrepareRetry successfully completed. |
| |
| """ |
| |
| # TODO: Cleanup calling conventions, make them explicit. |
| def Feedback(self, *args): |
| """Sends feedback from the LU code to the end-user. |
| |
| """ |
| |
| def CurrentPriority(self): # pylint: disable=R0201 |
| """Returns current priority or C{None}. |
| |
| """ |
| return None |
| |
| def SubmitManyJobs(self, jobs): |
| """Submits jobs for processing. |
| |
| See L{jqueue.JobQueue.SubmitManyJobs}. |
| |
| """ |
| raise NotImplementedError |
| |
| |
| def _LUNameForOpName(opname): |
| """Computes the LU name for a given OpCode name. |
| |
| """ |
| assert opname.startswith(_OP_PREFIX), \ |
| "Invalid OpCode name, doesn't start with %s: %s" % (_OP_PREFIX, opname) |
| |
| return _LU_PREFIX + opname[len(_OP_PREFIX):] |
| |
| |
| def _ComputeDispatchTable(): |
| """Computes the opcode-to-lu dispatch table. |
| |
| """ |
| return dict((op, getattr(cmdlib, _LUNameForOpName(op.__name__))) |
| for op in opcodes.OP_MAPPING.values() |
| if op.WITH_LU) |
| |
| |
| def _SetBaseOpParams(src, defcomment, dst): |
| """Copies basic opcode parameters. |
| |
| @type src: L{opcodes.OpCode} |
| @param src: Source opcode |
| @type defcomment: string |
| @param defcomment: Comment to specify if not already given |
| @type dst: L{opcodes.OpCode} |
| @param dst: Destination opcode |
| |
| """ |
| if hasattr(src, "debug_level"): |
| dst.debug_level = src.debug_level |
| |
| if (getattr(dst, "priority", None) is None and |
| hasattr(src, "priority")): |
| dst.priority = src.priority |
| |
| if not getattr(dst, opcodes_base.COMMENT_ATTR, None): |
| dst.comment = defcomment |
| |
| if hasattr(src, constants.OPCODE_REASON): |
| dst.reason = list(getattr(dst, constants.OPCODE_REASON, [])) |
| dst.reason.extend(getattr(src, constants.OPCODE_REASON, [])) |
| |
| |
| def _ProcessResult(submit_fn, op, result): |
| """Examines opcode result. |
| |
| If necessary, additional processing on the result is done. |
| |
| """ |
| if isinstance(result, cmdlib.ResultWithJobs): |
| # Copy basic parameters (e.g. priority) |
| for op2 in itertools.chain(*result.jobs): |
| _SetBaseOpParams(op, "Submitted by %s" % op.OP_ID, op2) |
| |
| # Submit jobs |
| job_submission = submit_fn(result.jobs) |
| |
| # Build dictionary |
| result = result.other |
| |
| assert constants.JOB_IDS_KEY not in result, \ |
| "Key '%s' found in additional return values" % constants.JOB_IDS_KEY |
| |
| result[constants.JOB_IDS_KEY] = job_submission |
| |
| return result |
| |
| |
| def _FailingSubmitManyJobs(_): |
| """Implementation of L{OpExecCbBase.SubmitManyJobs} to raise an exception. |
| |
| """ |
| raise errors.ProgrammerError("Opcodes processed without callbacks (e.g." |
| " queries) can not submit jobs") |
| |
| |
| def _LockList(names): |
| """If 'names' is a string, make it a single-element list. |
| |
| @type names: list or string or NoneType |
| @param names: Lock names |
| @rtype: a list of strings |
| @return: if 'names' argument is an iterable, a list of it; |
| if it's a string, make it a one-element list; |
| if L{locking.ALL_SET}, L{locking.ALL_SET} |
| |
| """ |
| if names == locking.ALL_SET: |
| return names |
| elif isinstance(names, basestring): |
| return [names] |
| else: |
| return list(names) |
| |
| |
| def _CheckSecretParameters(op): |
| """Check if secret parameters are expected, but missing. |
| |
| """ |
| if hasattr(op, "osparams_secret") and op.osparams_secret: |
| for secret_param in op.osparams_secret: |
| if op.osparams_secret[secret_param].Get() == constants.REDACTED: |
| raise errors.OpPrereqError("Please re-submit secret parameters to job.", |
| errors.ECODE_INVAL) |
| |
| |
| class Processor(object): |
| """Object which runs OpCodes""" |
| DISPATCH_TABLE = _ComputeDispatchTable() |
| |
| def __init__(self, context, ec_id, enable_locks=True): |
| """Constructor for Processor |
| |
| @type context: GanetiContext |
| @param context: global Ganeti context |
| @type ec_id: string |
| @param ec_id: execution context identifier |
| |
| """ |
| self._ec_id = ec_id |
| self._cbs = None |
| self.cfg = context.GetConfig(ec_id) |
| self.rpc = context.GetRpc(self.cfg) |
| self.hmclass = hooksmaster.HooksMaster |
| self._enable_locks = enable_locks |
| self.wconfd = wconfd # Indirection to allow testing |
| self._wconfdcontext = context.GetWConfdContext(ec_id) |
| |
| def _CheckLocksEnabled(self): |
| """Checks if locking is enabled. |
| |
| @raise errors.ProgrammerError: In case locking is not enabled |
| |
| """ |
| if not self._enable_locks: |
| raise errors.ProgrammerError("Attempted to use disabled locks") |
| |
| def _RequestAndWait(self, request, timeout): |
| """Request locks from WConfD and wait for them to be granted. |
| |
| @type request: list |
| @param request: the lock request to be sent to WConfD |
| @type timeout: float |
| @param timeout: the time to wait for the request to be granted |
| @raise LockAcquireTimeout: In case locks couldn't be acquired in specified |
| amount of time; in this case, locks still might be acquired or a request |
| pending. |
| |
| """ |
| logging.debug("Trying %ss to request %s for %s", |
| timeout, request, self._wconfdcontext) |
| if self._cbs: |
| priority = self._cbs.CurrentPriority() # pylint: disable=W0612 |
| else: |
| priority = None |
| |
| if priority is None: |
| priority = constants.OP_PRIO_DEFAULT |
| |
| ## Expect a signal |
| if sighupReceived[0]: |
| logging.warning("Ignoring unexpected SIGHUP") |
| sighupReceived[0] = False |
| |
| # Request locks |
| self.wconfd.Client().UpdateLocksWaiting(self._wconfdcontext, priority, |
| request) |
| pending = self.wconfd.Client().HasPendingRequest(self._wconfdcontext) |
| |
| if pending: |
| def _HasPending(): |
| if sighupReceived[0]: |
| return self.wconfd.Client().HasPendingRequest(self._wconfdcontext) |
| else: |
| return True |
| |
| pending = utils.SimpleRetry(False, _HasPending, 0.05, timeout) |
| |
| signal = sighupReceived[0] |
| |
| if pending: |
| pending = self.wconfd.Client().HasPendingRequest(self._wconfdcontext) |
| |
| if pending and signal: |
| logging.warning("Ignoring unexpected SIGHUP") |
| sighupReceived[0] = False |
| |
| logging.debug("Finished trying. Pending: %s", pending) |
| if pending: |
| raise LockAcquireTimeout() |
| |
| def _AcquireLocks(self, level, names, shared, opportunistic, timeout, |
| opportunistic_count=1, request_only=False): |
| """Acquires locks via the Ganeti lock manager. |
| |
| @type level: int |
| @param level: Lock level |
| @type names: list or string |
| @param names: Lock names |
| @type shared: bool |
| @param shared: Whether the locks should be acquired in shared mode |
| @type opportunistic: bool |
| @param opportunistic: Whether to acquire opportunistically |
| @type timeout: None or float |
| @param timeout: Timeout for acquiring the locks |
| @type request_only: bool |
| @param request_only: do not acquire the locks, just return the request |
| @raise LockAcquireTimeout: In case locks couldn't be acquired in specified |
| amount of time; in this case, locks still might be acquired or a request |
| pending. |
| |
| """ |
| self._CheckLocksEnabled() |
| |
| if self._cbs: |
| priority = self._cbs.CurrentPriority() # pylint: disable=W0612 |
| else: |
| priority = None |
| |
| if priority is None: |
| priority = constants.OP_PRIO_DEFAULT |
| |
| if names == locking.ALL_SET: |
| if opportunistic: |
| expand_fns = { |
| locking.LEVEL_CLUSTER: (lambda: [locking.BGL]), |
| locking.LEVEL_INSTANCE: self.cfg.GetInstanceList, |
| locking.LEVEL_NODEGROUP: self.cfg.GetNodeGroupList, |
| locking.LEVEL_NODE: self.cfg.GetNodeList, |
| locking.LEVEL_NODE_RES: self.cfg.GetNodeList, |
| locking.LEVEL_NETWORK: self.cfg.GetNetworkList, |
| } |
| names = expand_fns[level]() |
| else: |
| names = locking.LOCKSET_NAME |
| |
| names = _LockList(names) |
| |
| # For locks of the same level, the lock order is lexicographic |
| names.sort() |
| |
| levelname = locking.LEVEL_NAMES[level] |
| |
| locks = ["%s/%s" % (levelname, lock) for lock in list(names)] |
| |
| if not names: |
| logging.debug("Acquiring no locks for (%s) at level %s", |
| self._wconfdcontext, levelname) |
| return [] |
| |
| if shared: |
| request = [[lock, "shared"] for lock in locks] |
| else: |
| request = [[lock, "exclusive"] for lock in locks] |
| |
| if request_only: |
| logging.debug("Lock request for level %s is %s", level, request) |
| return request |
| |
| self.cfg.OutDate() |
| |
| if timeout is None: |
| ## Note: once we are so desperate for locks to request them |
| ## unconditionally, we no longer care about an original plan |
| ## to acquire locks opportunistically. |
| logging.info("Definitely requesting %s for %s", |
| request, self._wconfdcontext) |
| ## The only way to be sure of not getting starved is to sequentially |
| ## acquire the locks one by one (in lock order). |
| for r in request: |
| logging.debug("Definite request %s for %s", r, self._wconfdcontext) |
| self.wconfd.Client().UpdateLocksWaiting(self._wconfdcontext, priority, |
| [r]) |
| while True: |
| pending = self.wconfd.Client().HasPendingRequest(self._wconfdcontext) |
| if not pending: |
| break |
| time.sleep(10.0 * random.random()) |
| |
| elif opportunistic: |
| logging.debug("For %ss trying to opportunistically acquire" |
| " at least %d of %s for %s.", |
| timeout, opportunistic_count, locks, self._wconfdcontext) |
| locks = utils.SimpleRetry( |
| lambda l: l != [], self.wconfd.Client().GuardedOpportunisticLockUnion, |
| 2.0, timeout, args=[opportunistic_count, self._wconfdcontext, request]) |
| logging.debug("Managed to get the following locks: %s", locks) |
| if locks == []: |
| raise LockAcquireTimeout() |
| else: |
| self._RequestAndWait(request, timeout) |
| |
| return locks |
| |
| def _ExecLU(self, lu): |
| """Logical Unit execution sequence. |
| |
| """ |
| write_count = self.cfg.write_count |
| lu.cfg.OutDate() |
| lu.CheckPrereq() |
| |
| hm = self.BuildHooksManager(lu) |
| h_results = hm.RunPhase(constants.HOOKS_PHASE_PRE) |
| lu.HooksCallBack(constants.HOOKS_PHASE_PRE, h_results, |
| self.Log, None) |
| |
| if getattr(lu.op, "dry_run", False): |
| # in this mode, no post-hooks are run, and the config is not |
| # written (as it might have been modified by another LU, and we |
| # shouldn't do writeout on behalf of other threads |
| self.LogInfo("dry-run mode requested, not actually executing" |
| " the operation") |
| return lu.dry_run_result |
| |
| if self._cbs: |
| submit_mj_fn = self._cbs.SubmitManyJobs |
| else: |
| submit_mj_fn = _FailingSubmitManyJobs |
| |
| lusExecuting[0] += 1 |
| try: |
| result = _ProcessResult(submit_mj_fn, lu.op, lu.Exec(self.Log)) |
| h_results = hm.RunPhase(constants.HOOKS_PHASE_POST) |
| result = lu.HooksCallBack(constants.HOOKS_PHASE_POST, h_results, |
| self.Log, result) |
| finally: |
| # FIXME: This needs locks if not lu_class.REQ_BGL |
| lusExecuting[0] -= 1 |
| if write_count != self.cfg.write_count: |
| hm.RunConfigUpdate() |
| |
| return result |
| |
| def BuildHooksManager(self, lu): |
| return self.hmclass.BuildFromLu(lu.rpc.call_hooks_runner, lu) |
| |
| def _LockAndExecLU(self, lu, level, calc_timeout, pending=None): |
| """Execute a Logical Unit, with the needed locks. |
| |
| This is a recursive function that starts locking the given level, and |
| proceeds up, till there are no more locks to acquire. Then it executes the |
| given LU and its opcodes. |
| |
| """ |
| pending = pending or [] |
| logging.debug("Looking at locks of level %s, still need to obtain %s", |
| level, pending) |
| adding_locks = level in lu.add_locks |
| acquiring_locks = level in lu.needed_locks |
| |
| if level not in locking.LEVELS: |
| if pending: |
| self._RequestAndWait(pending, calc_timeout()) |
| lu.cfg.OutDate() |
| lu.wconfdlocks = self.wconfd.Client().ListLocks(self._wconfdcontext) |
| pending = [] |
| |
| logging.debug("Finished acquiring locks") |
| |
| if self._cbs: |
| self._cbs.NotifyStart() |
| |
| try: |
| result = self._ExecLU(lu) |
| except errors.OpPrereqError, err: |
| if len(err.args) < 2 or err.args[1] != errors.ECODE_TEMP_NORES: |
| raise |
| |
| logging.debug("Temporarily out of resources; will retry internally") |
| try: |
| lu.PrepareRetry(self.Log) |
| if self._cbs: |
| self._cbs.NotifyRetry() |
| except errors.OpRetryNotSupportedError: |
| logging.debug("LU does not know how to retry.") |
| raise err |
| raise LockAcquireTimeout() |
| except AssertionError, err: |
| # this is a bit ugly, as we don't know from which phase |
| # (prereq, exec) this comes; but it's better than an exception |
| # with no information |
| (_, _, tb) = sys.exc_info() |
| err_info = traceback.format_tb(tb) |
| del tb |
| logging.exception("Detected AssertionError") |
| raise errors.OpExecError("Internal assertion error: please report" |
| " this as a bug.\nError message: '%s';" |
| " location:\n%s" % (str(err), err_info[-1])) |
| return result |
| |
| # Determine if the acquiring is opportunistic up front |
| opportunistic = lu.opportunistic_locks[level] |
| |
| dont_collate = lu.dont_collate_locks[level] |
| |
| if dont_collate and pending: |
| self._RequestAndWait(pending, calc_timeout()) |
| lu.cfg.OutDate() |
| lu.wconfdlocks = self.wconfd.Client().ListLocks(self._wconfdcontext) |
| pending = [] |
| |
| if adding_locks and opportunistic: |
| # We could simultaneously acquire locks opportunistically and add new |
| # ones, but that would require altering the API, and no use cases are |
| # present in the system at the moment. |
| raise NotImplementedError("Can't opportunistically acquire locks when" |
| " adding new ones") |
| |
| if adding_locks and acquiring_locks and \ |
| lu.needed_locks[level] == locking.ALL_SET: |
| # It would also probably be possible to acquire all locks of a certain |
| # type while adding new locks, but there is no use case at the moment. |
| raise NotImplementedError("Can't request all locks of a certain level" |
| " and add new locks") |
| |
| if adding_locks or acquiring_locks: |
| self._CheckLocksEnabled() |
| |
| lu.DeclareLocks(level) |
| share = lu.share_locks[level] |
| opportunistic_count = lu.opportunistic_locks_count[level] |
| |
| try: |
| if acquiring_locks: |
| needed_locks = _LockList(lu.needed_locks[level]) |
| else: |
| needed_locks = [] |
| |
| if adding_locks: |
| needed_locks.extend(_LockList(lu.add_locks[level])) |
| |
| timeout = calc_timeout() |
| if timeout is not None and not opportunistic: |
| pending = pending + self._AcquireLocks(level, needed_locks, share, |
| opportunistic, timeout, |
| request_only=True) |
| else: |
| if pending: |
| self._RequestAndWait(pending, calc_timeout()) |
| lu.cfg.OutDate() |
| lu.wconfdlocks = self.wconfd.Client().ListLocks(self._wconfdcontext) |
| pending = [] |
| self._AcquireLocks(level, needed_locks, share, opportunistic, |
| timeout, |
| opportunistic_count=opportunistic_count) |
| lu.wconfdlocks = self.wconfd.Client().ListLocks(self._wconfdcontext) |
| |
| result = self._LockAndExecLU(lu, level + 1, calc_timeout, |
| pending=pending) |
| finally: |
| levelname = locking.LEVEL_NAMES[level] |
| logging.debug("Freeing locks at level %s for %s", |
| levelname, self._wconfdcontext) |
| self.wconfd.Client().FreeLocksLevel(self._wconfdcontext, levelname) |
| else: |
| result = self._LockAndExecLU(lu, level + 1, calc_timeout, pending=pending) |
| |
| return result |
| |
| # pylint: disable=R0201 |
| def _CheckLUResult(self, op, result): |
| """Check the LU result against the contract in the opcode. |
| |
| """ |
| resultcheck_fn = op.OP_RESULT |
| if not (resultcheck_fn is None or resultcheck_fn(result)): |
| logging.error("Expected opcode result matching %s, got %s", |
| resultcheck_fn, result) |
| if not getattr(op, "dry_run", False): |
| # FIXME: LUs should still behave in dry_run mode, or |
| # alternately we should have OP_DRYRUN_RESULT; in the |
| # meantime, we simply skip the OP_RESULT check in dry-run mode |
| raise errors.OpResultError("Opcode result does not match %s: %s" % |
| (resultcheck_fn, utils.Truncate(result, 80))) |
| |
| def ExecOpCode(self, op, cbs, timeout=None): |
| """Execute an opcode. |
| |
| @type op: an OpCode instance |
| @param op: the opcode to be executed |
| @type cbs: L{OpExecCbBase} |
| @param cbs: Runtime callbacks |
| @type timeout: float or None |
| @param timeout: Maximum time to acquire all locks, None for no timeout |
| @raise LockAcquireTimeout: In case locks couldn't be acquired in specified |
| amount of time |
| |
| """ |
| if not isinstance(op, opcodes.OpCode): |
| raise errors.ProgrammerError("Non-opcode instance passed" |
| " to ExecOpcode (%s)" % type(op)) |
| |
| lu_class = self.DISPATCH_TABLE.get(op.__class__, None) |
| if lu_class is None: |
| raise errors.OpCodeUnknown("Unknown opcode") |
| |
| if timeout is None: |
| calc_timeout = lambda: None |
| else: |
| calc_timeout = utils.RunningTimeout(timeout, False).Remaining |
| |
| self._cbs = cbs |
| try: |
| if self._enable_locks: |
| # Acquire the Big Ganeti Lock exclusively if this LU requires it, |
| # and in a shared fashion otherwise (to prevent concurrent run with |
| # an exclusive LU. |
| self._AcquireLocks(locking.LEVEL_CLUSTER, locking.BGL, |
| not lu_class.REQ_BGL, False, calc_timeout()) |
| elif lu_class.REQ_BGL: |
| raise errors.ProgrammerError("Opcode '%s' requires BGL, but locks are" |
| " disabled" % op.OP_ID) |
| |
| lu = lu_class(self, op, self.cfg, self.rpc, |
| self._wconfdcontext, self.wconfd) |
| lu.wconfdlocks = self.wconfd.Client().ListLocks(self._wconfdcontext) |
| _CheckSecretParameters(op) |
| lu.ExpandNames() |
| assert lu.needed_locks is not None, "needed_locks not set by LU" |
| |
| try: |
| result = self._LockAndExecLU(lu, locking.LEVEL_CLUSTER + 1, |
| calc_timeout) |
| finally: |
| if self._ec_id: |
| self.cfg.DropECReservations(self._ec_id) |
| finally: |
| self.wconfd.Client().FreeLocksLevel( |
| self._wconfdcontext, locking.LEVEL_NAMES[locking.LEVEL_CLUSTER]) |
| self._cbs = None |
| |
| self._CheckLUResult(op, result) |
| |
| return result |
| |
| def Log(self, *args): |
| """Forward call to feedback callback function. |
| |
| """ |
| if self._cbs: |
| self._cbs.Feedback(*args) |
| |
| def LogStep(self, current, total, message): |
| """Log a change in LU execution progress. |
| |
| """ |
| logging.debug("Step %d/%d %s", current, total, message) |
| self.Log("STEP %d/%d %s" % (current, total, message)) |
| |
| def LogWarning(self, message, *args, **kwargs): |
| """Log a warning to the logs and the user. |
| |
| The optional keyword argument is 'hint' and can be used to show a |
| hint to the user (presumably related to the warning). If the |
| message is empty, it will not be printed at all, allowing one to |
| show only a hint. |
| |
| """ |
| assert not kwargs or (len(kwargs) == 1 and "hint" in kwargs), \ |
| "Invalid keyword arguments for LogWarning (%s)" % str(kwargs) |
| if args: |
| message = message % tuple(args) |
| if message: |
| logging.warning(message) |
| self.Log(" - WARNING: %s" % message) |
| if "hint" in kwargs: |
| self.Log(" Hint: %s" % kwargs["hint"]) |
| |
| def LogInfo(self, message, *args): |
| """Log an informational message to the logs and the user. |
| |
| """ |
| if args: |
| message = message % tuple(args) |
| logging.info(message) |
| self.Log(" - INFO: %s" % message) |
| |
| def GetECId(self): |
| """Returns the current execution context ID. |
| |
| """ |
| if not self._ec_id: |
| raise errors.ProgrammerError("Tried to use execution context id when" |
| " not set") |
| return self._ec_id |
