Synnefo » snf-ganeti » ganeti-local

/daemons/ganeti-confd

	lib/confd/client.py \

	lib/confd/querylib.py \

	lib/confd/server.py

	lib/server/confd.py \

if ENABLE_CONFD

	mv $(DESTDIR)$(sbindir)/hconfd $(DESTDIR)$(sbindir)/ganeti-confd

endif

nodist_sbin_SCRIPTS += htools/hconfd

#

#

# Copyright (C) 2009, 2012 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.

"""Ganeti configuration daemon queries library.

"""

import logging

from ganeti import constants

# constants for some common errors to return from a query

QUERY_UNKNOWN_ENTRY_ERROR = (constants.CONFD_REPL_STATUS_ERROR,

                             constants.CONFD_ERROR_UNKNOWN_ENTRY)

QUERY_INTERNAL_ERROR = (constants.CONFD_REPL_STATUS_ERROR,

                        constants.CONFD_ERROR_INTERNAL)

QUERY_ARGUMENT_ERROR = (constants.CONFD_REPL_STATUS_ERROR,

                        constants.CONFD_ERROR_ARGUMENT)

class ConfdQuery(object):

  """Confd Query base class.

  """

  def __init__(self, reader):

    """Constructor for Confd Query

    @type reader: L{ssconf.SimpleConfigReader}

    @param reader: ConfigReader to use to access the config

    """

    self.reader = reader

  def Exec(self, query): # pylint: disable=R0201,W0613

    """Process a single UDP request from a client.

    Different queries should override this function, which by defaults returns

    a "non-implemented" answer.

    @type query: (undefined)

    @param query: ConfdRequest 'query' field

    @rtype: (integer, undefined)

    @return: status and answer to give to the client

    """

    status = constants.CONFD_REPL_STATUS_NOTIMPLEMENTED

    answer = "not implemented"

    return status, answer

class PingQuery(ConfdQuery):

  """An empty confd query.

  It will return success on an empty argument, and an error on any other

  argument.

  """

  def Exec(self, query):

    """PingQuery main execution.

    """

    if query is None:

      status = constants.CONFD_REPL_STATUS_OK

      answer = "ok"

    else:

      status = constants.CONFD_REPL_STATUS_ERROR

      answer = "non-empty ping query"

    return status, answer

class ClusterMasterQuery(ConfdQuery):

  """Cluster master query.

  It accepts no arguments, and returns the current cluster master.

  """

  def _GetMasterNode(self):

    return self.reader.GetMasterNode()

  def Exec(self, query):

    """ClusterMasterQuery main execution

    """

    if isinstance(query, dict):

      if constants.CONFD_REQQ_FIELDS in query:

        status = constants.CONFD_REPL_STATUS_OK

        req_fields = query[constants.CONFD_REQQ_FIELDS]

        if not isinstance(req_fields, (list, tuple)):

          logging.debug("FIELDS request should be a list")

          return QUERY_ARGUMENT_ERROR

        answer = []

        for field in req_fields:

          if field == constants.CONFD_REQFIELD_NAME:

            answer.append(self._GetMasterNode())

          elif field == constants.CONFD_REQFIELD_IP:

            answer.append(self.reader.GetMasterIP())

          elif field == constants.CONFD_REQFIELD_MNODE_PIP:

            answer.append(self.reader.GetNodePrimaryIp(self._GetMasterNode()))

      else:

        logging.debug("missing FIELDS in query dict")

        return QUERY_ARGUMENT_ERROR

    elif not query:

      status = constants.CONFD_REPL_STATUS_OK

      answer = self.reader.GetMasterNode()

    else:

      logging.debug("Invalid master query argument: not dict or empty")

      return QUERY_ARGUMENT_ERROR

    return status, answer

class NodeRoleQuery(ConfdQuery):

  """A query for the role of a node.

  It will return one of CONFD_NODE_ROLE_*, or an error for non-existing nodes.

  """

  def Exec(self, query):

    """EmptyQuery main execution

    """

    node = query

    if self.reader.GetMasterNode() == node:

      status = constants.CONFD_REPL_STATUS_OK

      answer = constants.CONFD_NODE_ROLE_MASTER

      return status, answer

    flags = self.reader.GetNodeStatusFlags(node)

    if flags is None:

      return QUERY_UNKNOWN_ENTRY_ERROR

    master_candidate, drained, offline = flags

    if master_candidate:

      answer = constants.CONFD_NODE_ROLE_CANDIDATE

    elif drained:

      answer = constants.CONFD_NODE_ROLE_DRAINED

    elif offline:

      answer = constants.CONFD_NODE_ROLE_OFFLINE

    else:

      answer = constants.CONFD_NODE_ROLE_REGULAR

    return constants.CONFD_REPL_STATUS_OK, answer

class InstanceIpToNodePrimaryIpQuery(ConfdQuery):

  """A query for the location of one or more instance's ips.

  """

  def Exec(self, query):

    """InstanceIpToNodePrimaryIpQuery main execution.

    @type query: string or dict

    @param query: instance ip or dict containing:

                  constants.CONFD_REQQ_LINK: nic link (optional)

                  constants.CONFD_REQQ_IPLIST: list of ips

                  constants.CONFD_REQQ_IP: single ip

                  (one IP type request is mandatory)

    @rtype: (integer, ...)

    @return: ((status, answer) or (success, [(status, answer)...])

    """

    if isinstance(query, dict):

      if constants.CONFD_REQQ_IP in query:

        instances_list = [query[constants.CONFD_REQQ_IP]]

        mode = constants.CONFD_REQQ_IP

      elif constants.CONFD_REQQ_IPLIST in query:

        instances_list = query[constants.CONFD_REQQ_IPLIST]

        mode = constants.CONFD_REQQ_IPLIST

      else:

        logging.debug("missing IP or IPLIST in query dict")

        return QUERY_ARGUMENT_ERROR

      if constants.CONFD_REQQ_LINK in query:

        network_link = query[constants.CONFD_REQQ_LINK]

      else:

        network_link = None # default will be used

    else:

      logging.debug("Invalid query argument type for: %s", query)

      return QUERY_ARGUMENT_ERROR

    pnodes_list = []

    for instance_ip in instances_list:

      if not isinstance(instance_ip, basestring):

        logging.debug("Invalid IP type for: %s", instance_ip)

        return QUERY_ARGUMENT_ERROR

      instance = self.reader.GetInstanceByLinkIp(instance_ip, network_link)

      if not instance:

        logging.debug("Unknown instance IP: %s", instance_ip)

        pnodes_list.append(QUERY_UNKNOWN_ENTRY_ERROR)

        continue

      pnode = self.reader.GetInstancePrimaryNode(instance)

      if not pnode:

        logging.error("Instance '%s' doesn't have an associated primary"

                      " node", instance)

        pnodes_list.append(QUERY_INTERNAL_ERROR)

        continue

      pnode_primary_ip = self.reader.GetNodePrimaryIp(pnode)

      if not pnode_primary_ip:

        logging.error("Primary node '%s' doesn't have an associated"

                      " primary IP", pnode)

        pnodes_list.append(QUERY_INTERNAL_ERROR)

        continue

      pnodes_list.append((constants.CONFD_REPL_STATUS_OK, pnode_primary_ip))

    # If a single ip was requested, return a single answer, otherwise

    # the whole list, with a success status (since each entry has its

    # own success/failure)

    if mode == constants.CONFD_REQQ_IP:

      return pnodes_list[0]

    return constants.CONFD_REPL_STATUS_OK, pnodes_list

class NodesPipsQuery(ConfdQuery):

  """A query for nodes primary IPs.

  It returns the list of nodes primary IPs.

  """

  def Exec(self, query):

    """NodesPipsQuery main execution.

    """

    if query is None:

      status = constants.CONFD_REPL_STATUS_OK

      answer = self.reader.GetNodesPrimaryIps()

    else:

      status = constants.CONFD_REPL_STATUS_ERROR

      answer = "non-empty node primary IPs query"

    return status, answer

class MasterCandidatesPipsQuery(ConfdQuery):

  """A query for master candidates primary IPs.

  It returns the list of master candidates primary IPs.

  """

  def Exec(self, query):

    """MasterCandidatesPipsQuery main execution.

    """

    if query is None:

      status = constants.CONFD_REPL_STATUS_OK

      answer = self.reader.GetMasterCandidatesPrimaryIps()

    else:

      status = constants.CONFD_REPL_STATUS_ERROR

      answer = "non-empty master candidates primary IPs query"

    return status, answer

class InstancesIpsQuery(ConfdQuery):

  """A query for instances IPs.

  It returns the list of IPs of NICs connected to the requested link or all the

  instances IPs if no link is submitted.

  """

  def Exec(self, query):

    """InstancesIpsQuery main execution.

    """

    link = query

    status = constants.CONFD_REPL_STATUS_OK

    answer = self.reader.GetInstancesIps(link)

    return status, answer

class NodeDrbdQuery(ConfdQuery):

  """A query for node drbd minors.

  This is not implemented in the Python confd.

  """

#

#

# Copyright (C) 2009, 2012 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.

"""Ganeti configuration daemon server library.

Ganeti-confd is a daemon to query master candidates for configuration values.

It uses UDP+HMAC for authentication with a global cluster key.

"""

import logging

import time

from ganeti import constants

from ganeti import objects

from ganeti import errors

from ganeti import utils

from ganeti import serializer

from ganeti import ssconf

from ganeti import pathutils

from ganeti.confd import querylib

class ConfdProcessor(object):

  """A processor for confd requests.

  @ivar reader: confd SimpleConfigReader

  @ivar disabled: whether confd serving is disabled

  """

  DISPATCH_TABLE = {

    constants.CONFD_REQ_PING: querylib.PingQuery,

    constants.CONFD_REQ_NODE_ROLE_BYNAME: querylib.NodeRoleQuery,

    constants.CONFD_REQ_NODE_PIP_BY_INSTANCE_IP:

      querylib.InstanceIpToNodePrimaryIpQuery,

    constants.CONFD_REQ_CLUSTER_MASTER: querylib.ClusterMasterQuery,

    constants.CONFD_REQ_NODE_PIP_LIST: querylib.NodesPipsQuery,

    constants.CONFD_REQ_MC_PIP_LIST: querylib.MasterCandidatesPipsQuery,

    constants.CONFD_REQ_INSTANCES_IPS_LIST: querylib.InstancesIpsQuery,

    constants.CONFD_REQ_NODE_DRBD: querylib.NodeDrbdQuery,

    }

  def __init__(self):

    """Constructor for ConfdProcessor

    """

    self.disabled = True

    self.hmac_key = utils.ReadFile(pathutils.CONFD_HMAC_KEY)

    self.reader = None

    assert \

      not constants.CONFD_REQS.symmetric_difference(self.DISPATCH_TABLE), \

      "DISPATCH_TABLE is unaligned with CONFD_REQS"

  def Enable(self):

    try:

      self.reader = ssconf.SimpleConfigReader()

      self.disabled = False

    except errors.ConfigurationError:

      self.disabled = True

      raise

  def Disable(self):

    self.disabled = True

    self.reader = None

  def ExecQuery(self, payload_in, ip, port):

    """Process a single UDP request from a client.

    @type payload_in: string

    @param payload_in: request raw data

    @type ip: string

    @param ip: source ip address

    @param port: integer

    @type port: source port

    """

    if self.disabled:

      logging.debug("Confd is disabled. Ignoring query.")

      return

    try:

      request = self.ExtractRequest(payload_in)

      reply, rsalt = self.ProcessRequest(request)

      payload_out = self.PackReply(reply, rsalt)

      return payload_out

    except errors.ConfdRequestError, err:

      logging.info("Ignoring broken query from %s:%d: %s", ip, port, err)

      return None

  def ExtractRequest(self, payload):

    """Extracts a ConfdRequest object from a serialized hmac signed string.

    This functions also performs signature/timestamp validation.

    """

    current_time = time.time()

    logging.debug("Extracting request with size: %d", len(payload))

    try:

      (message, salt) = serializer.LoadSigned(payload, self.hmac_key)

    except errors.SignatureError, err:

      msg = "invalid signature: %s" % err

      raise errors.ConfdRequestError(msg)

    try:

      message_timestamp = int(salt)

    except (ValueError, TypeError):

      msg = "non-integer timestamp: %s" % salt

      raise errors.ConfdRequestError(msg)

    skew = abs(current_time - message_timestamp)

    if skew > constants.CONFD_MAX_CLOCK_SKEW:

      msg = "outside time range (skew: %d)" % skew

      raise errors.ConfdRequestError(msg)

    try:

      request = objects.ConfdRequest.FromDict(message)

    except AttributeError, err:

      raise errors.ConfdRequestError(str(err))

    return request

  def ProcessRequest(self, request):

    """Process one ConfdRequest request, and produce an answer

    @type request: L{objects.ConfdRequest}

    @rtype: (L{objects.ConfdReply}, string)

    @return: tuple of reply and salt to add to the signature

    """

    logging.debug("Processing request: %s", request)

    if request.protocol != constants.CONFD_PROTOCOL_VERSION:

      msg = "wrong protocol version %d" % request.protocol

      raise errors.ConfdRequestError(msg)

    if request.type not in constants.CONFD_REQS:

      msg = "wrong request type %d" % request.type

      raise errors.ConfdRequestError(msg)

    rsalt = request.rsalt

    if not rsalt:

      msg = "missing requested salt"

      raise errors.ConfdRequestError(msg)

    query_object = self.DISPATCH_TABLE[request.type](self.reader)

    status, answer = query_object.Exec(request.query)

    reply = objects.ConfdReply(

              protocol=constants.CONFD_PROTOCOL_VERSION,

              status=status,

              answer=answer,

              serial=self.reader.GetConfigSerialNo(),

              )

    logging.debug("Sending reply: %s", reply)

    return (reply, rsalt)

  def PackReply(self, reply, rsalt):

    """Serialize and sign the given reply, with salt rsalt

    @type reply: L{objects.ConfdReply}

    @type rsalt: string

    """

    return serializer.DumpSigned(reply.ToDict(), self.hmac_key, rsalt)

#

#

# Copyright (C) 2009, 2010 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.

"""Ganeti configuration daemon

Ganeti-confd is a daemon to query master candidates for configuration values.

It uses UDP+HMAC for authentication with a global cluster key.

"""

# pylint: disable=C0103

# C0103: Invalid name ganeti-confd

import os

import sys

import logging

import time

try:

  # pylint: disable=E0611

  from pyinotify import pyinotify

except ImportError:

  import pyinotify

from optparse import OptionParser

from ganeti import asyncnotifier

from ganeti import confd

from ganeti.confd import server as confd_server

from ganeti import constants

from ganeti import errors

from ganeti import daemon

from ganeti import netutils

from ganeti import pathutils

class ConfdAsyncUDPServer(daemon.AsyncUDPSocket):

  """The confd udp server, suitable for use with asyncore.

  """

  def __init__(self, bind_address, port, processor):

    """Constructor for ConfdAsyncUDPServer

    @type bind_address: string

    @param bind_address: socket bind address

    @type port: int

    @param port: udp port

    @type processor: L{confd.server.ConfdProcessor}

    @param processor: ConfdProcessor to use to handle queries

    """

    family = netutils.IPAddress.GetAddressFamily(bind_address)

    daemon.AsyncUDPSocket.__init__(self, family)

    self.bind_address = bind_address

    self.port = port

    self.processor = processor

    self.bind((bind_address, port))

    logging.debug("listening on ('%s':%d)", bind_address, port)

  # this method is overriding a daemon.AsyncUDPSocket method

  def handle_datagram(self, payload_in, ip, port):

    try:

      query = confd.UnpackMagic(payload_in)

    except errors.ConfdMagicError, err:

      logging.debug(err)

      return

    answer = self.processor.ExecQuery(query, ip, port)

    if answer is not None:

      try:

        self.enqueue_send(ip, port, confd.PackMagic(answer))

      except errors.UdpDataSizeError:

        logging.error("Reply too big to fit in an udp packet.")

class ConfdConfigurationReloader(object):

  """Logic to control when to reload the ganeti configuration

  This class is able to alter between inotify and polling, to rate-limit the

  number of reloads. When using inotify it also supports a fallback timed

  check, to verify that the reload hasn't failed.

  """

  def __init__(self, processor, mainloop):

    """Constructor for ConfdConfigurationReloader

    @type processor: L{confd.server.ConfdProcessor}

    @param processor: ganeti-confd ConfdProcessor

    @type mainloop: L{daemon.Mainloop}

    @param mainloop: ganeti-confd mainloop

    """

    self.processor = processor

    self.mainloop = mainloop

    self.polling = True

    self.last_notification = 0

    # Asyncronous inotify handler for config changes

    cfg_file = pathutils.CLUSTER_CONF_FILE

    self.wm = pyinotify.WatchManager()

    self.inotify_handler = asyncnotifier.SingleFileEventHandler(self.wm,

                                                                self.OnInotify,

                                                                cfg_file)

    notifier_class = asyncnotifier.ErrorLoggingAsyncNotifier

    self.notifier = notifier_class(self.wm, self.inotify_handler)

    self.timer_handle = None

    self._EnableTimer()

  def OnInotify(self, notifier_enabled):

    """Receive an inotify notification.

    @type notifier_enabled: boolean

    @param notifier_enabled: whether the notifier is still enabled

    """

    current_time = time.time()

    time_delta = current_time - self.last_notification

    self.last_notification = current_time

    if time_delta < constants.CONFD_CONFIG_RELOAD_RATELIMIT:

      logging.debug("Moving from inotify mode to polling mode")

      self.polling = True

      if notifier_enabled:

        self.inotify_handler.disable()

    if not self.polling and not notifier_enabled:

      try:

        self.inotify_handler.enable()

      except errors.InotifyError:

        self.polling = True

    try:

      reloaded = self.processor.reader.Reload()

      if reloaded:

        logging.info("Reloaded ganeti config")

      else:

        logging.debug("Skipped double config reload")

    except errors.ConfigurationError:

      self.DisableConfd()

      self.inotify_handler.disable()

      return

    # Reset the timer. If we're polling it will go to the polling rate, if

    # we're not it will delay it again to its base safe timeout.

    self._ResetTimer()

  def _DisableTimer(self):

    if self.timer_handle is not None:

      self.mainloop.scheduler.cancel(self.timer_handle)

      self.timer_handle = None

  def _EnableTimer(self):

    if self.polling:

      timeout = constants.CONFD_CONFIG_RELOAD_RATELIMIT

    else:

      timeout = constants.CONFD_CONFIG_RELOAD_TIMEOUT

    if self.timer_handle is None:

      self.timer_handle = self.mainloop.scheduler.enter(

        timeout, 1, self.OnTimer, [])

  def _ResetTimer(self):

    self._DisableTimer()

    self._EnableTimer()

  def OnTimer(self):

    """Function called when the timer fires

    """

    self.timer_handle = None

    reloaded = False

    was_disabled = False

    try:

      if self.processor.reader is None:

        was_disabled = True

        self.EnableConfd()

        reloaded = True

      else:

        reloaded = self.processor.reader.Reload()

    except errors.ConfigurationError:

      self.DisableConfd(silent=was_disabled)

      return

    if self.polling and reloaded:

      logging.info("Reloaded ganeti config")

    elif reloaded:

      # We have reloaded the config files, but received no inotify event.  If

      # an event is pending though, we just happen to have timed out before

      # receiving it, so this is not a problem, and we shouldn't alert

      if not self.notifier.check_events() and not was_disabled:

        logging.warning("Config file reload at timeout (inotify failure)")

    elif self.polling:

      # We're polling, but we haven't reloaded the config:

      # Going back to inotify mode

      logging.debug("Moving from polling mode to inotify mode")

      self.polling = False

      try:

        self.inotify_handler.enable()

      except errors.InotifyError:

        self.polling = True

    else:

      logging.debug("Performed configuration check")

    self._EnableTimer()

  def DisableConfd(self, silent=False):

    """Puts confd in non-serving mode

    """

    if not silent:

      logging.warning("Confd is being disabled")

    self.processor.Disable()

    self.polling = False

    self._ResetTimer()

  def EnableConfd(self):

    self.processor.Enable()

    logging.warning("Confd is being enabled")

    self.polling = True

    self._ResetTimer()

def CheckConfd(_, args):

  """Initial checks whether to run exit with a failure.

  """

  if args: # confd doesn't take any arguments

    print >> sys.stderr, ("Usage: %s [-f] [-d] [-b ADDRESS]" % sys.argv[0])

    sys.exit(constants.EXIT_FAILURE)

  # TODO: collapse HMAC daemons handling in daemons GenericMain, when we'll

  # have more than one.

  if not os.path.isfile(pathutils.CONFD_HMAC_KEY):

    print >> sys.stderr, "Need HMAC key %s to run" % pathutils.CONFD_HMAC_KEY

    sys.exit(constants.EXIT_FAILURE)

  # TODO: once we have a cluster param specifying the address family

  # preference, we need to check if the requested options.bind_address does not

  # conflict with that. If so, we might warn or EXIT_FAILURE.

def PrepConfd(options, _):

  """Prep confd function, executed with PID file held

  """

  # TODO: clarify how the server and reloader variables work (they are

  # not used)

  # pylint: disable=W0612

  mainloop = daemon.Mainloop()

  # Asyncronous confd UDP server

  processor = confd_server.ConfdProcessor()

  try:

    processor.Enable()

  except errors.ConfigurationError:

    # If enabling the processor has failed, we can still go on, but confd will

    # be disabled

    logging.warning("Confd is starting in disabled mode")

  server = ConfdAsyncUDPServer(options.bind_address, options.port, processor)

  # Configuration reloader

  reloader = ConfdConfigurationReloader(processor, mainloop)

  return mainloop

def ExecConfd(options, args, prep_data): # pylint: disable=W0613

  """Main confd function, executed with PID file held

  """

  mainloop = prep_data

  mainloop.Run()

def Main():

  """Main function for the confd daemon.

  """

  parser = OptionParser(description="Ganeti configuration daemon",

                        usage="%prog [-f] [-d] [-b ADDRESS]",

                        version="%%prog (ganeti) %s" %

                        constants.RELEASE_VERSION)

  daemon.GenericMain(constants.CONFD, parser, CheckConfd, PrepConfd, ExecConfd)