code.grnet.gr Git - ganeti-local/blob - lib/rpc.py

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 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 """Inter-node RPC library.
  23
  24 """
  25
  26 # pylint: disable=C0103,R0201,R0904
  27 # C0103: Invalid name, since call_ are not valid
  28 # R0201: Method could be a function, we keep all rpcs instance methods
  29 # as not to change them back and forth between static/instance methods
  30 # if they need to start using instance attributes
  31 # R0904: Too many public methods
  32
  33 import logging
  34 import zlib
  35 import base64
  36 import pycurl
  37 import threading
  38
  39 from ganeti import utils
  40 from ganeti import objects
  41 from ganeti import http
  42 from ganeti import serializer
  43 from ganeti import constants
  44 from ganeti import errors
  45 from ganeti import netutils
  46 from ganeti import ssconf
  47 from ganeti import runtime
  48 from ganeti import compat
  49 from ganeti import rpc_defs
  50
  51 # Special module generated at build time
  52 from ganeti import _generated_rpc
  53
  54 # pylint has a bug here, doesn't see this import
  55 import ganeti.http.client  # pylint: disable=W0611
  56
  57
  58 # Timeout for connecting to nodes (seconds)
  59 _RPC_CONNECT_TIMEOUT = 5
  60
  61 _RPC_CLIENT_HEADERS = [
  62   "Content-type: %s" % http.HTTP_APP_JSON,
  63   "Expect:",
  64   ]
  65
  66 # Various time constants for the timeout table
  67 _TMO_URGENT = 60 # one minute
  68 _TMO_FAST = 5 * 60 # five minutes
  69 _TMO_NORMAL = 15 * 60 # 15 minutes
  70 _TMO_SLOW = 3600 # one hour
  71 _TMO_4HRS = 4 * 3600
  72 _TMO_1DAY = 86400
  73
  74 #: Special value to describe an offline host
  75 _OFFLINE = object()
  76
  77
  78 def Init():
  79   """Initializes the module-global HTTP client manager.
  80
  81   Must be called before using any RPC function and while exactly one thread is
  82   running.
  83
  84   """
  85   # curl_global_init(3) and curl_global_cleanup(3) must be called with only
  86   # one thread running. This check is just a safety measure -- it doesn't
  87   # cover all cases.
  88   assert threading.activeCount() == 1, \
  89          "Found more than one active thread when initializing pycURL"
  90
  91   logging.info("Using PycURL %s", pycurl.version)
  92
  93   pycurl.global_init(pycurl.GLOBAL_ALL)
  94
  95
  96 def Shutdown():
  97   """Stops the module-global HTTP client manager.
  98
  99   Must be called before quitting the program and while exactly one thread is
 100   running.
 101
 102   """
 103   pycurl.global_cleanup()
 104
 105
 106 def _ConfigRpcCurl(curl):
 107   noded_cert = str(constants.NODED_CERT_FILE)
 108
 109   curl.setopt(pycurl.FOLLOWLOCATION, False)
 110   curl.setopt(pycurl.CAINFO, noded_cert)
 111   curl.setopt(pycurl.SSL_VERIFYHOST, 0)
 112   curl.setopt(pycurl.SSL_VERIFYPEER, True)
 113   curl.setopt(pycurl.SSLCERTTYPE, "PEM")
 114   curl.setopt(pycurl.SSLCERT, noded_cert)
 115   curl.setopt(pycurl.SSLKEYTYPE, "PEM")
 116   curl.setopt(pycurl.SSLKEY, noded_cert)
 117   curl.setopt(pycurl.CONNECTTIMEOUT, _RPC_CONNECT_TIMEOUT)
 118
 119
 120 def RunWithRPC(fn):
 121   """RPC-wrapper decorator.
 122
 123   When applied to a function, it runs it with the RPC system
 124   initialized, and it shutsdown the system afterwards. This means the
 125   function must be called without RPC being initialized.
 126
 127   """
 128   def wrapper(*args, **kwargs):
 129     Init()
 130     try:
 131       return fn(*args, **kwargs)
 132     finally:
 133       Shutdown()
 134   return wrapper
 135
 136
 137 def _Compress(data):
 138   """Compresses a string for transport over RPC.
 139
 140   Small amounts of data are not compressed.
 141
 142   @type data: str
 143   @param data: Data
 144   @rtype: tuple
 145   @return: Encoded data to send
 146
 147   """
 148   # Small amounts of data are not compressed
 149   if len(data) < 512:
 150     return (constants.RPC_ENCODING_NONE, data)
 151
 152   # Compress with zlib and encode in base64
 153   return (constants.RPC_ENCODING_ZLIB_BASE64,
 154           base64.b64encode(zlib.compress(data, 3)))
 155
 156
 157 class RpcResult(object):
 158   """RPC Result class.
 159
 160   This class holds an RPC result. It is needed since in multi-node
 161   calls we can't raise an exception just because one one out of many
 162   failed, and therefore we use this class to encapsulate the result.
 163
 164   @ivar data: the data payload, for successful results, or None
 165   @ivar call: the name of the RPC call
 166   @ivar node: the name of the node to which we made the call
 167   @ivar offline: whether the operation failed because the node was
 168       offline, as opposed to actual failure; offline=True will always
 169       imply failed=True, in order to allow simpler checking if
 170       the user doesn't care about the exact failure mode
 171   @ivar fail_msg: the error message if the call failed
 172
 173   """
 174   def __init__(self, data=None, failed=False, offline=False,
 175                call=None, node=None):
 176     self.offline = offline
 177     self.call = call
 178     self.node = node
 179
 180     if offline:
 181       self.fail_msg = "Node is marked offline"
 182       self.data = self.payload = None
 183     elif failed:
 184       self.fail_msg = self._EnsureErr(data)
 185       self.data = self.payload = None
 186     else:
 187       self.data = data
 188       if not isinstance(self.data, (tuple, list)):
 189         self.fail_msg = ("RPC layer error: invalid result type (%s)" %
 190                          type(self.data))
 191         self.payload = None
 192       elif len(data) != 2:
 193         self.fail_msg = ("RPC layer error: invalid result length (%d), "
 194                          "expected 2" % len(self.data))
 195         self.payload = None
 196       elif not self.data[0]:
 197         self.fail_msg = self._EnsureErr(self.data[1])
 198         self.payload = None
 199       else:
 200         # finally success
 201         self.fail_msg = None
 202         self.payload = data[1]
 203
 204     for attr_name in ["call", "data", "fail_msg",
 205                       "node", "offline", "payload"]:
 206       assert hasattr(self, attr_name), "Missing attribute %s" % attr_name
 207
 208   @staticmethod
 209   def _EnsureErr(val):
 210     """Helper to ensure we return a 'True' value for error."""
 211     if val:
 212       return val
 213     else:
 214       return "No error information"
 215
 216   def Raise(self, msg, prereq=False, ecode=None):
 217     """If the result has failed, raise an OpExecError.
 218
 219     This is used so that LU code doesn't have to check for each
 220     result, but instead can call this function.
 221
 222     """
 223     if not self.fail_msg:
 224       return
 225
 226     if not msg: # one could pass None for default message
 227       msg = ("Call '%s' to node '%s' has failed: %s" %
 228              (self.call, self.node, self.fail_msg))
 229     else:
 230       msg = "%s: %s" % (msg, self.fail_msg)
 231     if prereq:
 232       ec = errors.OpPrereqError
 233     else:
 234       ec = errors.OpExecError
 235     if ecode is not None:
 236       args = (msg, ecode)
 237     else:
 238       args = (msg, )
 239     raise ec(*args) # pylint: disable=W0142
 240
 241
 242 def _SsconfResolver(ssconf_ips, node_list, _,
 243                     ssc=ssconf.SimpleStore,
 244                     nslookup_fn=netutils.Hostname.GetIP):
 245   """Return addresses for given node names.
 246
 247   @type ssconf_ips: bool
 248   @param ssconf_ips: Use the ssconf IPs
 249   @type node_list: list
 250   @param node_list: List of node names
 251   @type ssc: class
 252   @param ssc: SimpleStore class that is used to obtain node->ip mappings
 253   @type nslookup_fn: callable
 254   @param nslookup_fn: function use to do NS lookup
 255   @rtype: list of tuple; (string, string)
 256   @return: List of tuples containing node name and IP address
 257
 258   """
 259   ss = ssc()
 260   family = ss.GetPrimaryIPFamily()
 261
 262   if ssconf_ips:
 263     iplist = ss.GetNodePrimaryIPList()
 264     ipmap = dict(entry.split() for entry in iplist)
 265   else:
 266     ipmap = {}
 267
 268   result = []
 269   for node in node_list:
 270     ip = ipmap.get(node)
 271     if ip is None:
 272       ip = nslookup_fn(node, family=family)
 273     result.append((node, ip))
 274
 275   return result
 276
 277
 278 class _StaticResolver:
 279   def __init__(self, addresses):
 280     """Initializes this class.
 281
 282     """
 283     self._addresses = addresses
 284
 285   def __call__(self, hosts, _):
 286     """Returns static addresses for hosts.
 287
 288     """
 289     assert len(hosts) == len(self._addresses)
 290     return zip(hosts, self._addresses)
 291
 292
 293 def _CheckConfigNode(name, node, accept_offline_node):
 294   """Checks if a node is online.
 295
 296   @type name: string
 297   @param name: Node name
 298   @type node: L{objects.Node} or None
 299   @param node: Node object
 300
 301   """
 302   if node is None:
 303     # Depend on DNS for name resolution
 304     ip = name
 305   elif node.offline and not accept_offline_node:
 306     ip = _OFFLINE
 307   else:
 308     ip = node.primary_ip
 309   return (name, ip)
 310
 311
 312 def _NodeConfigResolver(single_node_fn, all_nodes_fn, hosts, opts):
 313   """Calculate node addresses using configuration.
 314
 315   """
 316   accept_offline_node = (opts is rpc_defs.ACCEPT_OFFLINE_NODE)
 317
 318   assert accept_offline_node or opts is None, "Unknown option"
 319
 320   # Special case for single-host lookups
 321   if len(hosts) == 1:
 322     (name, ) = hosts
 323     return [_CheckConfigNode(name, single_node_fn(name), accept_offline_node)]
 324   else:
 325     all_nodes = all_nodes_fn()
 326     return [_CheckConfigNode(name, all_nodes.get(name, None),
 327                              accept_offline_node)
 328             for name in hosts]
 329
 330
 331 class _RpcProcessor:
 332   def __init__(self, resolver, port, lock_monitor_cb=None):
 333     """Initializes this class.
 334
 335     @param resolver: callable accepting a list of hostnames, returning a list
 336       of tuples containing name and IP address (IP address can be the name or
 337       the special value L{_OFFLINE} to mark offline machines)
 338     @type port: int
 339     @param port: TCP port
 340     @param lock_monitor_cb: Callable for registering with lock monitor
 341
 342     """
 343     self._resolver = resolver
 344     self._port = port
 345     self._lock_monitor_cb = lock_monitor_cb
 346
 347   @staticmethod
 348   def _PrepareRequests(hosts, port, procedure, body, read_timeout):
 349     """Prepares requests by sorting offline hosts into separate list.
 350
 351     @type body: dict
 352     @param body: a dictionary with per-host body data
 353
 354     """
 355     results = {}
 356     requests = {}
 357
 358     assert isinstance(body, dict)
 359     assert len(body) == len(hosts)
 360     assert compat.all(isinstance(v, str) for v in body.values())
 361     assert frozenset(map(compat.fst, hosts)) == frozenset(body.keys()), \
 362         "%s != %s" % (hosts, body.keys())
 363
 364     for (name, ip) in hosts:
 365       if ip is _OFFLINE:
 366         # Node is marked as offline
 367         results[name] = RpcResult(node=name, offline=True, call=procedure)
 368       else:
 369         requests[name] = \
 370           http.client.HttpClientRequest(str(ip), port,
 371                                         http.HTTP_POST, str("/%s" % procedure),
 372                                         headers=_RPC_CLIENT_HEADERS,
 373                                         post_data=body[name],
 374                                         read_timeout=read_timeout,
 375                                         nicename="%s/%s" % (name, procedure),
 376                                         curl_config_fn=_ConfigRpcCurl)
 377
 378     return (results, requests)
 379
 380   @staticmethod
 381   def _CombineResults(results, requests, procedure):
 382     """Combines pre-computed results for offline hosts with actual call results.
 383
 384     """
 385     for name, req in requests.items():
 386       if req.success and req.resp_status_code == http.HTTP_OK:
 387         host_result = RpcResult(data=serializer.LoadJson(req.resp_body),
 388                                 node=name, call=procedure)
 389       else:
 390         # TODO: Better error reporting
 391         if req.error:
 392           msg = req.error
 393         else:
 394           msg = req.resp_body
 395
 396         logging.error("RPC error in %s on node %s: %s", procedure, name, msg)
 397         host_result = RpcResult(data=msg, failed=True, node=name,
 398                                 call=procedure)
 399
 400       results[name] = host_result
 401
 402     return results
 403
 404   def __call__(self, hosts, procedure, body, read_timeout, resolver_opts,
 405                _req_process_fn=None):
 406     """Makes an RPC request to a number of nodes.
 407
 408     @type hosts: sequence
 409     @param hosts: Hostnames
 410     @type procedure: string
 411     @param procedure: Request path
 412     @type body: dictionary
 413     @param body: dictionary with request bodies per host
 414     @type read_timeout: int or None
 415     @param read_timeout: Read timeout for request
 416
 417     """
 418     assert read_timeout is not None, \
 419       "Missing RPC read timeout for procedure '%s'" % procedure
 420
 421     if _req_process_fn is None:
 422       _req_process_fn = http.client.ProcessRequests
 423
 424     (results, requests) = \
 425       self._PrepareRequests(self._resolver(hosts, resolver_opts), self._port,
 426                             procedure, body, read_timeout)
 427
 428     _req_process_fn(requests.values(), lock_monitor_cb=self._lock_monitor_cb)
 429
 430     assert not frozenset(results).intersection(requests)
 431
 432     return self._CombineResults(results, requests, procedure)
 433
 434
 435 class _RpcClientBase:
 436   def __init__(self, resolver, encoder_fn, lock_monitor_cb=None,
 437                _req_process_fn=None):
 438     """Initializes this class.
 439
 440     """
 441     proc = _RpcProcessor(resolver,
 442                          netutils.GetDaemonPort(constants.NODED),
 443                          lock_monitor_cb=lock_monitor_cb)
 444     self._proc = compat.partial(proc, _req_process_fn=_req_process_fn)
 445     self._encoder = compat.partial(self._EncodeArg, encoder_fn)
 446
 447   @staticmethod
 448   def _EncodeArg(encoder_fn, (argkind, value)):
 449     """Encode argument.
 450
 451     """
 452     if argkind is None:
 453       return value
 454     else:
 455       return encoder_fn(argkind)(value)
 456
 457   def _Call(self, cdef, node_list, args):
 458     """Entry point for automatically generated RPC wrappers.
 459
 460     """
 461     (procedure, _, resolver_opts, timeout, argdefs,
 462      prep_fn, postproc_fn, _) = cdef
 463
 464     if callable(timeout):
 465       read_timeout = timeout(args)
 466     else:
 467       read_timeout = timeout
 468
 469     if callable(resolver_opts):
 470       req_resolver_opts = resolver_opts(args)
 471     else:
 472       req_resolver_opts = resolver_opts
 473
 474     if len(args) != len(argdefs):
 475       raise errors.ProgrammerError("Number of passed arguments doesn't match")
 476
 477     enc_args = map(self._encoder, zip(map(compat.snd, argdefs), args))
 478     if prep_fn is None:
 479       # for a no-op prep_fn, we serialise the body once, and then we
 480       # reuse it in the dictionary values
 481       body = serializer.DumpJson(enc_args)
 482       pnbody = dict((n, body) for n in node_list)
 483     else:
 484       # for a custom prep_fn, we pass the encoded arguments and the
 485       # node name to the prep_fn, and we serialise its return value
 486       assert callable(prep_fn)
 487       pnbody = dict((n, serializer.DumpJson(prep_fn(n, enc_args)))
 488                     for n in node_list)
 489
 490     result = self._proc(node_list, procedure, pnbody, read_timeout,
 491                         req_resolver_opts)
 492
 493     if postproc_fn:
 494       return dict(map(lambda (key, value): (key, postproc_fn(value)),
 495                       result.items()))
 496     else:
 497       return result
 498
 499
 500 def _ObjectToDict(value):
 501   """Converts an object to a dictionary.
 502
 503   @note: See L{objects}.
 504
 505   """
 506   return value.ToDict()
 507
 508
 509 def _ObjectListToDict(value):
 510   """Converts a list of L{objects} to dictionaries.
 511
 512   """
 513   return map(_ObjectToDict, value)
 514
 515
 516 def _EncodeNodeToDiskDict(value):
 517   """Encodes a dictionary with node name as key and disk objects as values.
 518
 519   """
 520   return dict((name, _ObjectListToDict(disks))
 521               for name, disks in value.items())
 522
 523
 524 def _PrepareFileUpload(getents_fn, filename):
 525   """Loads a file and prepares it for an upload to nodes.
 526
 527   """
 528   statcb = utils.FileStatHelper()
 529   data = _Compress(utils.ReadFile(filename, preread=statcb))
 530   st = statcb.st
 531
 532   if getents_fn is None:
 533     getents_fn = runtime.GetEnts
 534
 535   getents = getents_fn()
 536
 537   return [filename, data, st.st_mode, getents.LookupUid(st.st_uid),
 538           getents.LookupGid(st.st_gid), st.st_atime, st.st_mtime]
 539
 540
 541 def _PrepareFinalizeExportDisks(snap_disks):
 542   """Encodes disks for finalizing export.
 543
 544   """
 545   flat_disks = []
 546
 547   for disk in snap_disks:
 548     if isinstance(disk, bool):
 549       flat_disks.append(disk)
 550     else:
 551       flat_disks.append(disk.ToDict())
 552
 553   return flat_disks
 554
 555
 556 def _EncodeImportExportIO((ieio, ieioargs)):
 557   """Encodes import/export I/O information.
 558
 559   """
 560   if ieio == constants.IEIO_RAW_DISK:
 561     assert len(ieioargs) == 1
 562     return (ieio, (ieioargs[0].ToDict(), ))
 563
 564   if ieio == constants.IEIO_SCRIPT:
 565     assert len(ieioargs) == 2
 566     return (ieio, (ieioargs[0].ToDict(), ieioargs[1]))
 567
 568   return (ieio, ieioargs)
 569
 570
 571 def _EncodeBlockdevRename(value):
 572   """Encodes information for renaming block devices.
 573
 574   """
 575   return [(d.ToDict(), uid) for d, uid in value]
 576
 577
 578 def _AnnotateDParamsDRBD(disk, (drbd_params, data_params, meta_params)):
 579   """Annotates just DRBD disks layouts.
 580
 581   """
 582   assert disk.dev_type == constants.LD_DRBD8
 583
 584   disk.params = objects.FillDict(drbd_params, disk.params)
 585   (dev_data, dev_meta) = disk.children
 586   dev_data.params = objects.FillDict(data_params, dev_data.params)
 587   dev_meta.params = objects.FillDict(meta_params, dev_meta.params)
 588
 589   return disk
 590
 591
 592 def _AnnotateDParamsGeneric(disk, (params, )):
 593   """Generic disk parameter annotation routine.
 594
 595   """
 596   assert disk.dev_type != constants.LD_DRBD8
 597
 598   disk.params = objects.FillDict(params, disk.params)
 599
 600   return disk
 601
 602
 603 def AnnotateDiskParams(template, disks, disk_params):
 604   """Annotates the disk objects with the disk parameters.
 605
 606   @param template: The disk template used
 607   @param disks: The list of disks objects to annotate
 608   @param disk_params: The disk paramaters for annotation
 609   @returns: A list of disk objects annotated
 610
 611   """
 612   ld_params = objects.Disk.ComputeLDParams(template, disk_params)
 613
 614   if template == constants.DT_DRBD8:
 615     annotation_fn = _AnnotateDParamsDRBD
 616   elif template == constants.DT_DISKLESS:
 617     annotation_fn = lambda disk, _: disk
 618   else:
 619     annotation_fn = _AnnotateDParamsGeneric
 620
 621   new_disks = []
 622   for disk in disks:
 623     new_disks.append(annotation_fn(disk.Copy(), ld_params))
 624
 625   return new_disks
 626
 627
 628 #: Generic encoders
 629 _ENCODERS = {
 630   rpc_defs.ED_OBJECT_DICT: _ObjectToDict,
 631   rpc_defs.ED_OBJECT_DICT_LIST: _ObjectListToDict,
 632   rpc_defs.ED_NODE_TO_DISK_DICT: _EncodeNodeToDiskDict,
 633   rpc_defs.ED_COMPRESS: _Compress,
 634   rpc_defs.ED_FINALIZE_EXPORT_DISKS: _PrepareFinalizeExportDisks,
 635   rpc_defs.ED_IMPEXP_IO: _EncodeImportExportIO,
 636   rpc_defs.ED_BLOCKDEV_RENAME: _EncodeBlockdevRename,
 637   }
 638
 639
 640 class RpcRunner(_RpcClientBase,
 641                 _generated_rpc.RpcClientDefault,
 642                 _generated_rpc.RpcClientBootstrap,
 643                 _generated_rpc.RpcClientDnsOnly,
 644                 _generated_rpc.RpcClientConfig):
 645   """RPC runner class.
 646
 647   """
 648   def __init__(self, cfg, lock_monitor_cb, _req_process_fn=None, _getents=None):
 649     """Initialized the RPC runner.
 650
 651     @type cfg: L{config.ConfigWriter}
 652     @param cfg: Configuration
 653     @type lock_monitor_cb: callable
 654     @param lock_monitor_cb: Lock monitor callback
 655
 656     """
 657     self._cfg = cfg
 658
 659     encoders = _ENCODERS.copy()
 660
 661     encoders.update({
 662       # Encoders requiring configuration object
 663       rpc_defs.ED_INST_DICT: self._InstDict,
 664       rpc_defs.ED_INST_DICT_HVP_BEP_DP: self._InstDictHvpBepDp,
 665       rpc_defs.ED_INST_DICT_OSP_DP: self._InstDictOspDp,
 666
 667       # Encoders annotating disk parameters
 668       rpc_defs.ED_DISKS_DICT_DP: self._DisksDictDP,
 669       rpc_defs.ED_SINGLE_DISK_DICT_DP: self._SingleDiskDictDP,
 670
 671       # Encoders with special requirements
 672       rpc_defs.ED_FILE_DETAILS: compat.partial(_PrepareFileUpload, _getents),
 673       })
 674
 675     # Resolver using configuration
 676     resolver = compat.partial(_NodeConfigResolver, cfg.GetNodeInfo,
 677                               cfg.GetAllNodesInfo)
 678
 679     # Pylint doesn't recognize multiple inheritance properly, see
 680     # <http://www.logilab.org/ticket/36586> and
 681     # <http://www.logilab.org/ticket/35642>
 682     # pylint: disable=W0233
 683     _RpcClientBase.__init__(self, resolver, encoders.get,
 684                             lock_monitor_cb=lock_monitor_cb,
 685                             _req_process_fn=_req_process_fn)
 686     _generated_rpc.RpcClientConfig.__init__(self)
 687     _generated_rpc.RpcClientBootstrap.__init__(self)
 688     _generated_rpc.RpcClientDnsOnly.__init__(self)
 689     _generated_rpc.RpcClientDefault.__init__(self)
 690
 691   def _InstDict(self, instance, hvp=None, bep=None, osp=None):
 692     """Convert the given instance to a dict.
 693
 694     This is done via the instance's ToDict() method and additionally
 695     we fill the hvparams with the cluster defaults.
 696
 697     @type instance: L{objects.Instance}
 698     @param instance: an Instance object
 699     @type hvp: dict or None
 700     @param hvp: a dictionary with overridden hypervisor parameters
 701     @type bep: dict or None
 702     @param bep: a dictionary with overridden backend parameters
 703     @type osp: dict or None
 704     @param osp: a dictionary with overridden os parameters
 705     @rtype: dict
 706     @return: the instance dict, with the hvparams filled with the
 707         cluster defaults
 708
 709     """
 710     idict = instance.ToDict()
 711     cluster = self._cfg.GetClusterInfo()
 712     idict["hvparams"] = cluster.FillHV(instance)
 713     if hvp is not None:
 714       idict["hvparams"].update(hvp)
 715     idict["beparams"] = cluster.FillBE(instance)
 716     if bep is not None:
 717       idict["beparams"].update(bep)
 718     idict["osparams"] = cluster.SimpleFillOS(instance.os, instance.osparams)
 719     if osp is not None:
 720       idict["osparams"].update(osp)
 721     for nic in idict["nics"]:
 722       nic["nicparams"] = objects.FillDict(
 723         cluster.nicparams[constants.PP_DEFAULT],
 724         nic["nicparams"])
 725     idict["disks"] = self._DisksDictDP((instance.disks, instance))
 726     return idict
 727
 728   def _InstDictHvpBepDp(self, (instance, hvp, bep)):
 729     """Wrapper for L{_InstDict}.
 730
 731     """
 732     return self._InstDict(instance, hvp=hvp, bep=bep)
 733
 734   def _InstDictOspDp(self, (instance, osparams)):
 735     """Wrapper for L{_InstDict}.
 736
 737     """
 738     return self._InstDict(instance, osp=osparams)
 739
 740   def _DisksDictDP(self, (disks, instance)):
 741     """Wrapper for L{AnnotateDiskParams}.
 742
 743     """
 744     diskparams = self._cfg.GetInstanceDiskParams(instance)
 745     return [disk.ToDict()
 746             for disk in AnnotateDiskParams(instance.disk_template,
 747                                            disks, diskparams)]
 748
 749   def _SingleDiskDictDP(self, (disk, instance)):
 750     """Wrapper for L{AnnotateDiskParams}.
 751
 752     """
 753     (anno_disk,) = self._DisksDictDP(([disk], instance))
 754     return anno_disk
 755
 756
 757 class JobQueueRunner(_RpcClientBase, _generated_rpc.RpcClientJobQueue):
 758   """RPC wrappers for job queue.
 759
 760   """
 761   def __init__(self, context, address_list):
 762     """Initializes this class.
 763
 764     """
 765     if address_list is None:
 766       resolver = compat.partial(_SsconfResolver, True)
 767     else:
 768       # Caller provided an address list
 769       resolver = _StaticResolver(address_list)
 770
 771     _RpcClientBase.__init__(self, resolver, _ENCODERS.get,
 772                             lock_monitor_cb=context.glm.AddToLockMonitor)
 773     _generated_rpc.RpcClientJobQueue.__init__(self)
 774
 775
 776 class BootstrapRunner(_RpcClientBase,
 777                       _generated_rpc.RpcClientBootstrap,
 778                       _generated_rpc.RpcClientDnsOnly):
 779   """RPC wrappers for bootstrapping.
 780
 781   """
 782   def __init__(self):
 783     """Initializes this class.
 784
 785     """
 786     # Pylint doesn't recognize multiple inheritance properly, see
 787     # <http://www.logilab.org/ticket/36586> and
 788     # <http://www.logilab.org/ticket/35642>
 789     # pylint: disable=W0233
 790     _RpcClientBase.__init__(self, compat.partial(_SsconfResolver, True),
 791                             _ENCODERS.get)
 792     _generated_rpc.RpcClientBootstrap.__init__(self)
 793     _generated_rpc.RpcClientDnsOnly.__init__(self)
 794
 795
 796 class DnsOnlyRunner(_RpcClientBase, _generated_rpc.RpcClientDnsOnly):
 797   """RPC wrappers for calls using only DNS.
 798
 799   """
 800   def __init__(self):
 801     """Initialize this class.
 802
 803     """
 804     _RpcClientBase.__init__(self, compat.partial(_SsconfResolver, False),
 805                             _ENCODERS.get)
 806     _generated_rpc.RpcClientDnsOnly.__init__(self)
 807
 808
 809 class ConfigRunner(_RpcClientBase, _generated_rpc.RpcClientConfig):
 810   """RPC wrappers for L{config}.
 811
 812   """
 813   def __init__(self, context, address_list, _req_process_fn=None,
 814                _getents=None):
 815     """Initializes this class.
 816
 817     """
 818     if context:
 819       lock_monitor_cb = context.glm.AddToLockMonitor
 820     else:
 821       lock_monitor_cb = None
 822
 823     if address_list is None:
 824       resolver = compat.partial(_SsconfResolver, True)
 825     else:
 826       # Caller provided an address list
 827       resolver = _StaticResolver(address_list)
 828
 829     encoders = _ENCODERS.copy()
 830
 831     encoders.update({
 832       rpc_defs.ED_FILE_DETAILS: compat.partial(_PrepareFileUpload, _getents),
 833       })
 834
 835     _RpcClientBase.__init__(self, resolver, encoders.get,
 836                             lock_monitor_cb=lock_monitor_cb,
 837                             _req_process_fn=_req_process_fn)
 838     _generated_rpc.RpcClientConfig.__init__(self)