Include hvparams in ssconf files
[ganeti-local] / lib / rpc.py
index fc56275..ad8396a 100644 (file)
@@ -1,7 +1,7 @@
 #
 #
 
-# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 Google Inc.
+# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 Google Inc.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -35,6 +35,7 @@ import zlib
 import base64
 import pycurl
 import threading
+import copy
 
 from ganeti import utils
 from ganeti import objects
@@ -47,6 +48,8 @@ from ganeti import ssconf
 from ganeti import runtime
 from ganeti import compat
 from ganeti import rpc_defs
+from ganeti import pathutils
+from ganeti import vcluster
 
 # Special module generated at build time
 from ganeti import _generated_rpc
@@ -93,7 +96,7 @@ def Shutdown():
 
 
 def _ConfigRpcCurl(curl):
-  noded_cert = str(constants.NODED_CERT_FILE)
+  noded_cert = str(pathutils.NODED_CERT_FILE)
 
   curl.setopt(pycurl.FOLLOWLOCATION, False)
   curl.setopt(pycurl.CAINFO, noded_cert)
@@ -147,7 +150,7 @@ class RpcResult(object):
   """RPC Result class.
 
   This class holds an RPC result. It is needed since in multi-node
-  calls we can't raise an exception just because one one out of many
+  calls we can't raise an exception just because one out of many
   failed, and therefore we use this class to encapsulate the result.
 
   @ivar data: the data payload, for successful results, or None
@@ -227,6 +230,19 @@ class RpcResult(object):
       args = (msg, )
     raise ec(*args) # pylint: disable=W0142
 
+  def Warn(self, msg, feedback_fn):
+    """If the result has failed, call the feedback_fn.
+
+    This is used to in cases were LU wants to warn the
+    user about a failure, but continue anyway.
+
+    """
+    if not self.fail_msg:
+      return
+
+    msg = "%s: %s" % (msg, self.fail_msg)
+    feedback_fn(msg)
+
 
 def _SsconfResolver(ssconf_ips, node_list, _,
                     ssc=ssconf.SimpleStore,
@@ -402,6 +418,8 @@ class _RpcProcessor:
     @param body: dictionary with request bodies per host
     @type read_timeout: int or None
     @param read_timeout: Read timeout for request
+    @rtype: dictionary
+    @return: a dictionary mapping host names to rpc.RpcResult objects
 
     """
     assert read_timeout is not None, \
@@ -523,7 +541,9 @@ def _PrepareFileUpload(getents_fn, filename):
 
   getents = getents_fn()
 
-  return [filename, data, st.st_mode, getents.LookupUid(st.st_uid),
+  virt_filename = vcluster.MakeVirtualPath(filename)
+
+  return [virt_filename, data, st.st_mode, getents.LookupUid(st.st_uid),
           getents.LookupGid(st.st_gid), st.st_atime, st.st_mtime]
 
 
@@ -564,18 +584,49 @@ def _EncodeBlockdevRename(value):
   return [(d.ToDict(), uid) for d, uid in value]
 
 
-def MakeLegacyNodeInfo(data):
+def BuildVgInfoQuery(cfg):
+  """Build a query about the default VG for C{node_info}.
+
+  The result of the RPC can be parsed with L{MakeLegacyNodeInfo}.
+
+  @type cfg: L{config.ConfigWriter}
+  @param cfg: Cluster configuration
+  @rtype: list
+  @return: argument suitable for L{rpc.RpcRunner.call_node_info}
+
+  """
+  vg_name = cfg.GetVGName()
+  if vg_name:
+    ret = [
+      (constants.ST_LVM_VG, vg_name),
+      (constants.ST_LVM_PV, vg_name),
+      ]
+  else:
+    ret = []
+  return ret
+
+
+def MakeLegacyNodeInfo(data, require_vg_info=True):
   """Formats the data returned by L{rpc.RpcRunner.call_node_info}.
 
   Converts the data into a single dictionary. This is fine for most use cases,
   but some require information from more than one volume group or hypervisor.
 
+  @param require_vg_info: raise an error if the returnd vg_info
+      doesn't have any values
+
   """
-  (bootid, (vg_info, ), (hv_info, )) = data
+  (bootid, vgs_info, (hv_info, )) = data
+
+  ret = utils.JoinDisjointDicts(hv_info, {"bootid": bootid})
+
+  if require_vg_info or vgs_info:
+    (vg0_info, vg0_spindles) = vgs_info
+    ret = utils.JoinDisjointDicts(vg0_info, ret)
+    ret["spindles_free"] = vg0_spindles["vg_free"]
+    ret["spindles_total"] = vg0_spindles["vg_size"]
 
-  return utils.JoinDisjointDicts(utils.JoinDisjointDicts(vg_info, hv_info), {
-    "bootid": bootid,
-    })
+  return ret
 
 
 def _AnnotateDParamsDRBD(disk, (drbd_params, data_params, meta_params)):
@@ -624,6 +675,31 @@ def AnnotateDiskParams(template, disks, disk_params):
   return [annotation_fn(disk.Copy(), ld_params) for disk in disks]
 
 
+def _GetESFlag(cfg, nodename):
+  ni = cfg.GetNodeInfo(nodename)
+  if ni is None:
+    raise errors.OpPrereqError("Invalid node name %s" % nodename,
+                               errors.ECODE_NOENT)
+  return cfg.GetNdParams(ni)[constants.ND_EXCLUSIVE_STORAGE]
+
+
+def GetExclusiveStorageForNodeNames(cfg, nodelist):
+  """Return the exclusive storage flag for all the given nodes.
+
+  @type cfg: L{config.ConfigWriter}
+  @param cfg: cluster configuration
+  @type nodelist: list or tuple
+  @param nodelist: node names for which to read the flag
+  @rtype: dict
+  @return: mapping from node names to exclusive storage flags
+  @raise errors.OpPrereqError: if any given node name has no corresponding node
+
+  """
+  getflag = lambda n: _GetESFlag(cfg, n)
+  flags = map(getflag, nodelist)
+  return dict(zip(nodelist, flags))
+
+
 #: Generic encoders
 _ENCODERS = {
   rpc_defs.ED_OBJECT_DICT: _ObjectToDict,
@@ -660,8 +736,9 @@ class RpcRunner(_RpcClientBase,
     encoders.update({
       # Encoders requiring configuration object
       rpc_defs.ED_INST_DICT: self._InstDict,
-      rpc_defs.ED_INST_DICT_HVP_BEP: self._InstDictHvpBep,
+      rpc_defs.ED_INST_DICT_HVP_BEP_DP: self._InstDictHvpBepDp,
       rpc_defs.ED_INST_DICT_OSP_DP: self._InstDictOspDp,
+      rpc_defs.ED_NIC_DICT: self._NicDict,
 
       # Encoders annotating disk parameters
       rpc_defs.ED_DISKS_DICT_DP: self._DisksDictDP,
@@ -687,6 +764,18 @@ class RpcRunner(_RpcClientBase,
     _generated_rpc.RpcClientDnsOnly.__init__(self)
     _generated_rpc.RpcClientDefault.__init__(self)
 
+  def _NicDict(self, nic):
+    """Convert the given nic to a dict and encapsulate netinfo
+
+    """
+    n = copy.deepcopy(nic)
+    if n.network:
+      net_uuid = self._cfg.LookupNetwork(n.network)
+      if net_uuid:
+        nobj = self._cfg.GetNetwork(net_uuid)
+        n.netinfo = objects.Network.ToDict(nobj)
+    return n.ToDict()
+
   def _InstDict(self, instance, hvp=None, bep=None, osp=None):
     """Convert the given instance to a dict.
 
@@ -717,13 +806,20 @@ class RpcRunner(_RpcClientBase,
     idict["osparams"] = cluster.SimpleFillOS(instance.os, instance.osparams)
     if osp is not None:
       idict["osparams"].update(osp)
+    idict["disks"] = self._DisksDictDP((instance.disks, instance))
     for nic in idict["nics"]:
       nic["nicparams"] = objects.FillDict(
         cluster.nicparams[constants.PP_DEFAULT],
         nic["nicparams"])
+      network = nic.get("network", None)
+      if network:
+        net_uuid = self._cfg.LookupNetwork(network)
+        if net_uuid:
+          nobj = self._cfg.GetNetwork(net_uuid)
+          nic["netinfo"] = objects.Network.ToDict(nobj)
     return idict
 
-  def _InstDictHvpBep(self, (instance, hvp, bep)):
+  def _InstDictHvpBepDp(self, (instance, hvp, bep)):
     """Wrapper for L{_InstDict}.
 
     """
@@ -733,9 +829,7 @@ class RpcRunner(_RpcClientBase,
     """Wrapper for L{_InstDict}.
 
     """
-    updated_inst = self._InstDict(instance, osp=osparams)
-    updated_inst["disks"] = self._DisksDictDP((instance.disks, instance))
-    return updated_inst
+    return self._InstDict(instance, osp=osparams)
 
   def _DisksDictDP(self, (disks, instance)):
     """Wrapper for L{AnnotateDiskParams}.