4 # Copyright (C) 2010 Google Inc.
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.
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.
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
22 """Classes and functions for import/export daemon.
33 from cStringIO import StringIO
35 from ganeti import constants
36 from ganeti import errors
37 from ganeti import utils
40 #: Used to recognize point at which socat(1) starts to listen on its socket.
41 #: The local address is required for the remote peer to connect (in particular
43 LISTENING_RE = re.compile(r"^listening on\s+"
44 r"AF=(?P<family>\d+)\s+"
45 r"(?P<address>.+):(?P<port>\d+)$", re.I)
47 #: Used to recognize point at which socat(1) is sending data over the wire
48 TRANSFER_LOOP_RE = re.compile(r"^starting data transfer loop with FDs\s+.*$",
53 SOCAT_LOG_NOTICE = "N"
54 SOCAT_LOG_WARNING = "W"
58 SOCAT_LOG_IGNORE = frozenset([
64 #: Used to parse GNU dd(1) statistics
65 DD_INFO_RE = re.compile(r"^(?P<bytes>\d+)\s*byte(?:|s)\s.*\scopied,\s*"
66 r"(?P<seconds>[\d.]+)\s*s(?:|econds),.*$", re.I)
68 #: Used to ignore "N+N records in/out" on dd(1)'s stderr
69 DD_STDERR_IGNORE = re.compile(r"^\d+\+\d+\s*records\s+(?:in|out)$", re.I)
71 #: Signal upon which dd(1) will print statistics (on some platforms, SIGINFO is
72 #: unavailable and SIGUSR1 is used instead)
73 DD_INFO_SIGNAL = getattr(signal, "SIGINFO", signal.SIGUSR1)
75 #: Buffer size: at most this many bytes are transferred at once
78 # Common options for socat
79 SOCAT_TCP_OPTS = ["keepalive", "keepidle=60", "keepintvl=10", "keepcnt=5"]
80 SOCAT_OPENSSL_OPTS = ["verify=1", "cipher=HIGH", "method=TLSv1"]
85 PROG_DD_PID) = range(1, 5)
86 PROG_ALL = frozenset([
94 class CommandBuilder(object):
95 def __init__(self, mode, opts, socat_stderr_fd, dd_stderr_fd, dd_pid_fd):
96 """Initializes this class.
98 @param mode: Daemon mode (import or export)
99 @param opts: Options object
100 @type socat_stderr_fd: int
101 @param socat_stderr_fd: File descriptor socat should write its stderr to
102 @type dd_stderr_fd: int
103 @param dd_stderr_fd: File descriptor dd should write its stderr to
105 @param dd_pid_fd: File descriptor the child should write dd's PID to
110 self._socat_stderr_fd = socat_stderr_fd
111 self._dd_stderr_fd = dd_stderr_fd
112 self._dd_pid_fd = dd_pid_fd
115 def GetBashCommand(cmd):
116 """Prepares a command to be run in Bash.
119 return ["bash", "-o", "errexit", "-o", "pipefail", "-c", cmd]
121 def _GetSocatCommand(self):
122 """Returns the socat command.
125 common_addr_opts = SOCAT_TCP_OPTS + SOCAT_OPENSSL_OPTS + [
126 "key=%s" % self._opts.key,
127 "cert=%s" % self._opts.cert,
128 "cafile=%s" % self._opts.ca,
131 if self._opts.bind is not None:
132 common_addr_opts.append("bind=%s" % self._opts.bind)
134 if self._mode == constants.IEM_IMPORT:
135 if self._opts.port is None:
138 port = self._opts.port
141 "OPENSSL-LISTEN:%s" % port,
144 # Retry to listen if connection wasn't established successfully, up to
145 # 100 times a second. Note that this still leaves room for DoS attacks.
151 elif self._mode == constants.IEM_EXPORT:
154 "OPENSSL:%s:%s" % (self._opts.host, self._opts.port),
156 # How long to wait per connection attempt
157 "connect-timeout=%s" % self._opts.connect_timeout,
159 # Retry a few times before giving up to connect (once per second)
160 "retry=%s" % self._opts.connect_retries,
165 raise errors.GenericError("Invalid mode '%s'" % self._mode)
167 for i in [addr1, addr2]:
170 raise errors.GenericError("Comma not allowed in socat option"
171 " value: %r" % value)
174 constants.SOCAT_PATH,
185 # Unidirectional mode, the first address is only used for reading, and the
186 # second address is only used for writing
189 ",".join(addr1), ",".join(addr2)
192 def _GetTransportCommand(self):
193 """Returns the command for the transport part of the daemon.
196 socat_cmd = ("%s 2>&%d" %
197 (utils.ShellQuoteArgs(self._GetSocatCommand()),
198 self._socat_stderr_fd))
201 # Setting LC_ALL since we want to parse the output and explicitely
202 # redirecting stdin, as the background process (dd) would have /dev/null as
204 dd_cmd.write("{ LC_ALL=C dd bs=%s <&0 2>&%d & pid=${!};" %
205 (BUFSIZE, self._dd_stderr_fd))
207 dd_cmd.write(" echo $pid >&%d;" % self._dd_pid_fd)
209 dd_cmd.write(" wait $pid;")
212 compr = self._opts.compress
214 assert compr in constants.IEC_ALL
216 if self._mode == constants.IEM_IMPORT:
217 if compr == constants.IEC_GZIP:
218 transport_cmd = "%s | gunzip -c" % socat_cmd
220 transport_cmd = socat_cmd
222 transport_cmd += " | %s" % dd_cmd.getvalue()
223 elif self._mode == constants.IEM_EXPORT:
224 if compr == constants.IEC_GZIP:
225 transport_cmd = "gzip -c | %s" % socat_cmd
227 transport_cmd = socat_cmd
229 transport_cmd = "%s | %s" % (dd_cmd.getvalue(), transport_cmd)
231 raise errors.GenericError("Invalid mode '%s'" % self._mode)
233 # TODO: Run transport as separate user
234 # The transport uses its own shell to simplify running it as a separate user
236 return self.GetBashCommand(transport_cmd)
238 def GetCommand(self):
239 """Returns the complete child process command.
242 transport_cmd = self._GetTransportCommand()
246 if self._opts.cmd_prefix:
247 buf.write(self._opts.cmd_prefix)
250 buf.write(utils.ShellQuoteArgs(transport_cmd))
252 if self._opts.cmd_suffix:
254 buf.write(self._opts.cmd_suffix)
256 return self.GetBashCommand(buf.getvalue())
259 def _VerifyListening(family, address, port):
260 """Verify address given as listening address by socat.
263 # TODO: Implement IPv6 support
264 if family != socket.AF_INET:
265 raise errors.GenericError("Address family %r not supported" % family)
268 packed_address = socket.inet_pton(family, address)
270 raise errors.GenericError("Invalid address %r for family %s" %
273 return (socket.inet_ntop(family, packed_address), port)
276 class ChildIOProcessor(object):
277 def __init__(self, debug, status_file, logger, throughput_samples):
278 """Initializes this class.
282 self._status_file = status_file
283 self._logger = logger
285 self._splitter = dict([(prog, utils.LineSplitter(self._ProcessOutput, prog))
286 for prog in PROG_ALL])
289 self._dd_ready = False
290 self._dd_tp_samples = throughput_samples
291 self._dd_progress = []
293 # Expected size of transferred data
294 self._exp_size = None
296 def GetLineSplitter(self, prog):
297 """Returns the line splitter for a program.
300 return self._splitter[prog]
303 """Flushes all line splitters.
306 for ls in self._splitter.itervalues():
310 """Closes all line splitters.
313 for ls in self._splitter.itervalues():
315 self._splitter.clear()
318 """Tells dd(1) to write statistics.
321 if self._dd_pid is None:
325 if not self._dd_ready:
326 # There's a race condition between starting the program and sending
327 # signals. The signal handler is only registered after some time, so we
328 # have to check whether the program is ready. If it isn't, sending a
329 # signal will invoke the default handler (and usually abort the program).
330 if not utils.IsProcessHandlingSignal(self._dd_pid, DD_INFO_SIGNAL):
331 logging.debug("dd is not yet ready for signal %s", DD_INFO_SIGNAL)
334 logging.debug("dd is now handling signal %s", DD_INFO_SIGNAL)
335 self._dd_ready = True
337 logging.debug("Sending signal %s to PID %s", DD_INFO_SIGNAL, self._dd_pid)
339 os.kill(self._dd_pid, DD_INFO_SIGNAL)
340 except EnvironmentError, err:
341 if err.errno != errno.ESRCH:
344 # Process no longer exists
349 def _ProcessOutput(self, line, prog):
350 """Takes care of child process output.
353 @param line: Child output line
355 @param prog: Program from which the line originates
361 if prog == PROG_SOCAT:
363 parts = line.split(None, 4)
366 (_, _, _, level, msg) = parts
368 force_update = self._ProcessSocatOutput(self._status_file, level, msg)
370 if self._debug or (level and level not in SOCAT_LOG_IGNORE):
371 forward_line = "socat: %s %s" % (level, msg)
375 forward_line = "socat: %s" % line
377 elif prog == PROG_DD:
378 (should_forward, force_update) = self._ProcessDdOutput(line)
380 if should_forward or self._debug:
381 forward_line = "dd: %s" % line
385 elif prog == PROG_DD_PID:
387 raise RuntimeError("dd PID reported more than once")
388 logging.debug("Received dd PID %r", line)
389 self._dd_pid = int(line)
393 self._logger.info(forward_line)
394 self._status_file.AddRecentOutput(forward_line)
396 self._status_file.Update(force_update)
399 def _ProcessSocatOutput(status_file, level, msg):
400 """Interprets socat log output.
403 if level == SOCAT_LOG_NOTICE:
404 if status_file.GetListenPort() is None:
405 # TODO: Maybe implement timeout to not listen forever
406 m = LISTENING_RE.match(msg)
408 (_, port) = _VerifyListening(int(m.group("family")),
410 int(m.group("port")))
412 status_file.SetListenPort(port)
415 if not status_file.GetConnected():
416 m = TRANSFER_LOOP_RE.match(msg)
418 logging.debug("Connection established")
419 status_file.SetConnected()
424 def _ProcessDdOutput(self, line):
425 """Interprets a line of dd(1)'s output.
428 m = DD_INFO_RE.match(line)
430 seconds = float(m.group("seconds"))
431 mbytes = utils.BytesToMebibyte(int(m.group("bytes")))
432 self._UpdateDdProgress(seconds, mbytes)
435 m = DD_STDERR_IGNORE.match(line)
438 return (False, False)
443 def _UpdateDdProgress(self, seconds, mbytes):
444 """Updates the internal status variables for dd(1) progress.
447 @param seconds: Timestamp of this update
449 @param mbytes: Total number of MiB transferred so far
453 self._dd_progress.append((seconds, mbytes))
456 del self._dd_progress[:-self._dd_tp_samples]
458 # Calculate throughput
459 throughput = _CalcThroughput(self._dd_progress)
461 # Calculate percent and ETA
465 if self._exp_size is not None:
466 if self._exp_size != 0:
467 percent = max(0, min(100, (100.0 * mbytes) / self._exp_size))
470 eta = max(0, float(self._exp_size - mbytes) / throughput)
472 self._status_file.SetProgress(mbytes, throughput, percent, eta)
475 def _CalcThroughput(samples):
476 """Calculates the throughput in MiB/second.
478 @type samples: sequence
479 @param samples: List of samples, each consisting of a (timestamp, mbytes)
481 @rtype: float or None
482 @return: Throughput in MiB/second
486 # Can't calculate throughput
489 (start_time, start_mbytes) = samples[0]
490 (end_time, end_mbytes) = samples[-1]
492 return (float(end_mbytes) - start_mbytes) / (float(end_time) - start_time)