code.grnet.gr Git - ganeti-local/blob - lib/impexpd/__init__.py

   1 #
   2 #
   3
   4 # Copyright (C) 2010 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Classes and functions for import/export daemon.
  23
  24 """
  25
  26 import os
  27 import re
  28 import socket
  29 import logging
  30 import signal
  31 import errno
  32 import time
  33 from cStringIO import StringIO
  34
  35 from ganeti import constants
  36 from ganeti import errors
  37 from ganeti import utils
  38 from ganeti import netutils
  39
  40
  41 #: Used to recognize point at which socat(1) starts to listen on its socket.
  42 #: The local address is required for the remote peer to connect (in particular
  43 #: the port number).
  44 LISTENING_RE = re.compile(r"^listening on\s+"
  45                           r"AF=(?P<family>\d+)\s+"
  46                           r"(?P<address>.+):(?P<port>\d+)$", re.I)
  47
  48 #: Used to recognize point at which socat(1) is sending data over the wire
  49 TRANSFER_LOOP_RE = re.compile(r"^starting data transfer loop with FDs\s+.*$",
  50                               re.I)
  51
  52 SOCAT_LOG_DEBUG = "D"
  53 SOCAT_LOG_INFO = "I"
  54 SOCAT_LOG_NOTICE = "N"
  55 SOCAT_LOG_WARNING = "W"
  56 SOCAT_LOG_ERROR = "E"
  57 SOCAT_LOG_FATAL = "F"
  58
  59 SOCAT_LOG_IGNORE = frozenset([
  60   SOCAT_LOG_DEBUG,
  61   SOCAT_LOG_INFO,
  62   SOCAT_LOG_NOTICE,
  63   ])
  64
  65 #: Used to parse GNU dd(1) statistics
  66 DD_INFO_RE = re.compile(r"^(?P<bytes>\d+)\s*byte(?:|s)\s.*\scopied,\s*"
  67                         r"(?P<seconds>[\d.]+)\s*s(?:|econds),.*$", re.I)
  68
  69 #: Used to ignore "N+N records in/out" on dd(1)'s stderr
  70 DD_STDERR_IGNORE = re.compile(r"^\d+\+\d+\s*records\s+(?:in|out)$", re.I)
  71
  72 #: Signal upon which dd(1) will print statistics (on some platforms, SIGINFO is
  73 #: unavailable and SIGUSR1 is used instead)
  74 DD_INFO_SIGNAL = getattr(signal, "SIGINFO", signal.SIGUSR1)
  75
  76 #: Buffer size: at most this many bytes are transferred at once
  77 BUFSIZE = 1024 * 1024
  78
  79 # Common options for socat
  80 SOCAT_TCP_OPTS = ["keepalive", "keepidle=60", "keepintvl=10", "keepcnt=5"]
  81 SOCAT_OPENSSL_OPTS = ["verify=1", "method=TLSv1",
  82                       "cipher=%s" % constants.OPENSSL_CIPHERS]
  83
  84 if constants.SOCAT_USE_COMPRESS:
  85   # Disables all compression in by OpenSSL. Only supported in patched versions
  86   # of socat (as of November 2010). See INSTALL for more information.
  87   SOCAT_OPENSSL_OPTS.append("compress=none")
  88
  89 SOCAT_OPTION_MAXLEN = 400
  90
  91 (PROG_OTHER,
  92  PROG_SOCAT,
  93  PROG_DD,
  94  PROG_DD_PID,
  95  PROG_EXP_SIZE) = range(1, 6)
  96 PROG_ALL = frozenset([
  97   PROG_OTHER,
  98   PROG_SOCAT,
  99   PROG_DD,
 100   PROG_DD_PID,
 101   PROG_EXP_SIZE,
 102   ])
 103
 104
 105 class CommandBuilder(object):
 106   def __init__(self, mode, opts, socat_stderr_fd, dd_stderr_fd, dd_pid_fd):
 107     """Initializes this class.
 108
 109     @param mode: Daemon mode (import or export)
 110     @param opts: Options object
 111     @type socat_stderr_fd: int
 112     @param socat_stderr_fd: File descriptor socat should write its stderr to
 113     @type dd_stderr_fd: int
 114     @param dd_stderr_fd: File descriptor dd should write its stderr to
 115     @type dd_pid_fd: int
 116     @param dd_pid_fd: File descriptor the child should write dd's PID to
 117
 118     """
 119     self._opts = opts
 120     self._mode = mode
 121     self._socat_stderr_fd = socat_stderr_fd
 122     self._dd_stderr_fd = dd_stderr_fd
 123     self._dd_pid_fd = dd_pid_fd
 124
 125     assert (self._opts.magic is None or
 126             constants.IE_MAGIC_RE.match(self._opts.magic))
 127
 128   @staticmethod
 129   def GetBashCommand(cmd):
 130     """Prepares a command to be run in Bash.
 131
 132     """
 133     return ["bash", "-o", "errexit", "-o", "pipefail", "-c", cmd]
 134
 135   def _GetSocatCommand(self):
 136     """Returns the socat command.
 137
 138     """
 139     common_addr_opts = SOCAT_TCP_OPTS + SOCAT_OPENSSL_OPTS + [
 140       "key=%s" % self._opts.key,
 141       "cert=%s" % self._opts.cert,
 142       "cafile=%s" % self._opts.ca,
 143       ]
 144
 145     if self._opts.bind is not None:
 146       common_addr_opts.append("bind=%s" % self._opts.bind)
 147
 148     assert not (self._opts.ipv4 and self._opts.ipv6)
 149
 150     if self._opts.ipv4:
 151       common_addr_opts.append("pf=ipv4")
 152     elif self._opts.ipv6:
 153       common_addr_opts.append("pf=ipv6")
 154
 155     if self._mode == constants.IEM_IMPORT:
 156       if self._opts.port is None:
 157         port = 0
 158       else:
 159         port = self._opts.port
 160
 161       addr1 = [
 162         "OPENSSL-LISTEN:%s" % port,
 163         "reuseaddr",
 164
 165         # Retry to listen if connection wasn't established successfully, up to
 166         # 100 times a second. Note that this still leaves room for DoS attacks.
 167         "forever",
 168         "intervall=0.01",
 169         ] + common_addr_opts
 170       addr2 = ["stdout"]
 171
 172     elif self._mode == constants.IEM_EXPORT:
 173       if self._opts.host and netutils.IP6Address.IsValid(self._opts.host):
 174         host = "[%s]" % self._opts.host
 175       else:
 176         host = self._opts.host
 177
 178       addr1 = ["stdin"]
 179       addr2 = [
 180         "OPENSSL:%s:%s" % (host, self._opts.port),
 181
 182         # How long to wait per connection attempt
 183         "connect-timeout=%s" % self._opts.connect_timeout,
 184
 185         # Retry a few times before giving up to connect (once per second)
 186         "retry=%s" % self._opts.connect_retries,
 187         "intervall=1",
 188         ] + common_addr_opts
 189
 190     else:
 191       raise errors.GenericError("Invalid mode '%s'" % self._mode)
 192
 193     for i in [addr1, addr2]:
 194       for value in i:
 195         if len(value) > SOCAT_OPTION_MAXLEN:
 196           raise errors.GenericError("Socat option longer than %s"
 197                                     " characters: %r" %
 198                                     (SOCAT_OPTION_MAXLEN, value))
 199         if "," in value:
 200           raise errors.GenericError("Comma not allowed in socat option"
 201                                     " value: %r" % value)
 202
 203     return [
 204       constants.SOCAT_PATH,
 205
 206       # Log to stderr
 207       "-ls",
 208
 209       # Log level
 210       "-d", "-d",
 211
 212       # Buffer size
 213       "-b%s" % BUFSIZE,
 214
 215       # Unidirectional mode, the first address is only used for reading, and the
 216       # second address is only used for writing
 217       "-u",
 218
 219       ",".join(addr1), ",".join(addr2)
 220       ]
 221
 222   def _GetMagicCommand(self):
 223     """Returns the command to read/write the magic value.
 224
 225     """
 226     if not self._opts.magic:
 227       return None
 228
 229     # Prefix to ensure magic isn't interpreted as option to "echo"
 230     magic = "M=%s" % self._opts.magic
 231
 232     cmd = StringIO()
 233
 234     if self._mode == constants.IEM_IMPORT:
 235       cmd.write("{ ")
 236       cmd.write(utils.ShellQuoteArgs(["read", "-n", str(len(magic)), "magic"]))
 237       cmd.write(" && ")
 238       cmd.write("if test \"$magic\" != %s; then" % utils.ShellQuote(magic))
 239       cmd.write(" echo %s >&2;" % utils.ShellQuote("Magic value mismatch"))
 240       cmd.write(" exit 1;")
 241       cmd.write("fi;")
 242       cmd.write(" }")
 243
 244     elif self._mode == constants.IEM_EXPORT:
 245       cmd.write(utils.ShellQuoteArgs(["echo", "-E", "-n", magic]))
 246
 247     else:
 248       raise errors.GenericError("Invalid mode '%s'" % self._mode)
 249
 250     return cmd.getvalue()
 251
 252   def _GetDdCommand(self):
 253     """Returns the command for measuring throughput.
 254
 255     """
 256     dd_cmd = StringIO()
 257
 258     magic_cmd = self._GetMagicCommand()
 259     if magic_cmd:
 260       dd_cmd.write("{ ")
 261       dd_cmd.write(magic_cmd)
 262       dd_cmd.write(" && ")
 263
 264     dd_cmd.write("{ ")
 265     # Setting LC_ALL since we want to parse the output and explicitely
 266     # redirecting stdin, as the background process (dd) would have /dev/null as
 267     # stdin otherwise
 268     dd_cmd.write("LC_ALL=C dd bs=%s <&0 2>&%d & pid=${!};" %
 269                  (BUFSIZE, self._dd_stderr_fd))
 270     # Send PID to daemon
 271     dd_cmd.write(" echo $pid >&%d;" % self._dd_pid_fd)
 272     # And wait for dd
 273     dd_cmd.write(" wait $pid;")
 274     dd_cmd.write(" }")
 275
 276     if magic_cmd:
 277       dd_cmd.write(" }")
 278
 279     return dd_cmd.getvalue()
 280
 281   def _GetTransportCommand(self):
 282     """Returns the command for the transport part of the daemon.
 283
 284     """
 285     socat_cmd = ("%s 2>&%d" %
 286                  (utils.ShellQuoteArgs(self._GetSocatCommand()),
 287                   self._socat_stderr_fd))
 288     dd_cmd = self._GetDdCommand()
 289
 290     compr = self._opts.compress
 291
 292     assert compr in constants.IEC_ALL
 293
 294     parts = []
 295
 296     if self._mode == constants.IEM_IMPORT:
 297       parts.append(socat_cmd)
 298
 299       if compr == constants.IEC_GZIP:
 300         parts.append("gunzip -c")
 301
 302       parts.append(dd_cmd)
 303
 304     elif self._mode == constants.IEM_EXPORT:
 305       parts.append(dd_cmd)
 306
 307       if compr == constants.IEC_GZIP:
 308         parts.append("gzip -c")
 309
 310       parts.append(socat_cmd)
 311
 312     else:
 313       raise errors.GenericError("Invalid mode '%s'" % self._mode)
 314
 315     # TODO: Run transport as separate user
 316     # The transport uses its own shell to simplify running it as a separate user
 317     # in the future.
 318     return self.GetBashCommand(" | ".join(parts))
 319
 320   def GetCommand(self):
 321     """Returns the complete child process command.
 322
 323     """
 324     transport_cmd = self._GetTransportCommand()
 325
 326     buf = StringIO()
 327
 328     if self._opts.cmd_prefix:
 329       buf.write(self._opts.cmd_prefix)
 330       buf.write(" ")
 331
 332     buf.write(utils.ShellQuoteArgs(transport_cmd))
 333
 334     if self._opts.cmd_suffix:
 335       buf.write(" ")
 336       buf.write(self._opts.cmd_suffix)
 337
 338     return self.GetBashCommand(buf.getvalue())
 339
 340
 341 def _VerifyListening(family, address, port):
 342   """Verify address given as listening address by socat.
 343
 344   """
 345   if family not in (socket.AF_INET, socket.AF_INET6):
 346     raise errors.GenericError("Address family %r not supported" % family)
 347
 348   if (family == socket.AF_INET6 and address.startswith("[") and
 349       address.endswith("]")):
 350     address = address.lstrip("[").rstrip("]")
 351
 352   try:
 353     packed_address = socket.inet_pton(family, address)
 354   except socket.error:
 355     raise errors.GenericError("Invalid address %r for family %s" %
 356                               (address, family))
 357
 358   return (socket.inet_ntop(family, packed_address), port)
 359
 360
 361 class ChildIOProcessor(object):
 362   def __init__(self, debug, status_file, logger, throughput_samples, exp_size):
 363     """Initializes this class.
 364
 365     """
 366     self._debug = debug
 367     self._status_file = status_file
 368     self._logger = logger
 369
 370     self._splitter = dict([(prog, utils.LineSplitter(self._ProcessOutput, prog))
 371                            for prog in PROG_ALL])
 372
 373     self._dd_pid = None
 374     self._dd_ready = False
 375     self._dd_tp_samples = throughput_samples
 376     self._dd_progress = []
 377
 378     # Expected size of transferred data
 379     self._exp_size = exp_size
 380
 381   def GetLineSplitter(self, prog):
 382     """Returns the line splitter for a program.
 383
 384     """
 385     return self._splitter[prog]
 386
 387   def FlushAll(self):
 388     """Flushes all line splitters.
 389
 390     """
 391     for ls in self._splitter.itervalues():
 392       ls.flush()
 393
 394   def CloseAll(self):
 395     """Closes all line splitters.
 396
 397     """
 398     for ls in self._splitter.itervalues():
 399       ls.close()
 400     self._splitter.clear()
 401
 402   def NotifyDd(self):
 403     """Tells dd(1) to write statistics.
 404
 405     """
 406     if self._dd_pid is None:
 407       # Can't notify
 408       return False
 409
 410     if not self._dd_ready:
 411       # There's a race condition between starting the program and sending
 412       # signals.  The signal handler is only registered after some time, so we
 413       # have to check whether the program is ready. If it isn't, sending a
 414       # signal will invoke the default handler (and usually abort the program).
 415       if not utils.IsProcessHandlingSignal(self._dd_pid, DD_INFO_SIGNAL):
 416         logging.debug("dd is not yet ready for signal %s", DD_INFO_SIGNAL)
 417         return False
 418
 419       logging.debug("dd is now handling signal %s", DD_INFO_SIGNAL)
 420       self._dd_ready = True
 421
 422     logging.debug("Sending signal %s to PID %s", DD_INFO_SIGNAL, self._dd_pid)
 423     try:
 424       os.kill(self._dd_pid, DD_INFO_SIGNAL)
 425     except EnvironmentError, err:
 426       if err.errno != errno.ESRCH:
 427         raise
 428
 429       # Process no longer exists
 430       logging.debug("dd exited")
 431       self._dd_pid = None
 432
 433     return True
 434
 435   def _ProcessOutput(self, line, prog):
 436     """Takes care of child process output.
 437
 438     @type line: string
 439     @param line: Child output line
 440     @type prog: number
 441     @param prog: Program from which the line originates
 442
 443     """
 444     force_update = False
 445     forward_line = line
 446
 447     if prog == PROG_SOCAT:
 448       level = None
 449       parts = line.split(None, 4)
 450
 451       if len(parts) == 5:
 452         (_, _, _, level, msg) = parts
 453
 454         force_update = self._ProcessSocatOutput(self._status_file, level, msg)
 455
 456         if self._debug or (level and level not in SOCAT_LOG_IGNORE):
 457           forward_line = "socat: %s %s" % (level, msg)
 458         else:
 459           forward_line = None
 460       else:
 461         forward_line = "socat: %s" % line
 462
 463     elif prog == PROG_DD:
 464       (should_forward, force_update) = self._ProcessDdOutput(line)
 465
 466       if should_forward or self._debug:
 467         forward_line = "dd: %s" % line
 468       else:
 469         forward_line = None
 470
 471     elif prog == PROG_DD_PID:
 472       if self._dd_pid:
 473         raise RuntimeError("dd PID reported more than once")
 474       logging.debug("Received dd PID %r", line)
 475       self._dd_pid = int(line)
 476       forward_line = None
 477
 478     elif prog == PROG_EXP_SIZE:
 479       logging.debug("Received predicted size %r", line)
 480       forward_line = None
 481
 482       if line:
 483         try:
 484           exp_size = utils.BytesToMebibyte(int(line))
 485         except (ValueError, TypeError), err:
 486           logging.error("Failed to convert predicted size %r to number: %s",
 487                         line, err)
 488           exp_size = None
 489       else:
 490         exp_size = None
 491
 492       self._exp_size = exp_size
 493
 494     if forward_line:
 495       self._logger.info(forward_line)
 496       self._status_file.AddRecentOutput(forward_line)
 497
 498     self._status_file.Update(force_update)
 499
 500   @staticmethod
 501   def _ProcessSocatOutput(status_file, level, msg):
 502     """Interprets socat log output.
 503
 504     """
 505     if level == SOCAT_LOG_NOTICE:
 506       if status_file.GetListenPort() is None:
 507         # TODO: Maybe implement timeout to not listen forever
 508         m = LISTENING_RE.match(msg)
 509         if m:
 510           (_, port) = _VerifyListening(int(m.group("family")),
 511                                        m.group("address"),
 512                                        int(m.group("port")))
 513
 514           status_file.SetListenPort(port)
 515           return True
 516
 517       if not status_file.GetConnected():
 518         m = TRANSFER_LOOP_RE.match(msg)
 519         if m:
 520           logging.debug("Connection established")
 521           status_file.SetConnected()
 522           return True
 523
 524     return False
 525
 526   def _ProcessDdOutput(self, line):
 527     """Interprets a line of dd(1)'s output.
 528
 529     """
 530     m = DD_INFO_RE.match(line)
 531     if m:
 532       seconds = float(m.group("seconds"))
 533       mbytes = utils.BytesToMebibyte(int(m.group("bytes")))
 534       self._UpdateDdProgress(seconds, mbytes)
 535       return (False, True)
 536
 537     m = DD_STDERR_IGNORE.match(line)
 538     if m:
 539       # Ignore
 540       return (False, False)
 541
 542     # Forward line
 543     return (True, False)
 544
 545   def _UpdateDdProgress(self, seconds, mbytes):
 546     """Updates the internal status variables for dd(1) progress.
 547
 548     @type seconds: float
 549     @param seconds: Timestamp of this update
 550     @type mbytes: float
 551     @param mbytes: Total number of MiB transferred so far
 552
 553     """
 554     # Add latest sample
 555     self._dd_progress.append((seconds, mbytes))
 556
 557     # Remove old samples
 558     del self._dd_progress[:-self._dd_tp_samples]
 559
 560     # Calculate throughput
 561     throughput = _CalcThroughput(self._dd_progress)
 562
 563     # Calculate percent and ETA
 564     percent = None
 565     eta = None
 566
 567     if self._exp_size is not None:
 568       if self._exp_size != 0:
 569         percent = max(0, min(100, (100.0 * mbytes) / self._exp_size))
 570
 571       if throughput:
 572         eta = max(0, float(self._exp_size - mbytes) / throughput)
 573
 574     self._status_file.SetProgress(mbytes, throughput, percent, eta)
 575
 576
 577 def _CalcThroughput(samples):
 578   """Calculates the throughput in MiB/second.
 579
 580   @type samples: sequence
 581   @param samples: List of samples, each consisting of a (timestamp, mbytes)
 582                   tuple
 583   @rtype: float or None
 584   @return: Throughput in MiB/second
 585
 586   """
 587   if len(samples) < 2:
 588     # Can't calculate throughput
 589     return None
 590
 591   (start_time, start_mbytes) = samples[0]
 592   (end_time, end_mbytes) = samples[-1]
 593
 594   return (float(end_mbytes) - start_mbytes) / (float(end_time) - start_time)