Statistics
| Branch: | Tag: | Revision:

root / lib / mcpu.py @ afee0879

History | View | Annotate | Download (13.1 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007 Google Inc.
5
#
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
10
#
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14
# General Public License for more details.
15
#
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19
# 02110-1301, USA.
20

    
21

    
22
"""Module implementing the logic behind the cluster operations
23

24
This module implements the logic for doing operations in the cluster. There
25
are two kinds of classes defined:
26
  - logical units, which know how to deal with their specific opcode only
27
  - the processor, which dispatches the opcodes to their logical units
28

29
"""
30

    
31
import logging
32

    
33
from ganeti import opcodes
34
from ganeti import constants
35
from ganeti import errors
36
from ganeti import rpc
37
from ganeti import cmdlib
38
from ganeti import locking
39

    
40

    
41
class Processor(object):
42
  """Object which runs OpCodes"""
43
  DISPATCH_TABLE = {
44
    # Cluster
45
    opcodes.OpDestroyCluster: cmdlib.LUDestroyCluster,
46
    opcodes.OpQueryClusterInfo: cmdlib.LUQueryClusterInfo,
47
    opcodes.OpVerifyCluster: cmdlib.LUVerifyCluster,
48
    opcodes.OpQueryConfigValues: cmdlib.LUQueryConfigValues,
49
    opcodes.OpRenameCluster: cmdlib.LURenameCluster,
50
    opcodes.OpVerifyDisks: cmdlib.LUVerifyDisks,
51
    opcodes.OpSetClusterParams: cmdlib.LUSetClusterParams,
52
    opcodes.OpRedistributeConfig: cmdlib.LURedistributeConfig,
53
    # node lu
54
    opcodes.OpAddNode: cmdlib.LUAddNode,
55
    opcodes.OpQueryNodes: cmdlib.LUQueryNodes,
56
    opcodes.OpQueryNodeVolumes: cmdlib.LUQueryNodeVolumes,
57
    opcodes.OpRemoveNode: cmdlib.LURemoveNode,
58
    opcodes.OpSetNodeParams: cmdlib.LUSetNodeParams,
59
    # instance lu
60
    opcodes.OpCreateInstance: cmdlib.LUCreateInstance,
61
    opcodes.OpReinstallInstance: cmdlib.LUReinstallInstance,
62
    opcodes.OpRemoveInstance: cmdlib.LURemoveInstance,
63
    opcodes.OpRenameInstance: cmdlib.LURenameInstance,
64
    opcodes.OpActivateInstanceDisks: cmdlib.LUActivateInstanceDisks,
65
    opcodes.OpShutdownInstance: cmdlib.LUShutdownInstance,
66
    opcodes.OpStartupInstance: cmdlib.LUStartupInstance,
67
    opcodes.OpRebootInstance: cmdlib.LURebootInstance,
68
    opcodes.OpDeactivateInstanceDisks: cmdlib.LUDeactivateInstanceDisks,
69
    opcodes.OpReplaceDisks: cmdlib.LUReplaceDisks,
70
    opcodes.OpFailoverInstance: cmdlib.LUFailoverInstance,
71
    opcodes.OpConnectConsole: cmdlib.LUConnectConsole,
72
    opcodes.OpQueryInstances: cmdlib.LUQueryInstances,
73
    opcodes.OpQueryInstanceData: cmdlib.LUQueryInstanceData,
74
    opcodes.OpSetInstanceParams: cmdlib.LUSetInstanceParams,
75
    opcodes.OpGrowDisk: cmdlib.LUGrowDisk,
76
    # os lu
77
    opcodes.OpDiagnoseOS: cmdlib.LUDiagnoseOS,
78
    # exports lu
79
    opcodes.OpQueryExports: cmdlib.LUQueryExports,
80
    opcodes.OpExportInstance: cmdlib.LUExportInstance,
81
    opcodes.OpRemoveExport: cmdlib.LURemoveExport,
82
    # tags lu
83
    opcodes.OpGetTags: cmdlib.LUGetTags,
84
    opcodes.OpSearchTags: cmdlib.LUSearchTags,
85
    opcodes.OpAddTags: cmdlib.LUAddTags,
86
    opcodes.OpDelTags: cmdlib.LUDelTags,
87
    # test lu
88
    opcodes.OpTestDelay: cmdlib.LUTestDelay,
89
    opcodes.OpTestAllocator: cmdlib.LUTestAllocator,
90
    }
91

    
92
  def __init__(self, context):
93
    """Constructor for Processor
94

95
    Args:
96
     - feedback_fn: the feedback function (taking one string) to be run when
97
                    interesting events are happening
98
    """
99
    self.context = context
100
    self._feedback_fn = None
101
    self.exclusive_BGL = False
102
    self.rpc = rpc.RpcRunner(context.cfg)
103

    
104
  def _ExecLU(self, lu):
105
    """Logical Unit execution sequence.
106

107
    """
108
    write_count = self.context.cfg.write_count
109
    lu.CheckPrereq()
110
    hm = HooksMaster(self.rpc.call_hooks_runner, self, lu)
111
    h_results = hm.RunPhase(constants.HOOKS_PHASE_PRE)
112
    lu.HooksCallBack(constants.HOOKS_PHASE_PRE, h_results,
113
                     self._feedback_fn, None)
114
    try:
115
      result = lu.Exec(self._feedback_fn)
116
      h_results = hm.RunPhase(constants.HOOKS_PHASE_POST)
117
      result = lu.HooksCallBack(constants.HOOKS_PHASE_POST, h_results,
118
                                self._feedback_fn, result)
119
    finally:
120
      # FIXME: This needs locks if not lu_class.REQ_BGL
121
      if write_count != self.context.cfg.write_count:
122
        hm.RunConfigUpdate()
123

    
124
    return result
125

    
126
  def _LockAndExecLU(self, lu, level):
127
    """Execute a Logical Unit, with the needed locks.
128

129
    This is a recursive function that starts locking the given level, and
130
    proceeds up, till there are no more locks to acquire. Then it executes the
131
    given LU and its opcodes.
132

133
    """
134
    adding_locks = level in lu.add_locks
135
    acquiring_locks = level in lu.needed_locks
136
    if level not in locking.LEVELS:
137
      if callable(self._run_notifier):
138
        self._run_notifier()
139
      result = self._ExecLU(lu)
140
    elif adding_locks and acquiring_locks:
141
      # We could both acquire and add locks at the same level, but for now we
142
      # don't need this, so we'll avoid the complicated code needed.
143
      raise NotImplementedError(
144
        "Can't declare locks to acquire when adding others")
145
    elif adding_locks or acquiring_locks:
146
      lu.DeclareLocks(level)
147
      share = lu.share_locks[level]
148
      if acquiring_locks:
149
        needed_locks = lu.needed_locks[level]
150
        lu.acquired_locks[level] = self.context.glm.acquire(level,
151
                                                            needed_locks,
152
                                                            shared=share)
153
      else: # adding_locks
154
        add_locks = lu.add_locks[level]
155
        lu.remove_locks[level] = add_locks
156
        try:
157
          self.context.glm.add(level, add_locks, acquired=1, shared=share)
158
        except errors.LockError:
159
          raise errors.OpPrereqError(
160
            "Coudn't add locks (%s), probably because of a race condition"
161
            " with another job, who added them first" % add_locks)
162
      try:
163
        try:
164
          if adding_locks:
165
            lu.acquired_locks[level] = add_locks
166
          result = self._LockAndExecLU(lu, level + 1)
167
        finally:
168
          if level in lu.remove_locks:
169
            self.context.glm.remove(level, lu.remove_locks[level])
170
      finally:
171
        if self.context.glm.is_owned(level):
172
          self.context.glm.release(level)
173
    else:
174
      result = self._LockAndExecLU(lu, level + 1)
175

    
176
    return result
177

    
178
  def ExecOpCode(self, op, feedback_fn, run_notifier):
179
    """Execute an opcode.
180

181
    @type op: an OpCode instance
182
    @param op: the opcode to be executed
183
    @type feedback_fn: a function that takes a single argument
184
    @param feedback_fn: this function will be used as feedback from the LU
185
                        code to the end-user
186
    @type run_notifier: callable (no arguments) or None
187
    @param run_notifier:  this function (if callable) will be called when
188
                          we are about to call the lu's Exec() method, that
189
                          is, after we have aquired all locks
190

191
    """
192
    if not isinstance(op, opcodes.OpCode):
193
      raise errors.ProgrammerError("Non-opcode instance passed"
194
                                   " to ExecOpcode")
195

    
196
    self._feedback_fn = feedback_fn
197
    self._run_notifier = run_notifier
198
    lu_class = self.DISPATCH_TABLE.get(op.__class__, None)
199
    if lu_class is None:
200
      raise errors.OpCodeUnknown("Unknown opcode")
201

    
202
    # Acquire the Big Ganeti Lock exclusively if this LU requires it, and in a
203
    # shared fashion otherwise (to prevent concurrent run with an exclusive LU.
204
    self.context.glm.acquire(locking.LEVEL_CLUSTER, [locking.BGL],
205
                             shared=not lu_class.REQ_BGL)
206
    try:
207
      self.exclusive_BGL = lu_class.REQ_BGL
208
      lu = lu_class(self, op, self.context, self.rpc)
209
      lu.ExpandNames()
210
      assert lu.needed_locks is not None, "needed_locks not set by LU"
211
      result = self._LockAndExecLU(lu, locking.LEVEL_INSTANCE)
212
    finally:
213
      self.context.glm.release(locking.LEVEL_CLUSTER)
214
      self.exclusive_BGL = False
215

    
216
    return result
217

    
218
  def LogStep(self, current, total, message):
219
    """Log a change in LU execution progress.
220

221
    """
222
    logging.debug("Step %d/%d %s", current, total, message)
223
    self._feedback_fn("STEP %d/%d %s" % (current, total, message))
224

    
225
  def LogWarning(self, message, *args, **kwargs):
226
    """Log a warning to the logs and the user.
227

228
    The optional keyword argument is 'hint' and can be used to show a
229
    hint to the user (presumably related to the warning). If the
230
    message is empty, it will not be printed at all, allowing one to
231
    show only a hint.
232

233
    """
234
    assert not kwargs or (len(kwargs) == 1 and "hint" in kwargs), \
235
           "Invalid keyword arguments for LogWarning (%s)" % str(kwargs)
236
    if args:
237
      message = message % tuple(args)
238
    if message:
239
      logging.warning(message)
240
      self._feedback_fn(" - WARNING: %s" % message)
241
    if "hint" in kwargs:
242
      self._feedback_fn("      Hint: %s" % kwargs["hint"])
243

    
244
  def LogInfo(self, message, *args):
245
    """Log an informational message to the logs and the user.
246

247
    """
248
    if args:
249
      message = message % tuple(args)
250
    logging.info(message)
251
    self._feedback_fn(" - INFO: %s" % message)
252

    
253

    
254
class HooksMaster(object):
255
  """Hooks master.
256

257
  This class distributes the run commands to the nodes based on the
258
  specific LU class.
259

260
  In order to remove the direct dependency on the rpc module, the
261
  constructor needs a function which actually does the remote
262
  call. This will usually be rpc.call_hooks_runner, but any function
263
  which behaves the same works.
264

265
  """
266
  def __init__(self, callfn, proc, lu):
267
    self.callfn = callfn
268
    self.proc = proc
269
    self.lu = lu
270
    self.op = lu.op
271
    self.env, node_list_pre, node_list_post = self._BuildEnv()
272
    self.node_list = {constants.HOOKS_PHASE_PRE: node_list_pre,
273
                      constants.HOOKS_PHASE_POST: node_list_post}
274

    
275
  def _BuildEnv(self):
276
    """Compute the environment and the target nodes.
277

278
    Based on the opcode and the current node list, this builds the
279
    environment for the hooks and the target node list for the run.
280

281
    """
282
    env = {
283
      "PATH": "/sbin:/bin:/usr/sbin:/usr/bin",
284
      "GANETI_HOOKS_VERSION": constants.HOOKS_VERSION,
285
      "GANETI_OP_CODE": self.op.OP_ID,
286
      "GANETI_OBJECT_TYPE": self.lu.HTYPE,
287
      "GANETI_DATA_DIR": constants.DATA_DIR,
288
      }
289

    
290
    if self.lu.HPATH is not None:
291
      lu_env, lu_nodes_pre, lu_nodes_post = self.lu.BuildHooksEnv()
292
      if lu_env:
293
        for key in lu_env:
294
          env["GANETI_" + key] = lu_env[key]
295
    else:
296
      lu_nodes_pre = lu_nodes_post = []
297

    
298
    return env, frozenset(lu_nodes_pre), frozenset(lu_nodes_post)
299

    
300
  def _RunWrapper(self, node_list, hpath, phase):
301
    """Simple wrapper over self.callfn.
302

303
    This method fixes the environment before doing the rpc call.
304

305
    """
306
    env = self.env.copy()
307
    env["GANETI_HOOKS_PHASE"] = phase
308
    env["GANETI_HOOKS_PATH"] = hpath
309
    if self.lu.cfg is not None:
310
      env["GANETI_CLUSTER"] = self.lu.cfg.GetClusterName()
311
      env["GANETI_MASTER"] = self.lu.cfg.GetMasterNode()
312

    
313
    env = dict([(str(key), str(val)) for key, val in env.iteritems()])
314

    
315
    return self.callfn(node_list, hpath, phase, env)
316

    
317
  def RunPhase(self, phase):
318
    """Run all the scripts for a phase.
319

320
    This is the main function of the HookMaster.
321

322
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
323
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
324
    @return: the processed results of the hooks multi-node rpc call
325
    @raise errors.HooksFailure: on communication failure to the nodes
326

327
    """
328
    if not self.node_list[phase]:
329
      # empty node list, we should not attempt to run this as either
330
      # we're in the cluster init phase and the rpc client part can't
331
      # even attempt to run, or this LU doesn't do hooks at all
332
      return
333
    hpath = self.lu.HPATH
334
    results = self._RunWrapper(self.node_list[phase], hpath, phase)
335
    if phase == constants.HOOKS_PHASE_PRE:
336
      errs = []
337
      if not results:
338
        raise errors.HooksFailure("Communication failure")
339
      for node_name in results:
340
        res = results[node_name]
341
        if res.failed or res.data is False or not isinstance(res.data, list):
342
          if not res.offline:
343
            self.proc.LogWarning("Communication failure to node %s" %
344
                                 node_name)
345
          continue
346
        for script, hkr, output in res.data:
347
          if hkr == constants.HKR_FAIL:
348
            output = output.strip().encode("string_escape")
349
            errs.append((node_name, script, output))
350
      if errs:
351
        raise errors.HooksAbort(errs)
352
    return results
353

    
354
  def RunConfigUpdate(self):
355
    """Run the special configuration update hook
356

357
    This is a special hook that runs only on the master after each
358
    top-level LI if the configuration has been updated.
359

360
    """
361
    phase = constants.HOOKS_PHASE_POST
362
    hpath = constants.HOOKS_NAME_CFGUPDATE
363
    nodes = [self.lu.cfg.GetMasterNode()]
364
    results = self._RunWrapper(nodes, hpath, phase)