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

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2008, 2009, 2010 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-msg=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 os
  34 import logging
  35 import zlib
  36 import base64
  37 import pycurl
  38 import threading
  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
  49 # pylint has a bug here, doesn't see this import
  50 import ganeti.http.client  # pylint: disable-msg=W0611
  51
  52
  53 # Timeout for connecting to nodes (seconds)
  54 _RPC_CONNECT_TIMEOUT = 5
  55
  56 _RPC_CLIENT_HEADERS = [
  57   "Content-type: %s" % http.HTTP_APP_JSON,
  58   "Expect:",
  59   ]
  60
  61 # Various time constants for the timeout table
  62 _TMO_URGENT = 60 # one minute
  63 _TMO_FAST = 5 * 60 # five minutes
  64 _TMO_NORMAL = 15 * 60 # 15 minutes
  65 _TMO_SLOW = 3600 # one hour
  66 _TMO_4HRS = 4 * 3600
  67 _TMO_1DAY = 86400
  68
  69 # Timeout table that will be built later by decorators
  70 # Guidelines for choosing timeouts:
  71 # - call used during watcher: timeout -> 1min, _TMO_URGENT
  72 # - trivial (but be sure it is trivial) (e.g. reading a file): 5min, _TMO_FAST
  73 # - other calls: 15 min, _TMO_NORMAL
  74 # - special calls (instance add, etc.): either _TMO_SLOW (1h) or huge timeouts
  75
  76 _TIMEOUTS = {
  77 }
  78
  79
  80 def Init():
  81   """Initializes the module-global HTTP client manager.
  82
  83   Must be called before using any RPC function and while exactly one thread is
  84   running.
  85
  86   """
  87   # curl_global_init(3) and curl_global_cleanup(3) must be called with only
  88   # one thread running. This check is just a safety measure -- it doesn't
  89   # cover all cases.
  90   assert threading.activeCount() == 1, \
  91          "Found more than one active thread when initializing pycURL"
  92
  93   logging.info("Using PycURL %s", pycurl.version)
  94
  95   pycurl.global_init(pycurl.GLOBAL_ALL)
  96
  97
  98 def Shutdown():
  99   """Stops the module-global HTTP client manager.
 100
 101   Must be called before quitting the program and while exactly one thread is
 102   running.
 103
 104   """
 105   pycurl.global_cleanup()
 106
 107
 108 def _ConfigRpcCurl(curl):
 109   noded_cert = str(constants.NODED_CERT_FILE)
 110
 111   curl.setopt(pycurl.FOLLOWLOCATION, False)
 112   curl.setopt(pycurl.CAINFO, noded_cert)
 113   curl.setopt(pycurl.SSL_VERIFYHOST, 0)
 114   curl.setopt(pycurl.SSL_VERIFYPEER, True)
 115   curl.setopt(pycurl.SSLCERTTYPE, "PEM")
 116   curl.setopt(pycurl.SSLCERT, noded_cert)
 117   curl.setopt(pycurl.SSLKEYTYPE, "PEM")
 118   curl.setopt(pycurl.SSLKEY, noded_cert)
 119   curl.setopt(pycurl.CONNECTTIMEOUT, _RPC_CONNECT_TIMEOUT)
 120
 121
 122 # Aliasing this module avoids the following warning by epydoc: "Warning: No
 123 # information available for ganeti.rpc._RpcThreadLocal's base threading.local"
 124 _threading = threading
 125
 126
 127 class _RpcThreadLocal(_threading.local):
 128   def GetHttpClientPool(self):
 129     """Returns a per-thread HTTP client pool.
 130
 131     @rtype: L{http.client.HttpClientPool}
 132
 133     """
 134     try:
 135       pool = self.hcp
 136     except AttributeError:
 137       pool = http.client.HttpClientPool(_ConfigRpcCurl)
 138       self.hcp = pool
 139
 140     return pool
 141
 142
 143 # Remove module alias (see above)
 144 del _threading
 145
 146
 147 _thread_local = _RpcThreadLocal()
 148
 149
 150 def _RpcTimeout(secs):
 151   """Timeout decorator.
 152
 153   When applied to a rpc call_* function, it updates the global timeout
 154   table with the given function/timeout.
 155
 156   """
 157   def decorator(f):
 158     name = f.__name__
 159     assert name.startswith("call_")
 160     _TIMEOUTS[name[len("call_"):]] = secs
 161     return f
 162   return decorator
 163
 164
 165 def RunWithRPC(fn):
 166   """RPC-wrapper decorator.
 167
 168   When applied to a function, it runs it with the RPC system
 169   initialized, and it shutsdown the system afterwards. This means the
 170   function must be called without RPC being initialized.
 171
 172   """
 173   def wrapper(*args, **kwargs):
 174     Init()
 175     try:
 176       return fn(*args, **kwargs)
 177     finally:
 178       Shutdown()
 179   return wrapper
 180
 181
 182 class RpcResult(object):
 183   """RPC Result class.
 184
 185   This class holds an RPC result. It is needed since in multi-node
 186   calls we can't raise an exception just because one one out of many
 187   failed, and therefore we use this class to encapsulate the result.
 188
 189   @ivar data: the data payload, for successful results, or None
 190   @ivar call: the name of the RPC call
 191   @ivar node: the name of the node to which we made the call
 192   @ivar offline: whether the operation failed because the node was
 193       offline, as opposed to actual failure; offline=True will always
 194       imply failed=True, in order to allow simpler checking if
 195       the user doesn't care about the exact failure mode
 196   @ivar fail_msg: the error message if the call failed
 197
 198   """
 199   def __init__(self, data=None, failed=False, offline=False,
 200                call=None, node=None):
 201     self.offline = offline
 202     self.call = call
 203     self.node = node
 204
 205     if offline:
 206       self.fail_msg = "Node is marked offline"
 207       self.data = self.payload = None
 208     elif failed:
 209       self.fail_msg = self._EnsureErr(data)
 210       self.data = self.payload = None
 211     else:
 212       self.data = data
 213       if not isinstance(self.data, (tuple, list)):
 214         self.fail_msg = ("RPC layer error: invalid result type (%s)" %
 215                          type(self.data))
 216         self.payload = None
 217       elif len(data) != 2:
 218         self.fail_msg = ("RPC layer error: invalid result length (%d), "
 219                          "expected 2" % len(self.data))
 220         self.payload = None
 221       elif not self.data[0]:
 222         self.fail_msg = self._EnsureErr(self.data[1])
 223         self.payload = None
 224       else:
 225         # finally success
 226         self.fail_msg = None
 227         self.payload = data[1]
 228
 229     assert hasattr(self, "call")
 230     assert hasattr(self, "data")
 231     assert hasattr(self, "fail_msg")
 232     assert hasattr(self, "node")
 233     assert hasattr(self, "offline")
 234     assert hasattr(self, "payload")
 235
 236   @staticmethod
 237   def _EnsureErr(val):
 238     """Helper to ensure we return a 'True' value for error."""
 239     if val:
 240       return val
 241     else:
 242       return "No error information"
 243
 244   def Raise(self, msg, prereq=False, ecode=None):
 245     """If the result has failed, raise an OpExecError.
 246
 247     This is used so that LU code doesn't have to check for each
 248     result, but instead can call this function.
 249
 250     """
 251     if not self.fail_msg:
 252       return
 253
 254     if not msg: # one could pass None for default message
 255       msg = ("Call '%s' to node '%s' has failed: %s" %
 256              (self.call, self.node, self.fail_msg))
 257     else:
 258       msg = "%s: %s" % (msg, self.fail_msg)
 259     if prereq:
 260       ec = errors.OpPrereqError
 261     else:
 262       ec = errors.OpExecError
 263     if ecode is not None:
 264       args = (msg, ecode)
 265     else:
 266       args = (msg, )
 267     raise ec(*args) # pylint: disable-msg=W0142
 268
 269
 270 def _AddressLookup(node_list,
 271                    ssc=ssconf.SimpleStore,
 272                    nslookup_fn=netutils.Hostname.GetIP):
 273   """Return addresses for given node names.
 274
 275   @type node_list: list
 276   @param node_list: List of node names
 277   @type ssc: class
 278   @param ssc: SimpleStore class that is used to obtain node->ip mappings
 279   @type nslookup_fn: callable
 280   @param nslookup_fn: function use to do NS lookup
 281   @rtype: list of addresses and/or None's
 282   @returns: List of corresponding addresses, if found
 283
 284   """
 285   ss = ssc()
 286   iplist = ss.GetNodePrimaryIPList()
 287   family = ss.GetPrimaryIPFamily()
 288   addresses = []
 289   ipmap = dict(entry.split() for entry in iplist)
 290   for node in node_list:
 291     address = ipmap.get(node)
 292     if address is None:
 293       address = nslookup_fn(node, family=family)
 294     addresses.append(address)
 295
 296   return addresses
 297
 298
 299 class Client:
 300   """RPC Client class.
 301
 302   This class, given a (remote) method name, a list of parameters and a
 303   list of nodes, will contact (in parallel) all nodes, and return a
 304   dict of results (key: node name, value: result).
 305
 306   One current bug is that generic failure is still signaled by
 307   'False' result, which is not good. This overloading of values can
 308   cause bugs.
 309
 310   """
 311   def __init__(self, procedure, body, port, address_lookup_fn=_AddressLookup):
 312     assert procedure in _TIMEOUTS, ("New RPC call not declared in the"
 313                                     " timeouts table")
 314     self.procedure = procedure
 315     self.body = body
 316     self.port = port
 317     self._request = {}
 318     self._address_lookup_fn = address_lookup_fn
 319
 320   def ConnectList(self, node_list, address_list=None, read_timeout=None):
 321     """Add a list of nodes to the target nodes.
 322
 323     @type node_list: list
 324     @param node_list: the list of node names to connect
 325     @type address_list: list or None
 326     @keyword address_list: either None or a list with node addresses,
 327         which must have the same length as the node list
 328     @type read_timeout: int
 329     @param read_timeout: overwrites default timeout for operation
 330
 331     """
 332     if address_list is None:
 333       # Always use IP address instead of node name
 334       address_list = self._address_lookup_fn(node_list)
 335
 336     assert len(node_list) == len(address_list), \
 337            "Name and address lists must have the same length"
 338
 339     for node, address in zip(node_list, address_list):
 340       self.ConnectNode(node, address, read_timeout=read_timeout)
 341
 342   def ConnectNode(self, name, address=None, read_timeout=None):
 343     """Add a node to the target list.
 344
 345     @type name: str
 346     @param name: the node name
 347     @type address: str
 348     @param address: the node address, if known
 349     @type read_timeout: int
 350     @param read_timeout: overwrites default timeout for operation
 351
 352     """
 353     if address is None:
 354       # Always use IP address instead of node name
 355       address = self._address_lookup_fn([name])[0]
 356
 357     assert(address is not None)
 358
 359     if read_timeout is None:
 360       read_timeout = _TIMEOUTS[self.procedure]
 361
 362     self._request[name] = \
 363       http.client.HttpClientRequest(str(address), self.port,
 364                                     http.HTTP_PUT, str("/%s" % self.procedure),
 365                                     headers=_RPC_CLIENT_HEADERS,
 366                                     post_data=str(self.body),
 367                                     read_timeout=read_timeout)
 368
 369   def GetResults(self, http_pool=None):
 370     """Call nodes and return results.
 371
 372     @rtype: list
 373     @return: List of RPC results
 374
 375     """
 376     if not http_pool:
 377       http_pool = _thread_local.GetHttpClientPool()
 378
 379     http_pool.ProcessRequests(self._request.values())
 380
 381     results = {}
 382
 383     for name, req in self._request.iteritems():
 384       if req.success and req.resp_status_code == http.HTTP_OK:
 385         results[name] = RpcResult(data=serializer.LoadJson(req.resp_body),
 386                                   node=name, call=self.procedure)
 387         continue
 388
 389       # TODO: Better error reporting
 390       if req.error:
 391         msg = req.error
 392       else:
 393         msg = req.resp_body
 394
 395       logging.error("RPC error in %s from node %s: %s",
 396                     self.procedure, name, msg)
 397       results[name] = RpcResult(data=msg, failed=True, node=name,
 398                                 call=self.procedure)
 399
 400     return results
 401
 402
 403 def _EncodeImportExportIO(ieio, ieioargs):
 404   """Encodes import/export I/O information.
 405
 406   """
 407   if ieio == constants.IEIO_RAW_DISK:
 408     assert len(ieioargs) == 1
 409     return (ieioargs[0].ToDict(), )
 410
 411   if ieio == constants.IEIO_SCRIPT:
 412     assert len(ieioargs) == 2
 413     return (ieioargs[0].ToDict(), ieioargs[1])
 414
 415   return ieioargs
 416
 417
 418 class RpcRunner(object):
 419   """RPC runner class"""
 420
 421   def __init__(self, cfg):
 422     """Initialized the rpc runner.
 423
 424     @type cfg:  C{config.ConfigWriter}
 425     @param cfg: the configuration object that will be used to get data
 426                 about the cluster
 427
 428     """
 429     self._cfg = cfg
 430     self.port = netutils.GetDaemonPort(constants.NODED)
 431
 432   def _InstDict(self, instance, hvp=None, bep=None, osp=None):
 433     """Convert the given instance to a dict.
 434
 435     This is done via the instance's ToDict() method and additionally
 436     we fill the hvparams with the cluster defaults.
 437
 438     @type instance: L{objects.Instance}
 439     @param instance: an Instance object
 440     @type hvp: dict or None
 441     @param hvp: a dictionary with overridden hypervisor parameters
 442     @type bep: dict or None
 443     @param bep: a dictionary with overridden backend parameters
 444     @type osp: dict or None
 445     @param osp: a dictionary with overridden os parameters
 446     @rtype: dict
 447     @return: the instance dict, with the hvparams filled with the
 448         cluster defaults
 449
 450     """
 451     idict = instance.ToDict()
 452     cluster = self._cfg.GetClusterInfo()
 453     idict["hvparams"] = cluster.FillHV(instance)
 454     if hvp is not None:
 455       idict["hvparams"].update(hvp)
 456     idict["beparams"] = cluster.FillBE(instance)
 457     if bep is not None:
 458       idict["beparams"].update(bep)
 459     idict["osparams"] = cluster.SimpleFillOS(instance.os, instance.osparams)
 460     if osp is not None:
 461       idict["osparams"].update(osp)
 462     for nic in idict["nics"]:
 463       nic['nicparams'] = objects.FillDict(
 464         cluster.nicparams[constants.PP_DEFAULT],
 465         nic['nicparams'])
 466     return idict
 467
 468   def _ConnectList(self, client, node_list, call, read_timeout=None):
 469     """Helper for computing node addresses.
 470
 471     @type client: L{ganeti.rpc.Client}
 472     @param client: a C{Client} instance
 473     @type node_list: list
 474     @param node_list: the node list we should connect
 475     @type call: string
 476     @param call: the name of the remote procedure call, for filling in
 477         correctly any eventual offline nodes' results
 478     @type read_timeout: int
 479     @param read_timeout: overwrites the default read timeout for the
 480         given operation
 481
 482     """
 483     all_nodes = self._cfg.GetAllNodesInfo()
 484     name_list = []
 485     addr_list = []
 486     skip_dict = {}
 487     for node in node_list:
 488       if node in all_nodes:
 489         if all_nodes[node].offline:
 490           skip_dict[node] = RpcResult(node=node, offline=True, call=call)
 491           continue
 492         val = all_nodes[node].primary_ip
 493       else:
 494         val = None
 495       addr_list.append(val)
 496       name_list.append(node)
 497     if name_list:
 498       client.ConnectList(name_list, address_list=addr_list,
 499                          read_timeout=read_timeout)
 500     return skip_dict
 501
 502   def _ConnectNode(self, client, node, call, read_timeout=None):
 503     """Helper for computing one node's address.
 504
 505     @type client: L{ganeti.rpc.Client}
 506     @param client: a C{Client} instance
 507     @type node: str
 508     @param node: the node we should connect
 509     @type call: string
 510     @param call: the name of the remote procedure call, for filling in
 511         correctly any eventual offline nodes' results
 512     @type read_timeout: int
 513     @param read_timeout: overwrites the default read timeout for the
 514         given operation
 515
 516     """
 517     node_info = self._cfg.GetNodeInfo(node)
 518     if node_info is not None:
 519       if node_info.offline:
 520         return RpcResult(node=node, offline=True, call=call)
 521       addr = node_info.primary_ip
 522     else:
 523       addr = None
 524     client.ConnectNode(node, address=addr, read_timeout=read_timeout)
 525
 526   def _MultiNodeCall(self, node_list, procedure, args, read_timeout=None):
 527     """Helper for making a multi-node call
 528
 529     """
 530     body = serializer.DumpJson(args, indent=False)
 531     c = Client(procedure, body, self.port)
 532     skip_dict = self._ConnectList(c, node_list, procedure,
 533                                   read_timeout=read_timeout)
 534     skip_dict.update(c.GetResults())
 535     return skip_dict
 536
 537   @classmethod
 538   def _StaticMultiNodeCall(cls, node_list, procedure, args,
 539                            address_list=None, read_timeout=None):
 540     """Helper for making a multi-node static call
 541
 542     """
 543     body = serializer.DumpJson(args, indent=False)
 544     c = Client(procedure, body, netutils.GetDaemonPort(constants.NODED))
 545     c.ConnectList(node_list, address_list=address_list,
 546                   read_timeout=read_timeout)
 547     return c.GetResults()
 548
 549   def _SingleNodeCall(self, node, procedure, args, read_timeout=None):
 550     """Helper for making a single-node call
 551
 552     """
 553     body = serializer.DumpJson(args, indent=False)
 554     c = Client(procedure, body, self.port)
 555     result = self._ConnectNode(c, node, procedure, read_timeout=read_timeout)
 556     if result is None:
 557       # we did connect, node is not offline
 558       result = c.GetResults()[node]
 559     return result
 560
 561   @classmethod
 562   def _StaticSingleNodeCall(cls, node, procedure, args, read_timeout=None):
 563     """Helper for making a single-node static call
 564
 565     """
 566     body = serializer.DumpJson(args, indent=False)
 567     c = Client(procedure, body, netutils.GetDaemonPort(constants.NODED))
 568     c.ConnectNode(node, read_timeout=read_timeout)
 569     return c.GetResults()[node]
 570
 571   @staticmethod
 572   def _Compress(data):
 573     """Compresses a string for transport over RPC.
 574
 575     Small amounts of data are not compressed.
 576
 577     @type data: str
 578     @param data: Data
 579     @rtype: tuple
 580     @return: Encoded data to send
 581
 582     """
 583     # Small amounts of data are not compressed
 584     if len(data) < 512:
 585       return (constants.RPC_ENCODING_NONE, data)
 586
 587     # Compress with zlib and encode in base64
 588     return (constants.RPC_ENCODING_ZLIB_BASE64,
 589             base64.b64encode(zlib.compress(data, 3)))
 590
 591   #
 592   # Begin RPC calls
 593   #
 594
 595   @_RpcTimeout(_TMO_URGENT)
 596   def call_lv_list(self, node_list, vg_name):
 597     """Gets the logical volumes present in a given volume group.
 598
 599     This is a multi-node call.
 600
 601     """
 602     return self._MultiNodeCall(node_list, "lv_list", [vg_name])
 603
 604   @_RpcTimeout(_TMO_URGENT)
 605   def call_vg_list(self, node_list):
 606     """Gets the volume group list.
 607
 608     This is a multi-node call.
 609
 610     """
 611     return self._MultiNodeCall(node_list, "vg_list", [])
 612
 613   @_RpcTimeout(_TMO_NORMAL)
 614   def call_storage_list(self, node_list, su_name, su_args, name, fields):
 615     """Get list of storage units.
 616
 617     This is a multi-node call.
 618
 619     """
 620     return self._MultiNodeCall(node_list, "storage_list",
 621                                [su_name, su_args, name, fields])
 622
 623   @_RpcTimeout(_TMO_NORMAL)
 624   def call_storage_modify(self, node, su_name, su_args, name, changes):
 625     """Modify a storage unit.
 626
 627     This is a single-node call.
 628
 629     """
 630     return self._SingleNodeCall(node, "storage_modify",
 631                                 [su_name, su_args, name, changes])
 632
 633   @_RpcTimeout(_TMO_NORMAL)
 634   def call_storage_execute(self, node, su_name, su_args, name, op):
 635     """Executes an operation on a storage unit.
 636
 637     This is a single-node call.
 638
 639     """
 640     return self._SingleNodeCall(node, "storage_execute",
 641                                 [su_name, su_args, name, op])
 642
 643   @_RpcTimeout(_TMO_URGENT)
 644   def call_bridges_exist(self, node, bridges_list):
 645     """Checks if a node has all the bridges given.
 646
 647     This method checks if all bridges given in the bridges_list are
 648     present on the remote node, so that an instance that uses interfaces
 649     on those bridges can be started.
 650
 651     This is a single-node call.
 652
 653     """
 654     return self._SingleNodeCall(node, "bridges_exist", [bridges_list])
 655
 656   @_RpcTimeout(_TMO_NORMAL)
 657   def call_instance_start(self, node, instance, hvp, bep):
 658     """Starts an instance.
 659
 660     This is a single-node call.
 661
 662     """
 663     idict = self._InstDict(instance, hvp=hvp, bep=bep)
 664     return self._SingleNodeCall(node, "instance_start", [idict])
 665
 666   @_RpcTimeout(_TMO_NORMAL)
 667   def call_instance_shutdown(self, node, instance, timeout):
 668     """Stops an instance.
 669
 670     This is a single-node call.
 671
 672     """
 673     return self._SingleNodeCall(node, "instance_shutdown",
 674                                 [self._InstDict(instance), timeout])
 675
 676   @_RpcTimeout(_TMO_NORMAL)
 677   def call_migration_info(self, node, instance):
 678     """Gather the information necessary to prepare an instance migration.
 679
 680     This is a single-node call.
 681
 682     @type node: string
 683     @param node: the node on which the instance is currently running
 684     @type instance: C{objects.Instance}
 685     @param instance: the instance definition
 686
 687     """
 688     return self._SingleNodeCall(node, "migration_info",
 689                                 [self._InstDict(instance)])
 690
 691   @_RpcTimeout(_TMO_NORMAL)
 692   def call_accept_instance(self, node, instance, info, target):
 693     """Prepare a node to accept an instance.
 694
 695     This is a single-node call.
 696
 697     @type node: string
 698     @param node: the target node for the migration
 699     @type instance: C{objects.Instance}
 700     @param instance: the instance definition
 701     @type info: opaque/hypervisor specific (string/data)
 702     @param info: result for the call_migration_info call
 703     @type target: string
 704     @param target: target hostname (usually ip address) (on the node itself)
 705
 706     """
 707     return self._SingleNodeCall(node, "accept_instance",
 708                                 [self._InstDict(instance), info, target])
 709
 710   @_RpcTimeout(_TMO_NORMAL)
 711   def call_finalize_migration(self, node, instance, info, success):
 712     """Finalize any target-node migration specific operation.
 713
 714     This is called both in case of a successful migration and in case of error
 715     (in which case it should abort the migration).
 716
 717     This is a single-node call.
 718
 719     @type node: string
 720     @param node: the target node for the migration
 721     @type instance: C{objects.Instance}
 722     @param instance: the instance definition
 723     @type info: opaque/hypervisor specific (string/data)
 724     @param info: result for the call_migration_info call
 725     @type success: boolean
 726     @param success: whether the migration was a success or a failure
 727
 728     """
 729     return self._SingleNodeCall(node, "finalize_migration",
 730                                 [self._InstDict(instance), info, success])
 731
 732   @_RpcTimeout(_TMO_SLOW)
 733   def call_instance_migrate(self, node, instance, target, live):
 734     """Migrate an instance.
 735
 736     This is a single-node call.
 737
 738     @type node: string
 739     @param node: the node on which the instance is currently running
 740     @type instance: C{objects.Instance}
 741     @param instance: the instance definition
 742     @type target: string
 743     @param target: the target node name
 744     @type live: boolean
 745     @param live: whether the migration should be done live or not (the
 746         interpretation of this parameter is left to the hypervisor)
 747
 748     """
 749     return self._SingleNodeCall(node, "instance_migrate",
 750                                 [self._InstDict(instance), target, live])
 751
 752   @_RpcTimeout(_TMO_NORMAL)
 753   def call_instance_reboot(self, node, inst, reboot_type, shutdown_timeout):
 754     """Reboots an instance.
 755
 756     This is a single-node call.
 757
 758     """
 759     return self._SingleNodeCall(node, "instance_reboot",
 760                                 [self._InstDict(inst), reboot_type,
 761                                  shutdown_timeout])
 762
 763   @_RpcTimeout(_TMO_1DAY)
 764   def call_instance_os_add(self, node, inst, reinstall, debug, osparams=None):
 765     """Installs an OS on the given instance.
 766
 767     This is a single-node call.
 768
 769     """
 770     return self._SingleNodeCall(node, "instance_os_add",
 771                                 [self._InstDict(inst, osp=osparams),
 772                                  reinstall, debug])
 773
 774   @_RpcTimeout(_TMO_SLOW)
 775   def call_instance_run_rename(self, node, inst, old_name, debug):
 776     """Run the OS rename script for an instance.
 777
 778     This is a single-node call.
 779
 780     """
 781     return self._SingleNodeCall(node, "instance_run_rename",
 782                                 [self._InstDict(inst), old_name, debug])
 783
 784   @_RpcTimeout(_TMO_URGENT)
 785   def call_instance_info(self, node, instance, hname):
 786     """Returns information about a single instance.
 787
 788     This is a single-node call.
 789
 790     @type node: list
 791     @param node: the list of nodes to query
 792     @type instance: string
 793     @param instance: the instance name
 794     @type hname: string
 795     @param hname: the hypervisor type of the instance
 796
 797     """
 798     return self._SingleNodeCall(node, "instance_info", [instance, hname])
 799
 800   @_RpcTimeout(_TMO_NORMAL)
 801   def call_instance_migratable(self, node, instance):
 802     """Checks whether the given instance can be migrated.
 803
 804     This is a single-node call.
 805
 806     @param node: the node to query
 807     @type instance: L{objects.Instance}
 808     @param instance: the instance to check
 809
 810
 811     """
 812     return self._SingleNodeCall(node, "instance_migratable",
 813                                 [self._InstDict(instance)])
 814
 815   @_RpcTimeout(_TMO_URGENT)
 816   def call_all_instances_info(self, node_list, hypervisor_list):
 817     """Returns information about all instances on the given nodes.
 818
 819     This is a multi-node call.
 820
 821     @type node_list: list
 822     @param node_list: the list of nodes to query
 823     @type hypervisor_list: list
 824     @param hypervisor_list: the hypervisors to query for instances
 825
 826     """
 827     return self._MultiNodeCall(node_list, "all_instances_info",
 828                                [hypervisor_list])
 829
 830   @_RpcTimeout(_TMO_URGENT)
 831   def call_instance_list(self, node_list, hypervisor_list):
 832     """Returns the list of running instances on a given node.
 833
 834     This is a multi-node call.
 835
 836     @type node_list: list
 837     @param node_list: the list of nodes to query
 838     @type hypervisor_list: list
 839     @param hypervisor_list: the hypervisors to query for instances
 840
 841     """
 842     return self._MultiNodeCall(node_list, "instance_list", [hypervisor_list])
 843
 844   @_RpcTimeout(_TMO_FAST)
 845   def call_node_tcp_ping(self, node, source, target, port, timeout,
 846                          live_port_needed):
 847     """Do a TcpPing on the remote node
 848
 849     This is a single-node call.
 850
 851     """
 852     return self._SingleNodeCall(node, "node_tcp_ping",
 853                                 [source, target, port, timeout,
 854                                  live_port_needed])
 855
 856   @_RpcTimeout(_TMO_FAST)
 857   def call_node_has_ip_address(self, node, address):
 858     """Checks if a node has the given IP address.
 859
 860     This is a single-node call.
 861
 862     """
 863     return self._SingleNodeCall(node, "node_has_ip_address", [address])
 864
 865   @_RpcTimeout(_TMO_URGENT)
 866   def call_node_info(self, node_list, vg_name, hypervisor_type):
 867     """Return node information.
 868
 869     This will return memory information and volume group size and free
 870     space.
 871
 872     This is a multi-node call.
 873
 874     @type node_list: list
 875     @param node_list: the list of nodes to query
 876     @type vg_name: C{string}
 877     @param vg_name: the name of the volume group to ask for disk space
 878         information
 879     @type hypervisor_type: C{str}
 880     @param hypervisor_type: the name of the hypervisor to ask for
 881         memory information
 882
 883     """
 884     return self._MultiNodeCall(node_list, "node_info",
 885                                [vg_name, hypervisor_type])
 886
 887   @_RpcTimeout(_TMO_NORMAL)
 888   def call_etc_hosts_modify(self, node, mode, name, ip):
 889     """Modify hosts file with name
 890
 891     @type node: string
 892     @param node: The node to call
 893     @type mode: string
 894     @param mode: The mode to operate. Currently "add" or "remove"
 895     @type name: string
 896     @param name: The host name to be modified
 897     @type ip: string
 898     @param ip: The ip of the entry (just valid if mode is "add")
 899
 900     """
 901     return self._SingleNodeCall(node, "etc_hosts_modify", [mode, name, ip])
 902
 903   @_RpcTimeout(_TMO_NORMAL)
 904   def call_node_verify(self, node_list, checkdict, cluster_name):
 905     """Request verification of given parameters.
 906
 907     This is a multi-node call.
 908
 909     """
 910     return self._MultiNodeCall(node_list, "node_verify",
 911                                [checkdict, cluster_name])
 912
 913   @classmethod
 914   @_RpcTimeout(_TMO_FAST)
 915   def call_node_start_master(cls, node, start_daemons, no_voting):
 916     """Tells a node to activate itself as a master.
 917
 918     This is a single-node call.
 919
 920     """
 921     return cls._StaticSingleNodeCall(node, "node_start_master",
 922                                      [start_daemons, no_voting])
 923
 924   @classmethod
 925   @_RpcTimeout(_TMO_FAST)
 926   def call_node_stop_master(cls, node, stop_daemons):
 927     """Tells a node to demote itself from master status.
 928
 929     This is a single-node call.
 930
 931     """
 932     return cls._StaticSingleNodeCall(node, "node_stop_master", [stop_daemons])
 933
 934   @classmethod
 935   @_RpcTimeout(_TMO_URGENT)
 936   def call_master_info(cls, node_list):
 937     """Query master info.
 938
 939     This is a multi-node call.
 940
 941     """
 942     # TODO: should this method query down nodes?
 943     return cls._StaticMultiNodeCall(node_list, "master_info", [])
 944
 945   @classmethod
 946   @_RpcTimeout(_TMO_URGENT)
 947   def call_version(cls, node_list):
 948     """Query node version.
 949
 950     This is a multi-node call.
 951
 952     """
 953     return cls._StaticMultiNodeCall(node_list, "version", [])
 954
 955   @_RpcTimeout(_TMO_NORMAL)
 956   def call_blockdev_create(self, node, bdev, size, owner, on_primary, info):
 957     """Request creation of a given block device.
 958
 959     This is a single-node call.
 960
 961     """
 962     return self._SingleNodeCall(node, "blockdev_create",
 963                                 [bdev.ToDict(), size, owner, on_primary, info])
 964
 965   @_RpcTimeout(_TMO_SLOW)
 966   def call_blockdev_wipe(self, node, bdev, offset, size):
 967     """Request wipe at given offset with given size of a block device.
 968
 969     This is a single-node call.
 970
 971     """
 972     return self._SingleNodeCall(node, "blockdev_wipe",
 973                                 [bdev.ToDict(), offset, size])
 974
 975   @_RpcTimeout(_TMO_NORMAL)
 976   def call_blockdev_remove(self, node, bdev):
 977     """Request removal of a given block device.
 978
 979     This is a single-node call.
 980
 981     """
 982     return self._SingleNodeCall(node, "blockdev_remove", [bdev.ToDict()])
 983
 984   @_RpcTimeout(_TMO_NORMAL)
 985   def call_blockdev_rename(self, node, devlist):
 986     """Request rename of the given block devices.
 987
 988     This is a single-node call.
 989
 990     """
 991     return self._SingleNodeCall(node, "blockdev_rename",
 992                                 [(d.ToDict(), uid) for d, uid in devlist])
 993
 994   @_RpcTimeout(_TMO_NORMAL)
 995   def call_blockdev_assemble(self, node, disk, owner, on_primary):
 996     """Request assembling of a given block device.
 997
 998     This is a single-node call.
 999
1000     """
1001     return self._SingleNodeCall(node, "blockdev_assemble",
1002                                 [disk.ToDict(), owner, on_primary])
1003
1004   @_RpcTimeout(_TMO_NORMAL)
1005   def call_blockdev_shutdown(self, node, disk):
1006     """Request shutdown of a given block device.
1007
1008     This is a single-node call.
1009
1010     """
1011     return self._SingleNodeCall(node, "blockdev_shutdown", [disk.ToDict()])
1012
1013   @_RpcTimeout(_TMO_NORMAL)
1014   def call_blockdev_addchildren(self, node, bdev, ndevs):
1015     """Request adding a list of children to a (mirroring) device.
1016
1017     This is a single-node call.
1018
1019     """
1020     return self._SingleNodeCall(node, "blockdev_addchildren",
1021                                 [bdev.ToDict(),
1022                                  [disk.ToDict() for disk in ndevs]])
1023
1024   @_RpcTimeout(_TMO_NORMAL)
1025   def call_blockdev_removechildren(self, node, bdev, ndevs):
1026     """Request removing a list of children from a (mirroring) device.
1027
1028     This is a single-node call.
1029
1030     """
1031     return self._SingleNodeCall(node, "blockdev_removechildren",
1032                                 [bdev.ToDict(),
1033                                  [disk.ToDict() for disk in ndevs]])
1034
1035   @_RpcTimeout(_TMO_NORMAL)
1036   def call_blockdev_getmirrorstatus(self, node, disks):
1037     """Request status of a (mirroring) device.
1038
1039     This is a single-node call.
1040
1041     """
1042     result = self._SingleNodeCall(node, "blockdev_getmirrorstatus",
1043                                   [dsk.ToDict() for dsk in disks])
1044     if not result.fail_msg:
1045       result.payload = [objects.BlockDevStatus.FromDict(i)
1046                         for i in result.payload]
1047     return result
1048
1049   @_RpcTimeout(_TMO_NORMAL)
1050   def call_blockdev_getmirrorstatus_multi(self, node_list, node_disks):
1051     """Request status of (mirroring) devices from multiple nodes.
1052
1053     This is a multi-node call.
1054
1055     """
1056     result = self._MultiNodeCall(node_list, "blockdev_getmirrorstatus_multi",
1057                                  [dict((name, [dsk.ToDict() for dsk in disks])
1058                                        for name, disks in node_disks.items())])
1059     for nres in result.values():
1060       if nres.fail_msg:
1061         continue
1062
1063       for idx, (success, status) in enumerate(nres.payload):
1064         if success:
1065           nres.payload[idx] = (success, objects.BlockDevStatus.FromDict(status))
1066
1067     return result
1068
1069   @_RpcTimeout(_TMO_NORMAL)
1070   def call_blockdev_find(self, node, disk):
1071     """Request identification of a given block device.
1072
1073     This is a single-node call.
1074
1075     """
1076     result = self._SingleNodeCall(node, "blockdev_find", [disk.ToDict()])
1077     if not result.fail_msg and result.payload is not None:
1078       result.payload = objects.BlockDevStatus.FromDict(result.payload)
1079     return result
1080
1081   @_RpcTimeout(_TMO_NORMAL)
1082   def call_blockdev_close(self, node, instance_name, disks):
1083     """Closes the given block devices.
1084
1085     This is a single-node call.
1086
1087     """
1088     params = [instance_name, [cf.ToDict() for cf in disks]]
1089     return self._SingleNodeCall(node, "blockdev_close", params)
1090
1091   @_RpcTimeout(_TMO_NORMAL)
1092   def call_blockdev_getsize(self, node, disks):
1093     """Returns the size of the given disks.
1094
1095     This is a single-node call.
1096
1097     """
1098     params = [[cf.ToDict() for cf in disks]]
1099     return self._SingleNodeCall(node, "blockdev_getsize", params)
1100
1101   @_RpcTimeout(_TMO_NORMAL)
1102   def call_drbd_disconnect_net(self, node_list, nodes_ip, disks):
1103     """Disconnects the network of the given drbd devices.
1104
1105     This is a multi-node call.
1106
1107     """
1108     return self._MultiNodeCall(node_list, "drbd_disconnect_net",
1109                                [nodes_ip, [cf.ToDict() for cf in disks]])
1110
1111   @_RpcTimeout(_TMO_NORMAL)
1112   def call_drbd_attach_net(self, node_list, nodes_ip,
1113                            disks, instance_name, multimaster):
1114     """Disconnects the given drbd devices.
1115
1116     This is a multi-node call.
1117
1118     """
1119     return self._MultiNodeCall(node_list, "drbd_attach_net",
1120                                [nodes_ip, [cf.ToDict() for cf in disks],
1121                                 instance_name, multimaster])
1122
1123   @_RpcTimeout(_TMO_SLOW)
1124   def call_drbd_wait_sync(self, node_list, nodes_ip, disks):
1125     """Waits for the synchronization of drbd devices is complete.
1126
1127     This is a multi-node call.
1128
1129     """
1130     return self._MultiNodeCall(node_list, "drbd_wait_sync",
1131                                [nodes_ip, [cf.ToDict() for cf in disks]])
1132
1133   @_RpcTimeout(_TMO_URGENT)
1134   def call_drbd_helper(self, node_list):
1135     """Gets drbd helper.
1136
1137     This is a multi-node call.
1138
1139     """
1140     return self._MultiNodeCall(node_list, "drbd_helper", [])
1141
1142   @classmethod
1143   @_RpcTimeout(_TMO_NORMAL)
1144   def call_upload_file(cls, node_list, file_name, address_list=None):
1145     """Upload a file.
1146
1147     The node will refuse the operation in case the file is not on the
1148     approved file list.
1149
1150     This is a multi-node call.
1151
1152     @type node_list: list
1153     @param node_list: the list of node names to upload to
1154     @type file_name: str
1155     @param file_name: the filename to upload
1156     @type address_list: list or None
1157     @keyword address_list: an optional list of node addresses, in order
1158         to optimize the RPC speed
1159
1160     """
1161     file_contents = utils.ReadFile(file_name)
1162     data = cls._Compress(file_contents)
1163     st = os.stat(file_name)
1164     params = [file_name, data, st.st_mode, st.st_uid, st.st_gid,
1165               st.st_atime, st.st_mtime]
1166     return cls._StaticMultiNodeCall(node_list, "upload_file", params,
1167                                     address_list=address_list)
1168
1169   @classmethod
1170   @_RpcTimeout(_TMO_NORMAL)
1171   def call_write_ssconf_files(cls, node_list, values):
1172     """Write ssconf files.
1173
1174     This is a multi-node call.
1175
1176     """
1177     return cls._StaticMultiNodeCall(node_list, "write_ssconf_files", [values])
1178
1179   @_RpcTimeout(_TMO_FAST)
1180   def call_os_diagnose(self, node_list):
1181     """Request a diagnose of OS definitions.
1182
1183     This is a multi-node call.
1184
1185     """
1186     return self._MultiNodeCall(node_list, "os_diagnose", [])
1187
1188   @_RpcTimeout(_TMO_FAST)
1189   def call_os_get(self, node, name):
1190     """Returns an OS definition.
1191
1192     This is a single-node call.
1193
1194     """
1195     result = self._SingleNodeCall(node, "os_get", [name])
1196     if not result.fail_msg and isinstance(result.payload, dict):
1197       result.payload = objects.OS.FromDict(result.payload)
1198     return result
1199
1200   @_RpcTimeout(_TMO_FAST)
1201   def call_os_validate(self, required, nodes, name, checks, params):
1202     """Run a validation routine for a given OS.
1203
1204     This is a multi-node call.
1205
1206     """
1207     return self._MultiNodeCall(nodes, "os_validate",
1208                                [required, name, checks, params])
1209
1210   @_RpcTimeout(_TMO_NORMAL)
1211   def call_hooks_runner(self, node_list, hpath, phase, env):
1212     """Call the hooks runner.
1213
1214     Args:
1215       - op: the OpCode instance
1216       - env: a dictionary with the environment
1217
1218     This is a multi-node call.
1219
1220     """
1221     params = [hpath, phase, env]
1222     return self._MultiNodeCall(node_list, "hooks_runner", params)
1223
1224   @_RpcTimeout(_TMO_NORMAL)
1225   def call_iallocator_runner(self, node, name, idata):
1226     """Call an iallocator on a remote node
1227
1228     Args:
1229       - name: the iallocator name
1230       - input: the json-encoded input string
1231
1232     This is a single-node call.
1233
1234     """
1235     return self._SingleNodeCall(node, "iallocator_runner", [name, idata])
1236
1237   @_RpcTimeout(_TMO_NORMAL)
1238   def call_blockdev_grow(self, node, cf_bdev, amount):
1239     """Request a snapshot of the given block device.
1240
1241     This is a single-node call.
1242
1243     """
1244     return self._SingleNodeCall(node, "blockdev_grow",
1245                                 [cf_bdev.ToDict(), amount])
1246
1247   @_RpcTimeout(_TMO_1DAY)
1248   def call_blockdev_export(self, node, cf_bdev,
1249                            dest_node, dest_path, cluster_name):
1250     """Export a given disk to another node.
1251
1252     This is a single-node call.
1253
1254     """
1255     return self._SingleNodeCall(node, "blockdev_export",
1256                                 [cf_bdev.ToDict(), dest_node, dest_path,
1257                                  cluster_name])
1258
1259   @_RpcTimeout(_TMO_NORMAL)
1260   def call_blockdev_snapshot(self, node, cf_bdev):
1261     """Request a snapshot of the given block device.
1262
1263     This is a single-node call.
1264
1265     """
1266     return self._SingleNodeCall(node, "blockdev_snapshot", [cf_bdev.ToDict()])
1267
1268   @_RpcTimeout(_TMO_NORMAL)
1269   def call_finalize_export(self, node, instance, snap_disks):
1270     """Request the completion of an export operation.
1271
1272     This writes the export config file, etc.
1273
1274     This is a single-node call.
1275
1276     """
1277     flat_disks = []
1278     for disk in snap_disks:
1279       if isinstance(disk, bool):
1280         flat_disks.append(disk)
1281       else:
1282         flat_disks.append(disk.ToDict())
1283
1284     return self._SingleNodeCall(node, "finalize_export",
1285                                 [self._InstDict(instance), flat_disks])
1286
1287   @_RpcTimeout(_TMO_FAST)
1288   def call_export_info(self, node, path):
1289     """Queries the export information in a given path.
1290
1291     This is a single-node call.
1292
1293     """
1294     return self._SingleNodeCall(node, "export_info", [path])
1295
1296   @_RpcTimeout(_TMO_FAST)
1297   def call_export_list(self, node_list):
1298     """Gets the stored exports list.
1299
1300     This is a multi-node call.
1301
1302     """
1303     return self._MultiNodeCall(node_list, "export_list", [])
1304
1305   @_RpcTimeout(_TMO_FAST)
1306   def call_export_remove(self, node, export):
1307     """Requests removal of a given export.
1308
1309     This is a single-node call.
1310
1311     """
1312     return self._SingleNodeCall(node, "export_remove", [export])
1313
1314   @classmethod
1315   @_RpcTimeout(_TMO_NORMAL)
1316   def call_node_leave_cluster(cls, node, modify_ssh_setup):
1317     """Requests a node to clean the cluster information it has.
1318
1319     This will remove the configuration information from the ganeti data
1320     dir.
1321
1322     This is a single-node call.
1323
1324     """
1325     return cls._StaticSingleNodeCall(node, "node_leave_cluster",
1326                                      [modify_ssh_setup])
1327
1328   @_RpcTimeout(_TMO_FAST)
1329   def call_node_volumes(self, node_list):
1330     """Gets all volumes on node(s).
1331
1332     This is a multi-node call.
1333
1334     """
1335     return self._MultiNodeCall(node_list, "node_volumes", [])
1336
1337   @_RpcTimeout(_TMO_FAST)
1338   def call_node_demote_from_mc(self, node):
1339     """Demote a node from the master candidate role.
1340
1341     This is a single-node call.
1342
1343     """
1344     return self._SingleNodeCall(node, "node_demote_from_mc", [])
1345
1346   @_RpcTimeout(_TMO_NORMAL)
1347   def call_node_powercycle(self, node, hypervisor):
1348     """Tries to powercycle a node.
1349
1350     This is a single-node call.
1351
1352     """
1353     return self._SingleNodeCall(node, "node_powercycle", [hypervisor])
1354
1355   @_RpcTimeout(None)
1356   def call_test_delay(self, node_list, duration):
1357     """Sleep for a fixed time on given node(s).
1358
1359     This is a multi-node call.
1360
1361     """
1362     return self._MultiNodeCall(node_list, "test_delay", [duration],
1363                                read_timeout=int(duration + 5))
1364
1365   @_RpcTimeout(_TMO_FAST)
1366   def call_file_storage_dir_create(self, node, file_storage_dir):
1367     """Create the given file storage directory.
1368
1369     This is a single-node call.
1370
1371     """
1372     return self._SingleNodeCall(node, "file_storage_dir_create",
1373                                 [file_storage_dir])
1374
1375   @_RpcTimeout(_TMO_FAST)
1376   def call_file_storage_dir_remove(self, node, file_storage_dir):
1377     """Remove the given file storage directory.
1378
1379     This is a single-node call.
1380
1381     """
1382     return self._SingleNodeCall(node, "file_storage_dir_remove",
1383                                 [file_storage_dir])
1384
1385   @_RpcTimeout(_TMO_FAST)
1386   def call_file_storage_dir_rename(self, node, old_file_storage_dir,
1387                                    new_file_storage_dir):
1388     """Rename file storage directory.
1389
1390     This is a single-node call.
1391
1392     """
1393     return self._SingleNodeCall(node, "file_storage_dir_rename",
1394                                 [old_file_storage_dir, new_file_storage_dir])
1395
1396   @classmethod
1397   @_RpcTimeout(_TMO_FAST)
1398   def call_jobqueue_update(cls, node_list, address_list, file_name, content):
1399     """Update job queue.
1400
1401     This is a multi-node call.
1402
1403     """
1404     return cls._StaticMultiNodeCall(node_list, "jobqueue_update",
1405                                     [file_name, cls._Compress(content)],
1406                                     address_list=address_list)
1407
1408   @classmethod
1409   @_RpcTimeout(_TMO_NORMAL)
1410   def call_jobqueue_purge(cls, node):
1411     """Purge job queue.
1412
1413     This is a single-node call.
1414
1415     """
1416     return cls._StaticSingleNodeCall(node, "jobqueue_purge", [])
1417
1418   @classmethod
1419   @_RpcTimeout(_TMO_FAST)
1420   def call_jobqueue_rename(cls, node_list, address_list, rename):
1421     """Rename a job queue file.
1422
1423     This is a multi-node call.
1424
1425     """
1426     return cls._StaticMultiNodeCall(node_list, "jobqueue_rename", rename,
1427                                     address_list=address_list)
1428
1429   @_RpcTimeout(_TMO_NORMAL)
1430   def call_hypervisor_validate_params(self, node_list, hvname, hvparams):
1431     """Validate the hypervisor params.
1432
1433     This is a multi-node call.
1434
1435     @type node_list: list
1436     @param node_list: the list of nodes to query
1437     @type hvname: string
1438     @param hvname: the hypervisor name
1439     @type hvparams: dict
1440     @param hvparams: the hypervisor parameters to be validated
1441
1442     """
1443     cluster = self._cfg.GetClusterInfo()
1444     hv_full = objects.FillDict(cluster.hvparams.get(hvname, {}), hvparams)
1445     return self._MultiNodeCall(node_list, "hypervisor_validate_params",
1446                                [hvname, hv_full])
1447
1448   @_RpcTimeout(_TMO_NORMAL)
1449   def call_x509_cert_create(self, node, validity):
1450     """Creates a new X509 certificate for SSL/TLS.
1451
1452     This is a single-node call.
1453
1454     @type validity: int
1455     @param validity: Validity in seconds
1456
1457     """
1458     return self._SingleNodeCall(node, "x509_cert_create", [validity])
1459
1460   @_RpcTimeout(_TMO_NORMAL)
1461   def call_x509_cert_remove(self, node, name):
1462     """Removes a X509 certificate.
1463
1464     This is a single-node call.
1465
1466     @type name: string
1467     @param name: Certificate name
1468
1469     """
1470     return self._SingleNodeCall(node, "x509_cert_remove", [name])
1471
1472   @_RpcTimeout(_TMO_NORMAL)
1473   def call_import_start(self, node, opts, instance, dest, dest_args):
1474     """Starts a listener for an import.
1475
1476     This is a single-node call.
1477
1478     @type node: string
1479     @param node: Node name
1480     @type instance: C{objects.Instance}
1481     @param instance: Instance object
1482
1483     """
1484     return self._SingleNodeCall(node, "import_start",
1485                                 [opts.ToDict(),
1486                                  self._InstDict(instance), dest,
1487                                  _EncodeImportExportIO(dest, dest_args)])
1488
1489   @_RpcTimeout(_TMO_NORMAL)
1490   def call_export_start(self, node, opts, host, port,
1491                         instance, source, source_args):
1492     """Starts an export daemon.
1493
1494     This is a single-node call.
1495
1496     @type node: string
1497     @param node: Node name
1498     @type instance: C{objects.Instance}
1499     @param instance: Instance object
1500
1501     """
1502     return self._SingleNodeCall(node, "export_start",
1503                                 [opts.ToDict(), host, port,
1504                                  self._InstDict(instance), source,
1505                                  _EncodeImportExportIO(source, source_args)])
1506
1507   @_RpcTimeout(_TMO_FAST)
1508   def call_impexp_status(self, node, names):
1509     """Gets the status of an import or export.
1510
1511     This is a single-node call.
1512
1513     @type node: string
1514     @param node: Node name
1515     @type names: List of strings
1516     @param names: Import/export names
1517     @rtype: List of L{objects.ImportExportStatus} instances
1518     @return: Returns a list of the state of each named import/export or None if
1519              a status couldn't be retrieved
1520
1521     """
1522     result = self._SingleNodeCall(node, "impexp_status", [names])
1523
1524     if not result.fail_msg:
1525       decoded = []
1526
1527       for i in result.payload:
1528         if i is None:
1529           decoded.append(None)
1530           continue
1531         decoded.append(objects.ImportExportStatus.FromDict(i))
1532
1533       result.payload = decoded
1534
1535     return result
1536
1537   @_RpcTimeout(_TMO_NORMAL)
1538   def call_impexp_abort(self, node, name):
1539     """Aborts an import or export.
1540
1541     This is a single-node call.
1542
1543     @type node: string
1544     @param node: Node name
1545     @type name: string
1546     @param name: Import/export name
1547
1548     """
1549     return self._SingleNodeCall(node, "impexp_abort", [name])
1550
1551   @_RpcTimeout(_TMO_NORMAL)
1552   def call_impexp_cleanup(self, node, name):
1553     """Cleans up after an import or export.
1554
1555     This is a single-node call.
1556
1557     @type node: string
1558     @param node: Node name
1559     @type name: string
1560     @param name: Import/export name
1561
1562     """
1563     return self._SingleNodeCall(node, "impexp_cleanup", [name])