Synnefo » snf-ganeti

#

#

# 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.

"""Classes and functions for import/export daemon.

"""

import os

import re

import socket

import logging

import signal

import errno

import time

from cStringIO import StringIO

from ganeti import constants

from ganeti import errors

from ganeti import utils

#: Used to recognize point at which socat(1) starts to listen on its socket.

#: The local address is required for the remote peer to connect (in particular

#: the port number).

LISTENING_RE = re.compile(r"^listening on\s+"

                          r"AF=(?P<family>\d+)\s+"

                          r"(?P<address>.+):(?P<port>\d+)$", re.I)

#: Used to recognize point at which socat(1) is sending data over the wire

TRANSFER_LOOP_RE = re.compile(r"^starting data transfer loop with FDs\s+.*$",

                              re.I)

SOCAT_LOG_DEBUG = "D"

SOCAT_LOG_INFO = "I"

SOCAT_LOG_NOTICE = "N"

SOCAT_LOG_WARNING = "W"

SOCAT_LOG_ERROR = "E"

SOCAT_LOG_FATAL = "F"

SOCAT_LOG_IGNORE = frozenset([

  SOCAT_LOG_DEBUG,

  SOCAT_LOG_INFO,

  SOCAT_LOG_NOTICE,

  ])

#: Used to parse GNU dd(1) statistics

DD_INFO_RE = re.compile(r"^(?P<bytes>\d+)\s*byte(?:|s)\s.*\scopied,\s*"

                        r"(?P<seconds>[\d.]+)\s*s(?:|econds),.*$", re.I)

#: Used to ignore "N+N records in/out" on dd(1)'s stderr

DD_STDERR_IGNORE = re.compile(r"^\d+\+\d+\s*records\s+(?:in|out)$", re.I)

#: Signal upon which dd(1) will print statistics (on some platforms, SIGINFO is

#: unavailable and SIGUSR1 is used instead)

DD_INFO_SIGNAL = getattr(signal, "SIGINFO", signal.SIGUSR1)

#: Buffer size: at most this many bytes are transferred at once

BUFSIZE = 1024 * 1024

# Common options for socat

SOCAT_TCP_OPTS = ["keepalive", "keepidle=60", "keepintvl=10", "keepcnt=5"]

SOCAT_OPENSSL_OPTS = ["verify=1", "method=TLSv1",

                      "cipher=%s" % constants.OPENSSL_CIPHERS]

SOCAT_OPTION_MAXLEN = 400

(PROG_OTHER,

 PROG_SOCAT,

 PROG_DD,

 PROG_DD_PID,

 PROG_EXP_SIZE) = range(1, 6)

PROG_ALL = frozenset([

  PROG_OTHER,

  PROG_SOCAT,

  PROG_DD,

  PROG_DD_PID,

  PROG_EXP_SIZE,

  ])

class CommandBuilder(object):

  def __init__(self, mode, opts, socat_stderr_fd, dd_stderr_fd, dd_pid_fd):

    """Initializes this class.

    @param mode: Daemon mode (import or export)

    @param opts: Options object

    @type socat_stderr_fd: int

    @param socat_stderr_fd: File descriptor socat should write its stderr to

    @type dd_stderr_fd: int

    @param dd_stderr_fd: File descriptor dd should write its stderr to

    @type dd_pid_fd: int

    @param dd_pid_fd: File descriptor the child should write dd's PID to

    """

    self._opts = opts

    self._mode = mode

    self._socat_stderr_fd = socat_stderr_fd

    self._dd_stderr_fd = dd_stderr_fd

    self._dd_pid_fd = dd_pid_fd

    assert (self._opts.magic is None or

            constants.IE_MAGIC_RE.match(self._opts.magic))

  @staticmethod

  def GetBashCommand(cmd):

    """Prepares a command to be run in Bash.

    """

    return ["bash", "-o", "errexit", "-o", "pipefail", "-c", cmd]

  def _GetSocatCommand(self):

    """Returns the socat command.

    """

    common_addr_opts = SOCAT_TCP_OPTS + SOCAT_OPENSSL_OPTS + [

      "key=%s" % self._opts.key,

      "cert=%s" % self._opts.cert,

      "cafile=%s" % self._opts.ca,

      ]

    if self._opts.bind is not None:

      common_addr_opts.append("bind=%s" % self._opts.bind)

    if self._mode == constants.IEM_IMPORT:

      if self._opts.port is None:

        port = 0

      else:

        port = self._opts.port

      addr1 = [

        "OPENSSL-LISTEN:%s" % port,

        "reuseaddr",

        # Retry to listen if connection wasn't established successfully, up to

        # 100 times a second. Note that this still leaves room for DoS attacks.

        "forever",

        "intervall=0.01",

        ] + common_addr_opts

      addr2 = ["stdout"]

    elif self._mode == constants.IEM_EXPORT:

      addr1 = ["stdin"]

      addr2 = [

        "OPENSSL:%s:%s" % (self._opts.host, self._opts.port),

        # How long to wait per connection attempt

        "connect-timeout=%s" % self._opts.connect_timeout,

        # Retry a few times before giving up to connect (once per second)

        "retry=%s" % self._opts.connect_retries,

        "intervall=1",

        ] + common_addr_opts

    else:

      raise errors.GenericError("Invalid mode '%s'" % self._mode)

    for i in [addr1, addr2]:

      for value in i:

        if len(value) > SOCAT_OPTION_MAXLEN:

          raise errors.GenericError("Socat option longer than %s"

                                    " characters: %r" %

                                    (SOCAT_OPTION_MAXLEN, value))

        if "," in value:

          raise errors.GenericError("Comma not allowed in socat option"

                                    " value: %r" % value)

    return [

      constants.SOCAT_PATH,

      # Log to stderr

      "-ls",

      # Log level

      "-d", "-d",

      # Buffer size

      "-b%s" % BUFSIZE,

      # Unidirectional mode, the first address is only used for reading, and the

      # second address is only used for writing

      "-u",

      ",".join(addr1), ",".join(addr2)

      ]

  def _GetMagicCommand(self):

    """Returns the command to read/write the magic value.

    """

    if not self._opts.magic:

      return None

    # Prefix to ensure magic isn't interpreted as option to "echo"

    magic = "M=%s" % self._opts.magic

    cmd = StringIO()

    if self._mode == constants.IEM_IMPORT:

      cmd.write("{ ")

      cmd.write(utils.ShellQuoteArgs(["read", "-n", str(len(magic)), "magic"]))

      cmd.write(" && ")

      cmd.write("if test \"$magic\" != %s; then" % utils.ShellQuote(magic))

      cmd.write(" echo %s >&2;" % utils.ShellQuote("Magic value mismatch"))

      cmd.write(" exit 1;")

      cmd.write("fi;")

      cmd.write(" }")

    elif self._mode == constants.IEM_EXPORT:

      cmd.write(utils.ShellQuoteArgs(["echo", "-E", "-n", magic]))

    else:

      raise errors.GenericError("Invalid mode '%s'" % self._mode)

    return cmd.getvalue()

  def _GetDdCommand(self):

    """Returns the command for measuring throughput.

    """

    dd_cmd = StringIO()

    magic_cmd = self._GetMagicCommand()

    if magic_cmd:

      dd_cmd.write("{ ")

      dd_cmd.write(magic_cmd)

      dd_cmd.write(" && ")

    dd_cmd.write("{ ")

    # Setting LC_ALL since we want to parse the output and explicitely

    # redirecting stdin, as the background process (dd) would have /dev/null as

    # stdin otherwise

    dd_cmd.write("LC_ALL=C dd bs=%s <&0 2>&%d & pid=${!};" %

                 (BUFSIZE, self._dd_stderr_fd))

    # Send PID to daemon

    dd_cmd.write(" echo $pid >&%d;" % self._dd_pid_fd)

    # And wait for dd

    dd_cmd.write(" wait $pid;")

    dd_cmd.write(" }")

    if magic_cmd:

      dd_cmd.write(" }")

    return dd_cmd.getvalue()

  def _GetTransportCommand(self):

    """Returns the command for the transport part of the daemon.

    """

    socat_cmd = ("%s 2>&%d" %

                 (utils.ShellQuoteArgs(self._GetSocatCommand()),

                  self._socat_stderr_fd))

    dd_cmd = self._GetDdCommand()

    compr = self._opts.compress

    assert compr in constants.IEC_ALL

    parts = []

    if self._mode == constants.IEM_IMPORT:

      parts.append(socat_cmd)

      if compr == constants.IEC_GZIP:

        parts.append("gunzip -c")

      parts.append(dd_cmd)

    elif self._mode == constants.IEM_EXPORT:

      parts.append(dd_cmd)

      if compr == constants.IEC_GZIP:

        parts.append("gzip -c")

      parts.append(socat_cmd)

    else:

      raise errors.GenericError("Invalid mode '%s'" % self._mode)

    # TODO: Run transport as separate user

    # The transport uses its own shell to simplify running it as a separate user

    # in the future.

    return self.GetBashCommand(" | ".join(parts))

  def GetCommand(self):

    """Returns the complete child process command.

    """

    transport_cmd = self._GetTransportCommand()

    buf = StringIO()

    if self._opts.cmd_prefix:

      buf.write(self._opts.cmd_prefix)

      buf.write(" ")

    buf.write(utils.ShellQuoteArgs(transport_cmd))

    if self._opts.cmd_suffix:

      buf.write(" ")

      buf.write(self._opts.cmd_suffix)

    return self.GetBashCommand(buf.getvalue())

def _VerifyListening(family, address, port):

  """Verify address given as listening address by socat.

  """

  # TODO: Implement IPv6 support

  if family != socket.AF_INET:

    raise errors.GenericError("Address family %r not supported" % family)

  try:

    packed_address = socket.inet_pton(family, address)

  except socket.error:

    raise errors.GenericError("Invalid address %r for family %s" %

                              (address, family))

  return (socket.inet_ntop(family, packed_address), port)

class ChildIOProcessor(object):

  def __init__(self, debug, status_file, logger, throughput_samples, exp_size):

    """Initializes this class.

    """

    self._debug = debug

    self._status_file = status_file

    self._logger = logger

    self._splitter = dict([(prog, utils.LineSplitter(self._ProcessOutput, prog))

                           for prog in PROG_ALL])

    self._dd_pid = None

    self._dd_ready = False

    self._dd_tp_samples = throughput_samples

    self._dd_progress = []

    # Expected size of transferred data

    self._exp_size = exp_size

  def GetLineSplitter(self, prog):

    """Returns the line splitter for a program.

    """

    return self._splitter[prog]

  def FlushAll(self):

    """Flushes all line splitters.

    """

    for ls in self._splitter.itervalues():

      ls.flush()

  def CloseAll(self):

    """Closes all line splitters.

    """

    for ls in self._splitter.itervalues():

      ls.close()

    self._splitter.clear()

  def NotifyDd(self):

    """Tells dd(1) to write statistics.

    """

    if self._dd_pid is None:

      # Can't notify

      return False

    if not self._dd_ready:

      # There's a race condition between starting the program and sending

      # signals.  The signal handler is only registered after some time, so we

      # have to check whether the program is ready. If it isn't, sending a

      # signal will invoke the default handler (and usually abort the program).

      if not utils.IsProcessHandlingSignal(self._dd_pid, DD_INFO_SIGNAL):

        logging.debug("dd is not yet ready for signal %s", DD_INFO_SIGNAL)

        return False

      logging.debug("dd is now handling signal %s", DD_INFO_SIGNAL)

      self._dd_ready = True

    logging.debug("Sending signal %s to PID %s", DD_INFO_SIGNAL, self._dd_pid)

    try:

      os.kill(self._dd_pid, DD_INFO_SIGNAL)

    except EnvironmentError, err:

      if err.errno != errno.ESRCH:

        raise

      # Process no longer exists

      logging.debug("dd exited")

      self._dd_pid = None

    return True

  def _ProcessOutput(self, line, prog):

    """Takes care of child process output.

    @type line: string

    @param line: Child output line

    @type prog: number

    @param prog: Program from which the line originates

    """

    force_update = False

    forward_line = line

    if prog == PROG_SOCAT:

      level = None

      parts = line.split(None, 4)

      if len(parts) == 5:

        (_, _, _, level, msg) = parts

        force_update = self._ProcessSocatOutput(self._status_file, level, msg)

        if self._debug or (level and level not in SOCAT_LOG_IGNORE):

          forward_line = "socat: %s %s" % (level, msg)

        else:

          forward_line = None

      else:

        forward_line = "socat: %s" % line

    elif prog == PROG_DD:

      (should_forward, force_update) = self._ProcessDdOutput(line)

      if should_forward or self._debug:

        forward_line = "dd: %s" % line

      else:

        forward_line = None

    elif prog == PROG_DD_PID:

      if self._dd_pid:

        raise RuntimeError("dd PID reported more than once")

      logging.debug("Received dd PID %r", line)

      self._dd_pid = int(line)

      forward_line = None

    elif prog == PROG_EXP_SIZE:

      logging.debug("Received predicted size %r", line)

      forward_line = None

      if line:

        try:

          exp_size = utils.BytesToMebibyte(int(line))

        except (ValueError, TypeError), err:

          logging.error("Failed to convert predicted size %r to number: %s",

                        line, err)

          exp_size = None

      else:

        exp_size = None

      self._exp_size = exp_size

    if forward_line:

      self._logger.info(forward_line)

      self._status_file.AddRecentOutput(forward_line)

    self._status_file.Update(force_update)

  @staticmethod

  def _ProcessSocatOutput(status_file, level, msg):

    """Interprets socat log output.

    """

    if level == SOCAT_LOG_NOTICE:

      if status_file.GetListenPort() is None:

        # TODO: Maybe implement timeout to not listen forever

        m = LISTENING_RE.match(msg)

        if m:

          (_, port) = _VerifyListening(int(m.group("family")),

                                       m.group("address"),

                                       int(m.group("port")))

          status_file.SetListenPort(port)

          return True

      if not status_file.GetConnected():

        m = TRANSFER_LOOP_RE.match(msg)

        if m:

          logging.debug("Connection established")

          status_file.SetConnected()

          return True

    return False

  def _ProcessDdOutput(self, line):

    """Interprets a line of dd(1)'s output.

    """

    m = DD_INFO_RE.match(line)

    if m:

      seconds = float(m.group("seconds"))

      mbytes = utils.BytesToMebibyte(int(m.group("bytes")))

      self._UpdateDdProgress(seconds, mbytes)

      return (False, True)

    m = DD_STDERR_IGNORE.match(line)

    if m:

      # Ignore

      return (False, False)

    # Forward line

    return (True, False)

  def _UpdateDdProgress(self, seconds, mbytes):

    """Updates the internal status variables for dd(1) progress.

    @type seconds: float

    @param seconds: Timestamp of this update

    @type mbytes: float

    @param mbytes: Total number of MiB transferred so far

    """

    # Add latest sample

    self._dd_progress.append((seconds, mbytes))

    # Remove old samples

    del self._dd_progress[:-self._dd_tp_samples]

    # Calculate throughput

    throughput = _CalcThroughput(self._dd_progress)

    # Calculate percent and ETA

    percent = None

    eta = None

    if self._exp_size is not None:

      if self._exp_size != 0:

        percent = max(0, min(100, (100.0 * mbytes) / self._exp_size))

      if throughput:

        eta = max(0, float(self._exp_size - mbytes) / throughput)

    self._status_file.SetProgress(mbytes, throughput, percent, eta)

def _CalcThroughput(samples):

  """Calculates the throughput in MiB/second.

  @type samples: sequence

  @param samples: List of samples, each consisting of a (timestamp, mbytes)

                  tuple

  @rtype: float or None

  @return: Throughput in MiB/second

  """

  if len(samples) < 2:

    # Can't calculate throughput

    return None

  (start_time, start_mbytes) = samples[0]

  (end_time, end_mbytes) = samples[-1]

  return (float(end_mbytes) - start_mbytes) / (float(end_time) - start_time)