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

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 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 import copy
  39
  40 from ganeti import utils
  41 from ganeti import objects
  42 from ganeti import http
  43 from ganeti import serializer
  44 from ganeti import constants
  45 from ganeti import errors
  46 from ganeti import netutils
  47 from ganeti import ssconf
  48 from ganeti import runtime
  49 from ganeti import compat
  50 from ganeti import rpc_defs
  51 from ganeti import pathutils
  52 from ganeti import vcluster
  53
  54 # Special module generated at build time
  55 from ganeti import _generated_rpc
  56
  57 # pylint has a bug here, doesn't see this import
  58 import ganeti.http.client  # pylint: disable=W0611
  59
  60
  61 _RPC_CLIENT_HEADERS = [
  62   "Content-type: %s" % http.HTTP_APP_JSON,
  63   "Expect:",
  64   ]
  65
  66 #: Special value to describe an offline host
  67 _OFFLINE = object()
  68
  69
  70 def Init():
  71   """Initializes the module-global HTTP client manager.
  72
  73   Must be called before using any RPC function and while exactly one thread is
  74   running.
  75
  76   """
  77   # curl_global_init(3) and curl_global_cleanup(3) must be called with only
  78   # one thread running. This check is just a safety measure -- it doesn't
  79   # cover all cases.
  80   assert threading.activeCount() == 1, \
  81          "Found more than one active thread when initializing pycURL"
  82
  83   logging.info("Using PycURL %s", pycurl.version)
  84
  85   pycurl.global_init(pycurl.GLOBAL_ALL)
  86
  87
  88 def Shutdown():
  89   """Stops the module-global HTTP client manager.
  90
  91   Must be called before quitting the program and while exactly one thread is
  92   running.
  93
  94   """
  95   pycurl.global_cleanup()
  96
  97
  98 def _ConfigRpcCurl(curl):
  99   noded_cert = str(pathutils.NODED_CERT_FILE)
 100
 101   curl.setopt(pycurl.FOLLOWLOCATION, False)
 102   curl.setopt(pycurl.CAINFO, noded_cert)
 103   curl.setopt(pycurl.SSL_VERIFYHOST, 0)
 104   curl.setopt(pycurl.SSL_VERIFYPEER, True)
 105   curl.setopt(pycurl.SSLCERTTYPE, "PEM")
 106   curl.setopt(pycurl.SSLCERT, noded_cert)
 107   curl.setopt(pycurl.SSLKEYTYPE, "PEM")
 108   curl.setopt(pycurl.SSLKEY, noded_cert)
 109   curl.setopt(pycurl.CONNECTTIMEOUT, constants.RPC_CONNECT_TIMEOUT)
 110
 111
 112 def RunWithRPC(fn):
 113   """RPC-wrapper decorator.
 114
 115   When applied to a function, it runs it with the RPC system
 116   initialized, and it shutsdown the system afterwards. This means the
 117   function must be called without RPC being initialized.
 118
 119   """
 120   def wrapper(*args, **kwargs):
 121     Init()
 122     try:
 123       return fn(*args, **kwargs)
 124     finally:
 125       Shutdown()
 126   return wrapper
 127
 128
 129 def _Compress(_, data):
 130   """Compresses a string for transport over RPC.
 131
 132   Small amounts of data are not compressed.
 133
 134   @type data: str
 135   @param data: Data
 136   @rtype: tuple
 137   @return: Encoded data to send
 138
 139   """
 140   # Small amounts of data are not compressed
 141   if len(data) < 512:
 142     return (constants.RPC_ENCODING_NONE, data)
 143
 144   # Compress with zlib and encode in base64
 145   return (constants.RPC_ENCODING_ZLIB_BASE64,
 146           base64.b64encode(zlib.compress(data, 3)))
 147
 148
 149 class RpcResult(object):
 150   """RPC Result class.
 151
 152   This class holds an RPC result. It is needed since in multi-node
 153   calls we can't raise an exception just because one out of many
 154   failed, and therefore we use this class to encapsulate the result.
 155
 156   @ivar data: the data payload, for successful results, or None
 157   @ivar call: the name of the RPC call
 158   @ivar node: the name of the node to which we made the call
 159   @ivar offline: whether the operation failed because the node was
 160       offline, as opposed to actual failure; offline=True will always
 161       imply failed=True, in order to allow simpler checking if
 162       the user doesn't care about the exact failure mode
 163   @ivar fail_msg: the error message if the call failed
 164
 165   """
 166   def __init__(self, data=None, failed=False, offline=False,
 167                call=None, node=None):
 168     self.offline = offline
 169     self.call = call
 170     self.node = node
 171
 172     if offline:
 173       self.fail_msg = "Node is marked offline"
 174       self.data = self.payload = None
 175     elif failed:
 176       self.fail_msg = self._EnsureErr(data)
 177       self.data = self.payload = None
 178     else:
 179       self.data = data
 180       if not isinstance(self.data, (tuple, list)):
 181         self.fail_msg = ("RPC layer error: invalid result type (%s)" %
 182                          type(self.data))
 183         self.payload = None
 184       elif len(data) != 2:
 185         self.fail_msg = ("RPC layer error: invalid result length (%d), "
 186                          "expected 2" % len(self.data))
 187         self.payload = None
 188       elif not self.data[0]:
 189         self.fail_msg = self._EnsureErr(self.data[1])
 190         self.payload = None
 191       else:
 192         # finally success
 193         self.fail_msg = None
 194         self.payload = data[1]
 195
 196     for attr_name in ["call", "data", "fail_msg",
 197                       "node", "offline", "payload"]:
 198       assert hasattr(self, attr_name), "Missing attribute %s" % attr_name
 199
 200   @staticmethod
 201   def _EnsureErr(val):
 202     """Helper to ensure we return a 'True' value for error."""
 203     if val:
 204       return val
 205     else:
 206       return "No error information"
 207
 208   def Raise(self, msg, prereq=False, ecode=None):
 209     """If the result has failed, raise an OpExecError.
 210
 211     This is used so that LU code doesn't have to check for each
 212     result, but instead can call this function.
 213
 214     """
 215     if not self.fail_msg:
 216       return
 217
 218     if not msg: # one could pass None for default message
 219       msg = ("Call '%s' to node '%s' has failed: %s" %
 220              (self.call, self.node, self.fail_msg))
 221     else:
 222       msg = "%s: %s" % (msg, self.fail_msg)
 223     if prereq:
 224       ec = errors.OpPrereqError
 225     else:
 226       ec = errors.OpExecError
 227     if ecode is not None:
 228       args = (msg, ecode)
 229     else:
 230       args = (msg, )
 231     raise ec(*args) # pylint: disable=W0142
 232
 233   def Warn(self, msg, feedback_fn):
 234     """If the result has failed, call the feedback_fn.
 235
 236     This is used to in cases were LU wants to warn the
 237     user about a failure, but continue anyway.
 238
 239     """
 240     if not self.fail_msg:
 241       return
 242
 243     msg = "%s: %s" % (msg, self.fail_msg)
 244     feedback_fn(msg)
 245
 246
 247 def _SsconfResolver(ssconf_ips, node_list, _,
 248                     ssc=ssconf.SimpleStore,
 249                     nslookup_fn=netutils.Hostname.GetIP):
 250   """Return addresses for given node names.
 251
 252   @type ssconf_ips: bool
 253   @param ssconf_ips: Use the ssconf IPs
 254   @type node_list: list
 255   @param node_list: List of node names
 256   @type ssc: class
 257   @param ssc: SimpleStore class that is used to obtain node->ip mappings
 258   @type nslookup_fn: callable
 259   @param nslookup_fn: function use to do NS lookup
 260   @rtype: list of tuple; (string, string)
 261   @return: List of tuples containing node name and IP address
 262
 263   """
 264   ss = ssc()
 265   family = ss.GetPrimaryIPFamily()
 266
 267   if ssconf_ips:
 268     iplist = ss.GetNodePrimaryIPList()
 269     ipmap = dict(entry.split() for entry in iplist)
 270   else:
 271     ipmap = {}
 272
 273   result = []
 274   for node in node_list:
 275     ip = ipmap.get(node)
 276     if ip is None:
 277       ip = nslookup_fn(node, family=family)
 278     result.append((node, ip, node))
 279
 280   return result
 281
 282
 283 class _StaticResolver:
 284   def __init__(self, addresses):
 285     """Initializes this class.
 286
 287     """
 288     self._addresses = addresses
 289
 290   def __call__(self, hosts, _):
 291     """Returns static addresses for hosts.
 292
 293     """
 294     assert len(hosts) == len(self._addresses)
 295     return zip(hosts, self._addresses, hosts)
 296
 297
 298 def _CheckConfigNode(node_uuid_or_name, node, accept_offline_node):
 299   """Checks if a node is online.
 300
 301   @type node_uuid_or_name: string
 302   @param node_uuid_or_name: Node UUID
 303   @type node: L{objects.Node} or None
 304   @param node: Node object
 305
 306   """
 307   if node is None:
 308     # Assume that the passed parameter was actually a node name, so depend on
 309     # DNS for name resolution
 310     return (node_uuid_or_name, node_uuid_or_name, node_uuid_or_name)
 311   else:
 312     if node.offline and not accept_offline_node:
 313       ip = _OFFLINE
 314     else:
 315       ip = node.primary_ip
 316     return (node.name, ip, node_uuid_or_name)
 317
 318
 319 def _NodeConfigResolver(single_node_fn, all_nodes_fn, node_uuids, opts):
 320   """Calculate node addresses using configuration.
 321
 322   Note that strings in node_uuids are treated as node names if the UUID is not
 323   found in the configuration.
 324
 325   """
 326   accept_offline_node = (opts is rpc_defs.ACCEPT_OFFLINE_NODE)
 327
 328   assert accept_offline_node or opts is None, "Unknown option"
 329
 330   # Special case for single-host lookups
 331   if len(node_uuids) == 1:
 332     (uuid, ) = node_uuids
 333     return [_CheckConfigNode(uuid, single_node_fn(uuid), accept_offline_node)]
 334   else:
 335     all_nodes = all_nodes_fn()
 336     return [_CheckConfigNode(uuid, all_nodes.get(uuid, None),
 337                              accept_offline_node)
 338             for uuid in node_uuids]
 339
 340
 341 class _RpcProcessor:
 342   def __init__(self, resolver, port, lock_monitor_cb=None):
 343     """Initializes this class.
 344
 345     @param resolver: callable accepting a list of node UUIDs or hostnames,
 346       returning a list of tuples containing name, IP address and original name
 347       of the resolved node. IP address can be the name or the special value
 348       L{_OFFLINE} to mark offline machines.
 349     @type port: int
 350     @param port: TCP port
 351     @param lock_monitor_cb: Callable for registering with lock monitor
 352
 353     """
 354     self._resolver = resolver
 355     self._port = port
 356     self._lock_monitor_cb = lock_monitor_cb
 357
 358   @staticmethod
 359   def _PrepareRequests(hosts, port, procedure, body, read_timeout):
 360     """Prepares requests by sorting offline hosts into separate list.
 361
 362     @type body: dict
 363     @param body: a dictionary with per-host body data
 364
 365     """
 366     results = {}
 367     requests = {}
 368
 369     assert isinstance(body, dict)
 370     assert len(body) == len(hosts)
 371     assert compat.all(isinstance(v, str) for v in body.values())
 372     assert frozenset(map(lambda x: x[2], hosts)) == frozenset(body.keys()), \
 373         "%s != %s" % (hosts, body.keys())
 374
 375     for (name, ip, original_name) in hosts:
 376       if ip is _OFFLINE:
 377         # Node is marked as offline
 378         results[original_name] = RpcResult(node=name,
 379                                            offline=True,
 380                                            call=procedure)
 381       else:
 382         requests[original_name] = \
 383           http.client.HttpClientRequest(str(ip), port,
 384                                         http.HTTP_POST, str("/%s" % procedure),
 385                                         headers=_RPC_CLIENT_HEADERS,
 386                                         post_data=body[original_name],
 387                                         read_timeout=read_timeout,
 388                                         nicename="%s/%s" % (name, procedure),
 389                                         curl_config_fn=_ConfigRpcCurl)
 390
 391     return (results, requests)
 392
 393   @staticmethod
 394   def _CombineResults(results, requests, procedure):
 395     """Combines pre-computed results for offline hosts with actual call results.
 396
 397     """
 398     for name, req in requests.items():
 399       if req.success and req.resp_status_code == http.HTTP_OK:
 400         host_result = RpcResult(data=serializer.LoadJson(req.resp_body),
 401                                 node=name, call=procedure)
 402       else:
 403         # TODO: Better error reporting
 404         if req.error:
 405           msg = req.error
 406         else:
 407           msg = req.resp_body
 408
 409         logging.error("RPC error in %s on node %s: %s", procedure, name, msg)
 410         host_result = RpcResult(data=msg, failed=True, node=name,
 411                                 call=procedure)
 412
 413       results[name] = host_result
 414
 415     return results
 416
 417   def __call__(self, nodes, procedure, body, read_timeout, resolver_opts,
 418                _req_process_fn=None):
 419     """Makes an RPC request to a number of nodes.
 420
 421     @type nodes: sequence
 422     @param nodes: node UUIDs or Hostnames
 423     @type procedure: string
 424     @param procedure: Request path
 425     @type body: dictionary
 426     @param body: dictionary with request bodies per host
 427     @type read_timeout: int or None
 428     @param read_timeout: Read timeout for request
 429     @rtype: dictionary
 430     @return: a dictionary mapping host names to rpc.RpcResult objects
 431
 432     """
 433     assert read_timeout is not None, \
 434       "Missing RPC read timeout for procedure '%s'" % procedure
 435
 436     if _req_process_fn is None:
 437       _req_process_fn = http.client.ProcessRequests
 438
 439     (results, requests) = \
 440       self._PrepareRequests(self._resolver(nodes, resolver_opts), self._port,
 441                             procedure, body, read_timeout)
 442
 443     _req_process_fn(requests.values(), lock_monitor_cb=self._lock_monitor_cb)
 444
 445     assert not frozenset(results).intersection(requests)
 446
 447     return self._CombineResults(results, requests, procedure)
 448
 449
 450 class _RpcClientBase:
 451   def __init__(self, resolver, encoder_fn, lock_monitor_cb=None,
 452                _req_process_fn=None):
 453     """Initializes this class.
 454
 455     """
 456     proc = _RpcProcessor(resolver,
 457                          netutils.GetDaemonPort(constants.NODED),
 458                          lock_monitor_cb=lock_monitor_cb)
 459     self._proc = compat.partial(proc, _req_process_fn=_req_process_fn)
 460     self._encoder = compat.partial(self._EncodeArg, encoder_fn)
 461
 462   @staticmethod
 463   def _EncodeArg(encoder_fn, node, (argkind, value)):
 464     """Encode argument.
 465
 466     """
 467     if argkind is None:
 468       return value
 469     else:
 470       return encoder_fn(argkind)(node, value)
 471
 472   def _Call(self, cdef, node_list, args):
 473     """Entry point for automatically generated RPC wrappers.
 474
 475     """
 476     (procedure, _, resolver_opts, timeout, argdefs,
 477      prep_fn, postproc_fn, _) = cdef
 478
 479     if callable(timeout):
 480       read_timeout = timeout(args)
 481     else:
 482       read_timeout = timeout
 483
 484     if callable(resolver_opts):
 485       req_resolver_opts = resolver_opts(args)
 486     else:
 487       req_resolver_opts = resolver_opts
 488
 489     if len(args) != len(argdefs):
 490       raise errors.ProgrammerError("Number of passed arguments doesn't match")
 491
 492     if prep_fn is None:
 493       prep_fn = lambda _, args: args
 494     assert callable(prep_fn)
 495
 496     # encode the arguments for each node individually, pass them and the node
 497     # name to the prep_fn, and serialise its return value
 498     encode_args_fn = lambda node: map(compat.partial(self._encoder, node),
 499                                       zip(map(compat.snd, argdefs), args))
 500     pnbody = dict((n, serializer.DumpJson(prep_fn(n, encode_args_fn(n))))
 501                   for n in node_list)
 502
 503     result = self._proc(node_list, procedure, pnbody, read_timeout,
 504                         req_resolver_opts)
 505
 506     if postproc_fn:
 507       return dict(map(lambda (key, value): (key, postproc_fn(value)),
 508                       result.items()))
 509     else:
 510       return result
 511
 512
 513 def _ObjectToDict(_, value):
 514   """Converts an object to a dictionary.
 515
 516   @note: See L{objects}.
 517
 518   """
 519   return value.ToDict()
 520
 521
 522 def _ObjectListToDict(node, value):
 523   """Converts a list of L{objects} to dictionaries.
 524
 525   """
 526   return map(compat.partial(_ObjectToDict, node), value)
 527
 528
 529 def _PrepareFileUpload(getents_fn, node, filename):
 530   """Loads a file and prepares it for an upload to nodes.
 531
 532   """
 533   statcb = utils.FileStatHelper()
 534   data = _Compress(node, utils.ReadFile(filename, preread=statcb))
 535   st = statcb.st
 536
 537   if getents_fn is None:
 538     getents_fn = runtime.GetEnts
 539
 540   getents = getents_fn()
 541
 542   virt_filename = vcluster.MakeVirtualPath(filename)
 543
 544   return [virt_filename, data, st.st_mode, getents.LookupUid(st.st_uid),
 545           getents.LookupGid(st.st_gid), st.st_atime, st.st_mtime]
 546
 547
 548 def _PrepareFinalizeExportDisks(_, snap_disks):
 549   """Encodes disks for finalizing export.
 550
 551   """
 552   flat_disks = []
 553
 554   for disk in snap_disks:
 555     if isinstance(disk, bool):
 556       flat_disks.append(disk)
 557     else:
 558       flat_disks.append(disk.ToDict())
 559
 560   return flat_disks
 561
 562
 563 def _EncodeBlockdevRename(_, value):
 564   """Encodes information for renaming block devices.
 565
 566   """
 567   return [(d.ToDict(), uid) for d, uid in value]
 568
 569
 570 def _AddSpindlesToLegacyNodeInfo(result, space_info):
 571   """Extracts the spindle information from the space info and adds
 572   it to the result dictionary.
 573
 574   @type result: dict of strings
 575   @param result: dictionary holding the result of the legacy node info
 576   @type space_info: list of dicts of strings
 577   @param space_info: list, each row holding space information of one storage
 578     unit
 579   @rtype: None
 580   @return: does not return anything, manipulates the C{result} variable
 581
 582   """
 583   lvm_pv_info = utils.storage.LookupSpaceInfoByStorageType(
 584       space_info, constants.ST_LVM_PV)
 585   if lvm_pv_info:
 586     result["spindles_free"] = lvm_pv_info["storage_free"]
 587     result["spindles_total"] = lvm_pv_info["storage_size"]
 588   else:
 589     raise errors.OpExecError("No spindle storage information available.")
 590
 591
 592 def _AddDefaultStorageInfoToLegacyNodeInfo(result, space_info):
 593   """Extracts the storage space information of the default storage type from
 594   the space info and adds it to the result dictionary.
 595
 596   @see: C{_AddSpindlesToLegacyNodeInfo} for parameter information.
 597
 598   """
 599   # Check if there is at least one row for non-spindle storage info.
 600   no_defaults = (len(space_info) < 1) or \
 601       (space_info[0]["type"] == constants.ST_LVM_PV and len(space_info) == 1)
 602
 603   default_space_info = None
 604   if no_defaults:
 605     logging.warning("No storage info provided for default storage type.")
 606   else:
 607     default_space_info = space_info[0]
 608
 609   if default_space_info:
 610     result["name"] = default_space_info["name"]
 611     result["storage_free"] = default_space_info["storage_free"]
 612     result["storage_size"] = default_space_info["storage_size"]
 613
 614
 615 def MakeLegacyNodeInfo(data, require_spindles=False):
 616   """Formats the data returned by L{rpc.RpcRunner.call_node_info}.
 617
 618   Converts the data into a single dictionary. This is fine for most use cases,
 619   but some require information from more than one volume group or hypervisor.
 620
 621   @param require_spindles: add spindle storage information to the legacy node
 622       info
 623
 624   """
 625   (bootid, space_info, (hv_info, )) = data
 626
 627   ret = utils.JoinDisjointDicts(hv_info, {"bootid": bootid})
 628
 629   if require_spindles:
 630     _AddSpindlesToLegacyNodeInfo(ret, space_info)
 631   _AddDefaultStorageInfoToLegacyNodeInfo(ret, space_info)
 632
 633   return ret
 634
 635
 636 def _AnnotateDParamsDRBD(disk, (drbd_params, data_params, meta_params)):
 637   """Annotates just DRBD disks layouts.
 638
 639   """
 640   assert disk.dev_type == constants.DT_DRBD8
 641
 642   disk.params = objects.FillDict(drbd_params, disk.params)
 643   (dev_data, dev_meta) = disk.children
 644   dev_data.params = objects.FillDict(data_params, dev_data.params)
 645   dev_meta.params = objects.FillDict(meta_params, dev_meta.params)
 646
 647   return disk
 648
 649
 650 def _AnnotateDParamsGeneric(disk, (params, )):
 651   """Generic disk parameter annotation routine.
 652
 653   """
 654   assert disk.dev_type != constants.DT_DRBD8
 655
 656   disk.params = objects.FillDict(params, disk.params)
 657
 658   return disk
 659
 660
 661 def AnnotateDiskParams(disks, disk_params):
 662   """Annotates the disk objects with the disk parameters.
 663
 664   @param disks: The list of disks objects to annotate
 665   @param disk_params: The disk parameters for annotation
 666   @returns: A list of disk objects annotated
 667
 668   """
 669   def AnnotateDisk(disk):
 670     if disk.dev_type == constants.DT_DISKLESS:
 671       return disk
 672
 673     ld_params = objects.Disk.ComputeLDParams(disk.dev_type, disk_params)
 674
 675     if disk.dev_type == constants.DT_DRBD8:
 676       return _AnnotateDParamsDRBD(disk, ld_params)
 677     else:
 678       return _AnnotateDParamsGeneric(disk, ld_params)
 679
 680   return [AnnotateDisk(disk.Copy()) for disk in disks]
 681
 682
 683 def _GetExclusiveStorageFlag(cfg, node_uuid):
 684   ni = cfg.GetNodeInfo(node_uuid)
 685   if ni is None:
 686     raise errors.OpPrereqError("Invalid node name %s" % node_uuid,
 687                                errors.ECODE_NOENT)
 688   return cfg.GetNdParams(ni)[constants.ND_EXCLUSIVE_STORAGE]
 689
 690
 691 def _AddExclusiveStorageFlagToLvmStorageUnits(storage_units, es_flag):
 692   """Adds the exclusive storage flag to lvm units.
 693
 694   This function creates a copy of the storage_units lists, with the
 695   es_flag being added to all lvm storage units.
 696
 697   @type storage_units: list of pairs (string, string)
 698   @param storage_units: list of 'raw' storage units, consisting only of
 699     (storage_type, storage_key)
 700   @type es_flag: boolean
 701   @param es_flag: exclusive storage flag
 702   @rtype: list of tuples (string, string, list)
 703   @return: list of storage units (storage_type, storage_key, params) with
 704     the params containing the es_flag for lvm-vg storage units
 705
 706   """
 707   result = []
 708   for (storage_type, storage_key) in storage_units:
 709     if storage_type in [constants.ST_LVM_VG, constants.ST_LVM_PV]:
 710       result.append((storage_type, storage_key, [es_flag]))
 711     else:
 712       result.append((storage_type, storage_key, []))
 713   return result
 714
 715
 716 def GetExclusiveStorageForNodes(cfg, node_uuids):
 717   """Return the exclusive storage flag for all the given nodes.
 718
 719   @type cfg: L{config.ConfigWriter}
 720   @param cfg: cluster configuration
 721   @type node_uuids: list or tuple
 722   @param node_uuids: node UUIDs for which to read the flag
 723   @rtype: dict
 724   @return: mapping from node uuids to exclusive storage flags
 725   @raise errors.OpPrereqError: if any given node name has no corresponding
 726   node
 727
 728   """
 729   getflag = lambda n: _GetExclusiveStorageFlag(cfg, n)
 730   flags = map(getflag, node_uuids)
 731   return dict(zip(node_uuids, flags))
 732
 733
 734 def PrepareStorageUnitsForNodes(cfg, storage_units, node_uuids):
 735   """Return the lvm storage unit for all the given nodes.
 736
 737   Main purpose of this function is to map the exclusive storage flag, which
 738   can be different for each node, to the default LVM storage unit.
 739
 740   @type cfg: L{config.ConfigWriter}
 741   @param cfg: cluster configuration
 742   @type storage_units: list of pairs (string, string)
 743   @param storage_units: list of 'raw' storage units, e.g. pairs of
 744     (storage_type, storage_key)
 745   @type node_uuids: list or tuple
 746   @param node_uuids: node UUIDs for which to read the flag
 747   @rtype: dict
 748   @return: mapping from node uuids to a list of storage units which include
 749     the exclusive storage flag for lvm storage
 750   @raise errors.OpPrereqError: if any given node name has no corresponding
 751   node
 752
 753   """
 754   getunit = lambda n: _AddExclusiveStorageFlagToLvmStorageUnits(
 755       storage_units, _GetExclusiveStorageFlag(cfg, n))
 756   flags = map(getunit, node_uuids)
 757   return dict(zip(node_uuids, flags))
 758
 759
 760 #: Generic encoders
 761 _ENCODERS = {
 762   rpc_defs.ED_OBJECT_DICT: _ObjectToDict,
 763   rpc_defs.ED_OBJECT_DICT_LIST: _ObjectListToDict,
 764   rpc_defs.ED_COMPRESS: _Compress,
 765   rpc_defs.ED_FINALIZE_EXPORT_DISKS: _PrepareFinalizeExportDisks,
 766   rpc_defs.ED_BLOCKDEV_RENAME: _EncodeBlockdevRename,
 767   }
 768
 769
 770 class RpcRunner(_RpcClientBase,
 771                 _generated_rpc.RpcClientDefault,
 772                 _generated_rpc.RpcClientBootstrap,
 773                 _generated_rpc.RpcClientDnsOnly,
 774                 _generated_rpc.RpcClientConfig):
 775   """RPC runner class.
 776
 777   """
 778   def __init__(self, cfg, lock_monitor_cb, _req_process_fn=None, _getents=None):
 779     """Initialized the RPC runner.
 780
 781     @type cfg: L{config.ConfigWriter}
 782     @param cfg: Configuration
 783     @type lock_monitor_cb: callable
 784     @param lock_monitor_cb: Lock monitor callback
 785
 786     """
 787     self._cfg = cfg
 788
 789     encoders = _ENCODERS.copy()
 790
 791     encoders.update({
 792       # Encoders requiring configuration object
 793       rpc_defs.ED_INST_DICT: self._InstDict,
 794       rpc_defs.ED_INST_DICT_HVP_BEP_DP: self._InstDictHvpBepDp,
 795       rpc_defs.ED_INST_DICT_OSP_DP: self._InstDictOspDp,
 796       rpc_defs.ED_NIC_DICT: self._NicDict,
 797
 798       # Encoders annotating disk parameters
 799       rpc_defs.ED_DISKS_DICT_DP: self._DisksDictDP,
 800       rpc_defs.ED_MULTI_DISKS_DICT_DP: self._MultiDiskDictDP,
 801       rpc_defs.ED_SINGLE_DISK_DICT_DP: self._SingleDiskDictDP,
 802       rpc_defs.ED_NODE_TO_DISK_DICT_DP: self._EncodeNodeToDiskDictDP,
 803
 804       # Encoders with special requirements
 805       rpc_defs.ED_FILE_DETAILS: compat.partial(_PrepareFileUpload, _getents),
 806
 807       rpc_defs.ED_IMPEXP_IO: self._EncodeImportExportIO,
 808       })
 809
 810     # Resolver using configuration
 811     resolver = compat.partial(_NodeConfigResolver, cfg.GetNodeInfo,
 812                               cfg.GetAllNodesInfo)
 813
 814     # Pylint doesn't recognize multiple inheritance properly, see
 815     # <http://www.logilab.org/ticket/36586> and
 816     # <http://www.logilab.org/ticket/35642>
 817     # pylint: disable=W0233
 818     _RpcClientBase.__init__(self, resolver, encoders.get,
 819                             lock_monitor_cb=lock_monitor_cb,
 820                             _req_process_fn=_req_process_fn)
 821     _generated_rpc.RpcClientConfig.__init__(self)
 822     _generated_rpc.RpcClientBootstrap.__init__(self)
 823     _generated_rpc.RpcClientDnsOnly.__init__(self)
 824     _generated_rpc.RpcClientDefault.__init__(self)
 825
 826   def _NicDict(self, _, nic):
 827     """Convert the given nic to a dict and encapsulate netinfo
 828
 829     """
 830     n = copy.deepcopy(nic)
 831     if n.network:
 832       net_uuid = self._cfg.LookupNetwork(n.network)
 833       if net_uuid:
 834         nobj = self._cfg.GetNetwork(net_uuid)
 835         n.netinfo = objects.Network.ToDict(nobj)
 836     return n.ToDict()
 837
 838   def _InstDict(self, node, instance, hvp=None, bep=None, osp=None):
 839     """Convert the given instance to a dict.
 840
 841     This is done via the instance's ToDict() method and additionally
 842     we fill the hvparams with the cluster defaults.
 843
 844     @type instance: L{objects.Instance}
 845     @param instance: an Instance object
 846     @type hvp: dict or None
 847     @param hvp: a dictionary with overridden hypervisor parameters
 848     @type bep: dict or None
 849     @param bep: a dictionary with overridden backend parameters
 850     @type osp: dict or None
 851     @param osp: a dictionary with overridden os parameters
 852     @rtype: dict
 853     @return: the instance dict, with the hvparams filled with the
 854         cluster defaults
 855
 856     """
 857     idict = instance.ToDict()
 858     cluster = self._cfg.GetClusterInfo()
 859     idict["hvparams"] = cluster.FillHV(instance)
 860     if hvp is not None:
 861       idict["hvparams"].update(hvp)
 862     idict["beparams"] = cluster.FillBE(instance)
 863     if bep is not None:
 864       idict["beparams"].update(bep)
 865     idict["osparams"] = cluster.SimpleFillOS(instance.os, instance.osparams)
 866     if osp is not None:
 867       idict["osparams"].update(osp)
 868     idict["disks"] = self._DisksDictDP(node, (instance.disks, instance))
 869     for nic in idict["nics"]:
 870       nic["nicparams"] = objects.FillDict(
 871         cluster.nicparams[constants.PP_DEFAULT],
 872         nic["nicparams"])
 873       network = nic.get("network", None)
 874       if network:
 875         net_uuid = self._cfg.LookupNetwork(network)
 876         if net_uuid:
 877           nobj = self._cfg.GetNetwork(net_uuid)
 878           nic["netinfo"] = objects.Network.ToDict(nobj)
 879     return idict
 880
 881   def _InstDictHvpBepDp(self, node, (instance, hvp, bep)):
 882     """Wrapper for L{_InstDict}.
 883
 884     """
 885     return self._InstDict(node, instance, hvp=hvp, bep=bep)
 886
 887   def _InstDictOspDp(self, node, (instance, osparams)):
 888     """Wrapper for L{_InstDict}.
 889
 890     """
 891     return self._InstDict(node, instance, osp=osparams)
 892
 893   def _DisksDictDP(self, node, (disks, instance)):
 894     """Wrapper for L{AnnotateDiskParams}.
 895
 896     """
 897     diskparams = self._cfg.GetInstanceDiskParams(instance)
 898     ret = []
 899     for disk in AnnotateDiskParams(disks, diskparams):
 900       disk_node_uuids = disk.GetNodes(instance.primary_node)
 901       node_ips = dict((uuid, node.secondary_ip) for (uuid, node)
 902                       in self._cfg.GetMultiNodeInfo(disk_node_uuids))
 903
 904       disk.UpdateDynamicDiskParams(node, node_ips)
 905
 906       ret.append(disk.ToDict(include_dynamic_params=True))
 907
 908     return ret
 909
 910   def _MultiDiskDictDP(self, node, disks_insts):
 911     """Wrapper for L{AnnotateDiskParams}.
 912
 913     Supports a list of (disk, instance) tuples.
 914     """
 915     return [disk for disk_inst in disks_insts
 916             for disk in self._DisksDictDP(node, disk_inst)]
 917
 918   def _SingleDiskDictDP(self, node, (disk, instance)):
 919     """Wrapper for L{AnnotateDiskParams}.
 920
 921     """
 922     (anno_disk,) = self._DisksDictDP(node, ([disk], instance))
 923     return anno_disk
 924
 925   def _EncodeNodeToDiskDictDP(self, node, value):
 926     """Encode dict of node name -> list of (disk, instance) tuples as values.
 927
 928     """
 929     return dict((name, [self._SingleDiskDictDP(node, disk) for disk in disks])
 930                 for name, disks in value.items())
 931
 932   def _EncodeImportExportIO(self, node, (ieio, ieioargs)):
 933     """Encodes import/export I/O information.
 934
 935     """
 936     if ieio == constants.IEIO_RAW_DISK:
 937       assert len(ieioargs) == 1
 938       return (ieio, (self._SingleDiskDictDP(node, ieioargs[0]), ))
 939
 940     if ieio == constants.IEIO_SCRIPT:
 941       assert len(ieioargs) == 2
 942       return (ieio, (self._SingleDiskDictDP(node, ieioargs[0]), ieioargs[1]))
 943
 944     return (ieio, ieioargs)
 945
 946
 947 class JobQueueRunner(_RpcClientBase, _generated_rpc.RpcClientJobQueue):
 948   """RPC wrappers for job queue.
 949
 950   """
 951   def __init__(self, context, address_list):
 952     """Initializes this class.
 953
 954     """
 955     if address_list is None:
 956       resolver = compat.partial(_SsconfResolver, True)
 957     else:
 958       # Caller provided an address list
 959       resolver = _StaticResolver(address_list)
 960
 961     _RpcClientBase.__init__(self, resolver, _ENCODERS.get,
 962                             lock_monitor_cb=context.glm.AddToLockMonitor)
 963     _generated_rpc.RpcClientJobQueue.__init__(self)
 964
 965
 966 class BootstrapRunner(_RpcClientBase,
 967                       _generated_rpc.RpcClientBootstrap,
 968                       _generated_rpc.RpcClientDnsOnly):
 969   """RPC wrappers for bootstrapping.
 970
 971   """
 972   def __init__(self):
 973     """Initializes this class.
 974
 975     """
 976     # Pylint doesn't recognize multiple inheritance properly, see
 977     # <http://www.logilab.org/ticket/36586> and
 978     # <http://www.logilab.org/ticket/35642>
 979     # pylint: disable=W0233
 980     _RpcClientBase.__init__(self, compat.partial(_SsconfResolver, True),
 981                             _ENCODERS.get)
 982     _generated_rpc.RpcClientBootstrap.__init__(self)
 983     _generated_rpc.RpcClientDnsOnly.__init__(self)
 984
 985
 986 class DnsOnlyRunner(_RpcClientBase, _generated_rpc.RpcClientDnsOnly):
 987   """RPC wrappers for calls using only DNS.
 988
 989   """
 990   def __init__(self):
 991     """Initialize this class.
 992
 993     """
 994     _RpcClientBase.__init__(self, compat.partial(_SsconfResolver, False),
 995                             _ENCODERS.get)
 996     _generated_rpc.RpcClientDnsOnly.__init__(self)
 997
 998
 999 class ConfigRunner(_RpcClientBase, _generated_rpc.RpcClientConfig):
1000   """RPC wrappers for L{config}.
1001
1002   """
1003   def __init__(self, context, address_list, _req_process_fn=None,
1004                _getents=None):
1005     """Initializes this class.
1006
1007     """
1008     if context:
1009       lock_monitor_cb = context.glm.AddToLockMonitor
1010     else:
1011       lock_monitor_cb = None
1012
1013     if address_list is None:
1014       resolver = compat.partial(_SsconfResolver, True)
1015     else:
1016       # Caller provided an address list
1017       resolver = _StaticResolver(address_list)
1018
1019     encoders = _ENCODERS.copy()
1020
1021     encoders.update({
1022       rpc_defs.ED_FILE_DETAILS: compat.partial(_PrepareFileUpload, _getents),
1023       })
1024
1025     _RpcClientBase.__init__(self, resolver, encoders.get,
1026                             lock_monitor_cb=lock_monitor_cb,
1027                             _req_process_fn=_req_process_fn)
1028     _generated_rpc.RpcClientConfig.__init__(self)