Synnefo » snf-ganeti » ganeti-local

#

#

# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 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

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Inter-node RPC library.

"""

# pylint: disable-msg=C0103,R0201,R0904

# C0103: Invalid name, since call_ are not valid

# R0201: Method could be a function, we keep all rpcs instance methods

# as not to change them back and forth between static/instance methods

# if they need to start using instance attributes

# R0904: Too many public methods

import os

import logging

import zlib

import base64

import pycurl

import threading

from ganeti import utils

from ganeti import objects

from ganeti import http

from ganeti import serializer

from ganeti import constants

from ganeti import errors

from ganeti import netutils

from ganeti import ssconf

from ganeti import runtime

# pylint has a bug here, doesn't see this import

import ganeti.http.client  # pylint: disable-msg=W0611

# Timeout for connecting to nodes (seconds)

_RPC_CONNECT_TIMEOUT = 5

_RPC_CLIENT_HEADERS = [

  "Content-type: %s" % http.HTTP_APP_JSON,

  "Expect:",

  ]

# Various time constants for the timeout table

_TMO_URGENT = 60 # one minute

_TMO_FAST = 5 * 60 # five minutes

_TMO_NORMAL = 15 * 60 # 15 minutes

_TMO_SLOW = 3600 # one hour

_TMO_4HRS = 4 * 3600

_TMO_1DAY = 86400

# Timeout table that will be built later by decorators

# Guidelines for choosing timeouts:

# - call used during watcher: timeout -> 1min, _TMO_URGENT

# - trivial (but be sure it is trivial) (e.g. reading a file): 5min, _TMO_FAST

# - other calls: 15 min, _TMO_NORMAL

# - special calls (instance add, etc.): either _TMO_SLOW (1h) or huge timeouts

_TIMEOUTS = {

}

def Init():

  """Initializes the module-global HTTP client manager.

  Must be called before using any RPC function and while exactly one thread is

  running.

  """

  # curl_global_init(3) and curl_global_cleanup(3) must be called with only

  # one thread running. This check is just a safety measure -- it doesn't

  # cover all cases.

  assert threading.activeCount() == 1, \

         "Found more than one active thread when initializing pycURL"

  logging.info("Using PycURL %s", pycurl.version)

  pycurl.global_init(pycurl.GLOBAL_ALL)

def Shutdown():

  """Stops the module-global HTTP client manager.

  Must be called before quitting the program and while exactly one thread is

  running.

  """

  pycurl.global_cleanup()

def _ConfigRpcCurl(curl):

  noded_cert = str(constants.NODED_CERT_FILE)

  curl.setopt(pycurl.FOLLOWLOCATION, False)

  curl.setopt(pycurl.CAINFO, noded_cert)

  curl.setopt(pycurl.SSL_VERIFYHOST, 0)

  curl.setopt(pycurl.SSL_VERIFYPEER, True)

  curl.setopt(pycurl.SSLCERTTYPE, "PEM")

  curl.setopt(pycurl.SSLCERT, noded_cert)

  curl.setopt(pycurl.SSLKEYTYPE, "PEM")

  curl.setopt(pycurl.SSLKEY, noded_cert)

  curl.setopt(pycurl.CONNECTTIMEOUT, _RPC_CONNECT_TIMEOUT)

# Aliasing this module avoids the following warning by epydoc: "Warning: No

# information available for ganeti.rpc._RpcThreadLocal's base threading.local"

_threading = threading

class _RpcThreadLocal(_threading.local):

  def GetHttpClientPool(self):

    """Returns a per-thread HTTP client pool.

    @rtype: L{http.client.HttpClientPool}

    """

    try:

      pool = self.hcp

    except AttributeError:

      pool = http.client.HttpClientPool(_ConfigRpcCurl)

      self.hcp = pool

    return pool

# Remove module alias (see above)

del _threading

_thread_local = _RpcThreadLocal()

def _RpcTimeout(secs):

  """Timeout decorator.

  When applied to a rpc call_* function, it updates the global timeout

  table with the given function/timeout.

  """

  def decorator(f):

    name = f.__name__

    assert name.startswith("call_")

    _TIMEOUTS[name[len("call_"):]] = secs

    return f

  return decorator

def RunWithRPC(fn):

  """RPC-wrapper decorator.

  When applied to a function, it runs it with the RPC system

  initialized, and it shutsdown the system afterwards. This means the

  function must be called without RPC being initialized.

  """

  def wrapper(*args, **kwargs):

    Init()

    try:

      return fn(*args, **kwargs)

    finally:

      Shutdown()

  return wrapper

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

  failed, and therefore we use this class to encapsulate the result.

  @ivar data: the data payload, for successful results, or None

  @ivar call: the name of the RPC call

  @ivar node: the name of the node to which we made the call

  @ivar offline: whether the operation failed because the node was

      offline, as opposed to actual failure; offline=True will always

      imply failed=True, in order to allow simpler checking if

      the user doesn't care about the exact failure mode

  @ivar fail_msg: the error message if the call failed

  """

  def __init__(self, data=None, failed=False, offline=False,

               call=None, node=None):

    self.offline = offline

    self.call = call

    self.node = node

    if offline:

      self.fail_msg = "Node is marked offline"

      self.data = self.payload = None

    elif failed:

      self.fail_msg = self._EnsureErr(data)

      self.data = self.payload = None

    else:

      self.data = data

      if not isinstance(self.data, (tuple, list)):

        self.fail_msg = ("RPC layer error: invalid result type (%s)" %

                         type(self.data))

        self.payload = None

      elif len(data) != 2:

        self.fail_msg = ("RPC layer error: invalid result length (%d), "

                         "expected 2" % len(self.data))

        self.payload = None

      elif not self.data[0]:

        self.fail_msg = self._EnsureErr(self.data[1])

        self.payload = None

      else:

        # finally success

        self.fail_msg = None

        self.payload = data[1]

    for attr_name in ["call", "data", "fail_msg",

                      "node", "offline", "payload"]:

      assert hasattr(self, attr_name), "Missing attribute %s" % attr_name

  @staticmethod

  def _EnsureErr(val):

    """Helper to ensure we return a 'True' value for error."""

    if val:

      return val

    else:

      return "No error information"

  def Raise(self, msg, prereq=False, ecode=None):

    """If the result has failed, raise an OpExecError.

    This is used so that LU code doesn't have to check for each

    result, but instead can call this function.

    """

    if not self.fail_msg:

      return

    if not msg: # one could pass None for default message

      msg = ("Call '%s' to node '%s' has failed: %s" %

             (self.call, self.node, self.fail_msg))

    else:

      msg = "%s: %s" % (msg, self.fail_msg)

    if prereq:

      ec = errors.OpPrereqError

    else:

      ec = errors.OpExecError

    if ecode is not None:

      args = (msg, ecode)

    else:

      args = (msg, )

    raise ec(*args) # pylint: disable-msg=W0142

def _AddressLookup(node_list,

                   ssc=ssconf.SimpleStore,

                   nslookup_fn=netutils.Hostname.GetIP):

  """Return addresses for given node names.

  @type node_list: list

  @param node_list: List of node names

  @type ssc: class

  @param ssc: SimpleStore class that is used to obtain node->ip mappings

  @type nslookup_fn: callable

  @param nslookup_fn: function use to do NS lookup

  @rtype: list of addresses and/or None's

  @returns: List of corresponding addresses, if found

  """

  ss = ssc()

  iplist = ss.GetNodePrimaryIPList()

  family = ss.GetPrimaryIPFamily()

  addresses = []

  ipmap = dict(entry.split() for entry in iplist)

  for node in node_list:

    address = ipmap.get(node)

    if address is None:

      address = nslookup_fn(node, family=family)

    addresses.append(address)

  return addresses

class Client:

  """RPC Client class.

  This class, given a (remote) method name, a list of parameters and a

  list of nodes, will contact (in parallel) all nodes, and return a

  dict of results (key: node name, value: result).

  One current bug is that generic failure is still signaled by

  'False' result, which is not good. This overloading of values can

  cause bugs.

  """

  def __init__(self, procedure, body, port, address_lookup_fn=_AddressLookup):

    assert procedure in _TIMEOUTS, ("New RPC call not declared in the"

                                    " timeouts table")

    self.procedure = procedure

    self.body = body

    self.port = port

    self._request = {}

    self._address_lookup_fn = address_lookup_fn

  def ConnectList(self, node_list, address_list=None, read_timeout=None):

    """Add a list of nodes to the target nodes.

    @type node_list: list

    @param node_list: the list of node names to connect

    @type address_list: list or None

    @keyword address_list: either None or a list with node addresses,

        which must have the same length as the node list

    @type read_timeout: int

    @param read_timeout: overwrites default timeout for operation

    """

    if address_list is None:

      # Always use IP address instead of node name

      address_list = self._address_lookup_fn(node_list)

    assert len(node_list) == len(address_list), \

           "Name and address lists must have the same length"

    for node, address in zip(node_list, address_list):

      self.ConnectNode(node, address, read_timeout=read_timeout)

  def ConnectNode(self, name, address=None, read_timeout=None):

    """Add a node to the target list.

    @type name: str

    @param name: the node name

    @type address: str

    @param address: the node address, if known

    @type read_timeout: int

    @param read_timeout: overwrites default timeout for operation

    """

    if address is None:

      # Always use IP address instead of node name

      address = self._address_lookup_fn([name])[0]

    assert(address is not None)

    if read_timeout is None:

      read_timeout = _TIMEOUTS[self.procedure]

    self._request[name] = \

      http.client.HttpClientRequest(str(address), self.port,

                                    http.HTTP_PUT, str("/%s" % self.procedure),

                                    headers=_RPC_CLIENT_HEADERS,

                                    post_data=str(self.body),

                                    read_timeout=read_timeout)

  def GetResults(self, http_pool=None):

    """Call nodes and return results.

    @rtype: list

    @return: List of RPC results

    """

    if not http_pool:

      http_pool = _thread_local.GetHttpClientPool()

    http_pool.ProcessRequests(self._request.values())

    results = {}

    for name, req in self._request.iteritems():

      if req.success and req.resp_status_code == http.HTTP_OK:

        results[name] = RpcResult(data=serializer.LoadJson(req.resp_body),

                                  node=name, call=self.procedure)

        continue

      # TODO: Better error reporting

      if req.error:

        msg = req.error

      else:

        msg = req.resp_body

      logging.error("RPC error in %s from node %s: %s",

                    self.procedure, name, msg)

      results[name] = RpcResult(data=msg, failed=True, node=name,

                                call=self.procedure)

    return results

def _EncodeImportExportIO(ieio, ieioargs):

  """Encodes import/export I/O information.

  """

  if ieio == constants.IEIO_RAW_DISK:

    assert len(ieioargs) == 1

    return (ieioargs[0].ToDict(), )

  if ieio == constants.IEIO_SCRIPT:

    assert len(ieioargs) == 2

    return (ieioargs[0].ToDict(), ieioargs[1])

  return ieioargs

class RpcRunner(object):

  """RPC runner class"""

  def __init__(self, cfg):

    """Initialized the rpc runner.

    @type cfg:  C{config.ConfigWriter}

    @param cfg: the configuration object that will be used to get data

                about the cluster

    """

    self._cfg = cfg

    self.port = netutils.GetDaemonPort(constants.NODED)

  def _InstDict(self, instance, hvp=None, bep=None, osp=None):

    """Convert the given instance to a dict.

    This is done via the instance's ToDict() method and additionally

    we fill the hvparams with the cluster defaults.

    @type instance: L{objects.Instance}

    @param instance: an Instance object

    @type hvp: dict or None

    @param hvp: a dictionary with overridden hypervisor parameters

    @type bep: dict or None

    @param bep: a dictionary with overridden backend parameters

    @type osp: dict or None

    @param osp: a dictionary with overridden os parameters

    @rtype: dict

    @return: the instance dict, with the hvparams filled with the

        cluster defaults

    """

    idict = instance.ToDict()

    cluster = self._cfg.GetClusterInfo()

    idict["hvparams"] = cluster.FillHV(instance)

    if hvp is not None:

      idict["hvparams"].update(hvp)

    idict["beparams"] = cluster.FillBE(instance)

    if bep is not None:

      idict["beparams"].update(bep)

    idict["osparams"] = cluster.SimpleFillOS(instance.os, instance.osparams)

    if osp is not None:

      idict["osparams"].update(osp)

    for nic in idict["nics"]:

      nic['nicparams'] = objects.FillDict(

        cluster.nicparams[constants.PP_DEFAULT],

        nic['nicparams'])

    return idict

  def _ConnectList(self, client, node_list, call, read_timeout=None):

    """Helper for computing node addresses.

    @type client: L{ganeti.rpc.Client}

    @param client: a C{Client} instance

    @type node_list: list

    @param node_list: the node list we should connect

    @type call: string

    @param call: the name of the remote procedure call, for filling in

        correctly any eventual offline nodes' results

    @type read_timeout: int

    @param read_timeout: overwrites the default read timeout for the

        given operation

    """

    all_nodes = self._cfg.GetAllNodesInfo()

    name_list = []

    addr_list = []

    skip_dict = {}

    for node in node_list:

      if node in all_nodes:

        if all_nodes[node].offline:

          skip_dict[node] = RpcResult(node=node, offline=True, call=call)

          continue

        val = all_nodes[node].primary_ip

      else:

        val = None

      addr_list.append(val)

      name_list.append(node)

    if name_list:

      client.ConnectList(name_list, address_list=addr_list,

                         read_timeout=read_timeout)

    return skip_dict

  def _ConnectNode(self, client, node, call, read_timeout=None):

    """Helper for computing one node's address.

    @type client: L{ganeti.rpc.Client}

    @param client: a C{Client} instance

    @type node: str

    @param node: the node we should connect

    @type call: string

    @param call: the name of the remote procedure call, for filling in

        correctly any eventual offline nodes' results

    @type read_timeout: int

    @param read_timeout: overwrites the default read timeout for the

        given operation

    """

    node_info = self._cfg.GetNodeInfo(node)

    if node_info is not None:

      if node_info.offline:

        return RpcResult(node=node, offline=True, call=call)

      addr = node_info.primary_ip

    else:

      addr = None

    client.ConnectNode(node, address=addr, read_timeout=read_timeout)

  def _MultiNodeCall(self, node_list, procedure, args, read_timeout=None):

    """Helper for making a multi-node call

    """

    body = serializer.DumpJson(args, indent=False)

    c = Client(procedure, body, self.port)

    skip_dict = self._ConnectList(c, node_list, procedure,

                                  read_timeout=read_timeout)

    skip_dict.update(c.GetResults())

    return skip_dict

  @classmethod

  def _StaticMultiNodeCall(cls, node_list, procedure, args,

                           address_list=None, read_timeout=None):

    """Helper for making a multi-node static call

    """

    body = serializer.DumpJson(args, indent=False)

    c = Client(procedure, body, netutils.GetDaemonPort(constants.NODED))

    c.ConnectList(node_list, address_list=address_list,

                  read_timeout=read_timeout)

    return c.GetResults()

  def _SingleNodeCall(self, node, procedure, args, read_timeout=None):

    """Helper for making a single-node call

    """

    body = serializer.DumpJson(args, indent=False)

    c = Client(procedure, body, self.port)

    result = self._ConnectNode(c, node, procedure, read_timeout=read_timeout)

    if result is None:

      # we did connect, node is not offline

      result = c.GetResults()[node]

    return result

  @classmethod

  def _StaticSingleNodeCall(cls, node, procedure, args, read_timeout=None):

    """Helper for making a single-node static call

    """

    body = serializer.DumpJson(args, indent=False)

    c = Client(procedure, body, netutils.GetDaemonPort(constants.NODED))

    c.ConnectNode(node, read_timeout=read_timeout)

    return c.GetResults()[node]

  @staticmethod

  def _Compress(data):

    """Compresses a string for transport over RPC.

    Small amounts of data are not compressed.

    @type data: str

    @param data: Data

    @rtype: tuple

    @return: Encoded data to send

    """

    # Small amounts of data are not compressed

    if len(data) < 512:

      return (constants.RPC_ENCODING_NONE, data)

    # Compress with zlib and encode in base64

    return (constants.RPC_ENCODING_ZLIB_BASE64,

            base64.b64encode(zlib.compress(data, 3)))

  #

  # Begin RPC calls

  #

  @_RpcTimeout(_TMO_URGENT)

  def call_lv_list(self, node_list, vg_name):

    """Gets the logical volumes present in a given volume group.

    This is a multi-node call.

    """

    return self._MultiNodeCall(node_list, "lv_list", [vg_name])

  @_RpcTimeout(_TMO_URGENT)

  def call_vg_list(self, node_list):

    """Gets the volume group list.

    This is a multi-node call.

    """

    return self._MultiNodeCall(node_list, "vg_list", [])

  @_RpcTimeout(_TMO_NORMAL)

  def call_storage_list(self, node_list, su_name, su_args, name, fields):

    """Get list of storage units.

    This is a multi-node call.

    """

    return self._MultiNodeCall(node_list, "storage_list",

                               [su_name, su_args, name, fields])

  @_RpcTimeout(_TMO_NORMAL)

  def call_storage_modify(self, node, su_name, su_args, name, changes):

    """Modify a storage unit.

    This is a single-node call.

    """

    return self._SingleNodeCall(node, "storage_modify",

                                [su_name, su_args, name, changes])

  @_RpcTimeout(_TMO_NORMAL)

  def call_storage_execute(self, node, su_name, su_args, name, op):

    """Executes an operation on a storage unit.

    This is a single-node call.

    """

    return self._SingleNodeCall(node, "storage_execute",

                                [su_name, su_args, name, op])

  @_RpcTimeout(_TMO_URGENT)

  def call_bridges_exist(self, node, bridges_list):

    """Checks if a node has all the bridges given.

    This method checks if all bridges given in the bridges_list are

    present on the remote node, so that an instance that uses interfaces

    on those bridges can be started.

    This is a single-node call.

    """

    return self._SingleNodeCall(node, "bridges_exist", [bridges_list])

  @_RpcTimeout(_TMO_NORMAL)

  def call_instance_start(self, node, instance, hvp, bep):

    """Starts an instance.

    This is a single-node call.

    """

    idict = self._InstDict(instance, hvp=hvp, bep=bep)

    return self._SingleNodeCall(node, "instance_start", [idict])

  @_RpcTimeout(_TMO_NORMAL)

  def call_instance_shutdown(self, node, instance, timeout):

    """Stops an instance.

    This is a single-node call.

    """

    return self._SingleNodeCall(node, "instance_shutdown",

                                [self._InstDict(instance), timeout])

  @_RpcTimeout(_TMO_NORMAL)

  def call_migration_info(self, node, instance):

    """Gather the information necessary to prepare an instance migration.

    This is a single-node call.

    @type node: string

    @param node: the node on which the instance is currently running

    @type instance: C{objects.Instance}

    @param instance: the instance definition

    """

    return self._SingleNodeCall(node, "migration_info",

                                [self._InstDict(instance)])

  @_RpcTimeout(_TMO_NORMAL)

  def call_accept_instance(self, node, instance, info, target):

    """Prepare a node to accept an instance.

    This is a single-node call.

    @type node: string

    @param node: the target node for the migration

    @type instance: C{objects.Instance}

    @param instance: the instance definition

    @type info: opaque/hypervisor specific (string/data)

    @param info: result for the call_migration_info call

    @type target: string

    @param target: target hostname (usually ip address) (on the node itself)

    """

    return self._SingleNodeCall(node, "accept_instance",

                                [self._InstDict(instance), info, target])

  @_RpcTimeout(_TMO_NORMAL)

  def call_finalize_migration(self, node, instance, info, success):

    """Finalize any target-node migration specific operation.

    This is called both in case of a successful migration and in case of error

    (in which case it should abort the migration).

    This is a single-node call.

    @type node: string

    @param node: the target node for the migration

    @type instance: C{objects.Instance}

    @param instance: the instance definition

    @type info: opaque/hypervisor specific (string/data)

    @param info: result for the call_migration_info call

    @type success: boolean

    @param success: whether the migration was a success or a failure

    """

    return self._SingleNodeCall(node, "finalize_migration",

                                [self._InstDict(instance), info, success])

  @_RpcTimeout(_TMO_SLOW)

  def call_instance_migrate(self, node, instance, target, live):

    """Migrate an instance.

    This is a single-node call.

    @type node: string

    @param node: the node on which the instance is currently running

    @type instance: C{objects.Instance}

    @param instance: the instance definition

    @type target: string

    @param target: the target node name

    @type live: boolean

    @param live: whether the migration should be done live or not (the

        interpretation of this parameter is left to the hypervisor)

    """

    return self._SingleNodeCall(node, "instance_migrate",