Synnefo » snf-ganeti » ganeti-local

#

#

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

"""Instance-related functions and classes for masterd.

"""

import logging

import time

import OpenSSL

from ganeti import constants

from ganeti import errors

from ganeti import compat

from ganeti import utils

from ganeti import objects

class _ImportExportError(Exception):

  """Local exception to report import/export errors.

  """

class ImportExportTimeouts(object):

  #: Time until daemon starts writing status file

  DEFAULT_READY_TIMEOUT = 10

  #: Length of time until errors cause hard failure

  DEFAULT_ERROR_TIMEOUT = 10

  #: Time after which daemon must be listening

  DEFAULT_LISTEN_TIMEOUT = 10

  #: Progress update interval

  DEFAULT_PROGRESS_INTERVAL = 60

  __slots__ = [

    "error",

    "ready",

    "listen",

    "connect",

    "progress",

    ]

  def __init__(self, connect,

               listen=DEFAULT_LISTEN_TIMEOUT,

               error=DEFAULT_ERROR_TIMEOUT,

               ready=DEFAULT_READY_TIMEOUT,

               progress=DEFAULT_PROGRESS_INTERVAL):

    """Initializes this class.

    @type connect: number

    @param connect: Timeout for establishing connection

    @type listen: number

    @param listen: Timeout for starting to listen for connections

    @type error: number

    @param error: Length of time until errors cause hard failure

    @type ready: number

    @param ready: Timeout for daemon to become ready

    @type progress: number

    @param progress: Progress update interval

    """

    self.error = error

    self.ready = ready

    self.listen = listen

    self.connect = connect

    self.progress = progress

class ImportExportCbBase(object):

  """Callbacks for disk import/export.

  """

  def ReportListening(self, ie, private):

    """Called when daemon started listening.

    @type ie: Subclass of L{_DiskImportExportBase}

    @param ie: Import/export object

    @param private: Private data passed to import/export object

    """

  def ReportConnected(self, ie, private):

    """Called when a connection has been established.

    @type ie: Subclass of L{_DiskImportExportBase}

    @param ie: Import/export object

    @param private: Private data passed to import/export object

    """

  def ReportProgress(self, ie, private):

    """Called when new progress information should be reported.

    @type ie: Subclass of L{_DiskImportExportBase}

    @param ie: Import/export object

    @param private: Private data passed to import/export object

    """

  def ReportFinished(self, ie, private):

    """Called when a transfer has finished.

    @type ie: Subclass of L{_DiskImportExportBase}

    @param ie: Import/export object

    @param private: Private data passed to import/export object

    """

def _TimeoutExpired(epoch, timeout, _time_fn=time.time):

  """Checks whether a timeout has expired.

  """

  return _time_fn() > (epoch + timeout)

class _DiskImportExportBase(object):

  MODE_TEXT = None

  def __init__(self, lu, node_name, opts,

               instance, timeouts, cbs, private=None):

    """Initializes this class.

    @param lu: Logical unit instance

    @type node_name: string

    @param node_name: Node name for import

    @type opts: L{objects.ImportExportOptions}

    @param opts: Import/export daemon options

    @type instance: L{objects.Instance}

    @param instance: Instance object

    @type timeouts: L{ImportExportTimeouts}

    @param timeouts: Timeouts for this import

    @type cbs: L{ImportExportCbBase}

    @param cbs: Callbacks

    @param private: Private data for callback functions

    """

    assert self.MODE_TEXT

    self._lu = lu

    self.node_name = node_name

    self._opts = opts

    self._instance = instance

    self._timeouts = timeouts

    self._cbs = cbs

    self._private = private

    # Parent loop

    self._loop = None

    # Timestamps

    self._ts_begin = None

    self._ts_connected = None

    self._ts_finished = None

    self._ts_cleanup = None

    self._ts_last_progress = None

    self._ts_last_error = None

    # Transfer status

    self.success = None

    self.final_message = None

    # Daemon status

    self._daemon_name = None

    self._daemon = None

  @property

  def recent_output(self):

    """Returns the most recent output from the daemon.

    """

    if self._daemon:

      return self._daemon.recent_output

    return None

  @property

  def progress(self):

    """Returns transfer progress information.

    """

    if not self._daemon:

      return None

    return (self._daemon.progress_mbytes,

            self._daemon.progress_throughput,

            self._daemon.progress_percent,

            self._daemon.progress_eta)

  @property

  def active(self):

    """Determines whether this transport is still active.

    """

    return self.success is None

  @property

  def loop(self):

    """Returns parent loop.

    @rtype: L{ImportExportLoop}

    """

    return self._loop

  def SetLoop(self, loop):

    """Sets the parent loop.

    @type loop: L{ImportExportLoop}

    """

    if self._loop:

      raise errors.ProgrammerError("Loop can only be set once")

    self._loop = loop

  def _StartDaemon(self):

    """Starts the import/export daemon.

    """

    raise NotImplementedError()

  def CheckDaemon(self):

    """Checks whether daemon has been started and if not, starts it.

    @rtype: string

    @return: Daemon name

    """

    assert self._ts_cleanup is None

    if self._daemon_name is None:

      assert self._ts_begin is None

      result = self._StartDaemon()

      if result.fail_msg:

        raise _ImportExportError("Failed to start %s on %s: %s" %

                                 (self.MODE_TEXT, self.node_name,

                                  result.fail_msg))

      daemon_name = result.payload

      logging.info("Started %s %r on %s", self.MODE_TEXT, daemon_name,

                   self.node_name)

      self._ts_begin = time.time()

      self._daemon_name = daemon_name

    return self._daemon_name

  def GetDaemonName(self):

    """Returns the daemon name.

    """

    assert self._daemon_name, "Daemon has not been started"

    assert self._ts_cleanup is None

    return self._daemon_name

  def Abort(self):

    """Sends SIGTERM to import/export daemon (if still active).

    """

    if self._daemon_name:

      self._lu.LogWarning("Aborting %s %r on %s",

                          self.MODE_TEXT, self._daemon_name, self.node_name)

      result = self._lu.rpc.call_impexp_abort(self.node_name, self._daemon_name)

      if result.fail_msg:

        self._lu.LogWarning("Failed to abort %s %r on %s: %s",

                            self.MODE_TEXT, self._daemon_name,

                            self.node_name, result.fail_msg)

        return False

    return True

  def _SetDaemonData(self, data):

    """Internal function for updating status daemon data.

    @type data: L{objects.ImportExportStatus}

    @param data: Daemon status data

    """

    assert self._ts_begin is not None

    if not data:

      if _TimeoutExpired(self._ts_begin, self._timeouts.ready):

        raise _ImportExportError("Didn't become ready after %s seconds" %

                                 self._timeouts.ready)

      return False

    self._daemon = data

    return True

  def SetDaemonData(self, success, data):

    """Updates daemon status data.

    @type success: bool

    @param success: Whether fetching data was successful or not

    @type data: L{objects.ImportExportStatus}

    @param data: Daemon status data

    """

    if not success:

      if self._ts_last_error is None:

        self._ts_last_error = time.time()

      elif _TimeoutExpired(self._ts_last_error, self._timeouts.error):

        raise _ImportExportError("Too many errors while updating data")

      return False

    self._ts_last_error = None

    return self._SetDaemonData(data)

  def CheckListening(self):

    """Checks whether the daemon is listening.

    """

    raise NotImplementedError()

  def _GetConnectedCheckEpoch(self):

    """Returns timeout to calculate connect timeout.

    """

    raise NotImplementedError()

  def CheckConnected(self):

    """Checks whether the daemon is connected.

    @rtype: bool

    @return: Whether the daemon is connected

    """

    assert self._daemon, "Daemon status missing"

    if self._ts_connected is not None:

      return True

    if self._daemon.connected:

      self._ts_connected = time.time()

      # TODO: Log remote peer

      logging.debug("%s %r on %s is now connected",

                    self.MODE_TEXT, self._daemon_name, self.node_name)

      self._cbs.ReportConnected(self, self._private)

      return True

    if _TimeoutExpired(self._GetConnectedCheckEpoch(), self._timeouts.connect):

      raise _ImportExportError("Not connected after %s seconds" %

                               self._timeouts.connect)

    return False

  def _CheckProgress(self):

    """Checks whether a progress update should be reported.

    """

    if ((self._ts_last_progress is None or

         _TimeoutExpired(self._ts_last_progress, self._timeouts.progress)) and

        self._daemon and

        self._daemon.progress_mbytes is not None and

        self._daemon.progress_throughput is not None):

      self._cbs.ReportProgress(self, self._private)

      self._ts_last_progress = time.time()

  def CheckFinished(self):

    """Checks whether the daemon exited.

    @rtype: bool

    @return: Whether the transfer is finished

    """

    assert self._daemon, "Daemon status missing"

    if self._ts_finished:

      return True

    if self._daemon.exit_status is None:

      # TODO: Adjust delay for ETA expiring soon

      self._CheckProgress()

      return False

    self._ts_finished = time.time()

    self._ReportFinished(self._daemon.exit_status == 0,

                         self._daemon.error_message)

    return True

  def _ReportFinished(self, success, message):

    """Transfer is finished or daemon exited.

    @type success: bool

    @param success: Whether the transfer was successful

    @type message: string

    @param message: Error message

    """

    assert self.success is None

    self.success = success

    self.final_message = message

    if success:

      logging.info("%s %r on %s succeeded", self.MODE_TEXT, self._daemon_name,

                   self.node_name)

    elif self._daemon_name:

      self._lu.LogWarning("%s %r on %s failed: %s",

                          self.MODE_TEXT, self._daemon_name, self.node_name,

                          message)

    else:

      self._lu.LogWarning("%s on %s failed: %s", self.MODE_TEXT,

                          self.node_name, message)

    self._cbs.ReportFinished(self, self._private)

  def _Finalize(self):

    """Makes the RPC call to finalize this import/export.

    """

    return self._lu.rpc.call_impexp_cleanup(self.node_name, self._daemon_name)

  def Finalize(self, error=None):

    """Finalizes this import/export.

    """

    if self._daemon_name:

      logging.info("Finalizing %s %r on %s",

                   self.MODE_TEXT, self._daemon_name, self.node_name)

      result = self._Finalize()

      if result.fail_msg:

        self._lu.LogWarning("Failed to finalize %s %r on %s: %s",

                            self.MODE_TEXT, self._daemon_name,

                            self.node_name, result.fail_msg)

        return False

      # Daemon is no longer running

      self._daemon_name = None

      self._ts_cleanup = time.time()

    if error:

      self._ReportFinished(False, error)

    return True

class DiskImport(_DiskImportExportBase):

  MODE_TEXT = "import"

  def __init__(self, lu, node_name, opts, instance,

               dest, dest_args, timeouts, cbs, private=None):

    """Initializes this class.

    @param lu: Logical unit instance

    @type node_name: string

    @param node_name: Node name for import

    @type opts: L{objects.ImportExportOptions}

    @param opts: Import/export daemon options

    @type instance: L{objects.Instance}

    @param instance: Instance object

    @param dest: I/O destination

    @param dest_args: I/O arguments

    @type timeouts: L{ImportExportTimeouts}

    @param timeouts: Timeouts for this import

    @type cbs: L{ImportExportCbBase}

    @param cbs: Callbacks

    @param private: Private data for callback functions

    """

    _DiskImportExportBase.__init__(self, lu, node_name, opts,

                                   instance, timeouts, cbs, private)

    self._dest = dest

    self._dest_args = dest_args

    # Timestamps

    self._ts_listening = None

  @property

  def listen_port(self):

    """Returns the port the daemon is listening on.

    """

    if self._daemon:

      return self._daemon.listen_port

    return None

  def _StartDaemon(self):

    """Starts the import daemon.

    """

    return self._lu.rpc.call_import_start(self.node_name, self._opts,

                                          self._instance,

                                          self._dest, self._dest_args)

  def CheckListening(self):

    """Checks whether the daemon is listening.

    @rtype: bool

    @return: Whether the daemon is listening

    """

    assert self._daemon, "Daemon status missing"

    if self._ts_listening is not None:

      return True

    port = self._daemon.listen_port

    if port is not None:

      self._ts_listening = time.time()

      logging.debug("Import %r on %s is now listening on port %s",

                    self._daemon_name, self.node_name, port)

      self._cbs.ReportListening(self, self._private)

      return True

    if _TimeoutExpired(self._ts_begin, self._timeouts.listen):

      raise _ImportExportError("Not listening after %s seconds" %

                               self._timeouts.listen)

    return False

  def _GetConnectedCheckEpoch(self):

    """Returns the time since we started listening.

    """

    assert self._ts_listening is not None, \

           ("Checking whether an import is connected is only useful"

            " once it's been listening")

    return self._ts_listening

class DiskExport(_DiskImportExportBase):

  MODE_TEXT = "export"

  def __init__(self, lu, node_name, opts,

               dest_host, dest_port, instance, source, source_args,

               timeouts, cbs, private=None):

    """Initializes this class.

    @param lu: Logical unit instance

    @type node_name: string

    @param node_name: Node name for import

    @type opts: L{objects.ImportExportOptions}

    @param opts: Import/export daemon options

    @type dest_host: string

    @param dest_host: Destination host name or IP address

    @type dest_port: number

    @param dest_port: Destination port number

    @type instance: L{objects.Instance}

    @param instance: Instance object

    @param source: I/O source

    @param source_args: I/O source

    @type timeouts: L{ImportExportTimeouts}

    @param timeouts: Timeouts for this import

    @type cbs: L{ImportExportCbBase}

    @param cbs: Callbacks

    @param private: Private data for callback functions

    """

    _DiskImportExportBase.__init__(self, lu, node_name, opts,

                                   instance, timeouts, cbs, private)

    self._dest_host = dest_host

    self._dest_port = dest_port

    self._source = source

    self._source_args = source_args

  def _StartDaemon(self):

    """Starts the export daemon.

    """

    return self._lu.rpc.call_export_start(self.node_name, self._opts,

                                          self._dest_host, self._dest_port,

                                          self._instance, self._source,

                                          self._source_args)

  def CheckListening(self):

    """Checks whether the daemon is listening.

    """

    # Only an import can be listening

    return True

  def _GetConnectedCheckEpoch(self):

    """Returns the time since the daemon started.

    """

    assert self._ts_begin is not None

    return self._ts_begin

def FormatProgress(progress):

  """Formats progress information for user consumption

  """

  (mbytes, throughput, percent, eta) = progress

  parts = [

    utils.FormatUnit(mbytes, "h"),

    # Not using FormatUnit as it doesn't support kilobytes

    "%0.1f MiB/s" % throughput,

    ]

  if percent is not None:

    parts.append("%d%%" % percent)

  if eta is not None:

    parts.append("ETA %s" % utils.FormatSeconds(eta))

  return utils.CommaJoin(parts)

class ImportExportLoop:

  MIN_DELAY = 1.0

  MAX_DELAY = 20.0

  def __init__(self, lu):

    """Initializes this class.

    """

    self._lu = lu

    self._queue = []

    self._pending_add = []

  def Add(self, diskie):

    """Adds an import/export object to the loop.

    @type diskie: Subclass of L{_DiskImportExportBase}

    @param diskie: Import/export object

    """

    assert diskie not in self._pending_add

    assert diskie.loop is None

    diskie.SetLoop(self)

    # Adding new objects to a staging list is necessary, otherwise the main

    # loop gets confused if callbacks modify the queue while the main loop is

    # iterating over it.

    self._pending_add.append(diskie)

  @staticmethod

  def _CollectDaemonStatus(lu, daemons):

    """Collects the status for all import/export daemons.

    """

    daemon_status = {}

    for node_name, names in daemons.iteritems():

      result = lu.rpc.call_impexp_status(node_name, names)

      if result.fail_msg:

        lu.LogWarning("Failed to get daemon status on %s: %s",

                      node_name, result.fail_msg)

        continue

      assert len(names) == len(result.payload)

      daemon_status[node_name] = dict(zip(names, result.payload))

    return daemon_status

  @staticmethod

  def _GetActiveDaemonNames(queue):

    """Gets the names of all active daemons.

    """

    result = {}

    for diskie in queue:

      if not diskie.active:

        continue

      try:

        # Start daemon if necessary

        daemon_name = diskie.CheckDaemon()

      except _ImportExportError, err:

        logging.exception("%s failed", diskie.MODE_TEXT)

        diskie.Finalize(error=str(err))

        continue

      result.setdefault(diskie.node_name, []).append(daemon_name)

    assert len(queue) >= len(result)

    assert len(queue) >= sum([len(names) for names in result.itervalues()])

    logging.debug("daemons=%r", result)

    return result

  def _AddPendingToQueue(self):

    """Adds all pending import/export objects to the internal queue.

    """

    assert compat.all(diskie not in self._queue and diskie.loop == self

                      for diskie in self._pending_add)

    self._queue.extend(self._pending_add)

    del self._pending_add[:]

  def Run(self):

    """Utility main loop.

    """

    while True:

      self._AddPendingToQueue()

      # Collect all active daemon names

      daemons = self._GetActiveDaemonNames(self._queue)

      if not daemons:

        break

      # Collection daemon status data

      data = self._CollectDaemonStatus(self._lu, daemons)

      # Use data

      delay = self.MAX_DELAY

      for diskie in self._queue:

        if not diskie.active:

          continue

        try:

          try:

            all_daemon_data = data[diskie.node_name]

          except KeyError:

            result = diskie.SetDaemonData(False, None)

          else:

            result = \