"""Connect to node prior to running exec_command or scp.
If there already is a connection to the node, this method reuses it.
+
+ :param node: Node in topology.
+ :param attempts: Number of reconnect attempts.
+ :type node: dict
+ :type attempts: int
+ :raises IOError: If cannot connect to host.
"""
- try:
- self._node = node
- node_hash = self._node_hash(node)
- if node_hash in SSH.__existing_connections:
- self._ssh = SSH.__existing_connections[node_hash]
- logger.debug('reusing ssh: {0}'.format(self._ssh))
+ self._node = node
+ node_hash = self._node_hash(node)
+ if node_hash in SSH.__existing_connections:
+ self._ssh = SSH.__existing_connections[node_hash]
+ if self._ssh.get_transport().is_active():
+ logger.debug('Reusing SSH: {ssh}'.format(ssh=self._ssh))
else:
+ if attempts > 0:
+ self._reconnect(attempts-1)
+ else:
+ raise IOError('Cannot connect to {host}'.
+ format(host=node['host']))
+ else:
+ try:
start = time()
pkey = None
if 'priv_key' in node:
self._ssh.get_transport().set_keepalive(10)
SSH.__existing_connections[node_hash] = self._ssh
-
- logger.trace('connect took {} seconds'.format(time() - start))
- logger.debug('new ssh: {0}'.format(self._ssh))
-
- logger.debug('Connect peer: {0}'.
- format(self._ssh.get_transport().getpeername()))
- logger.debug('Connections: {0}'.
- format(str(SSH.__existing_connections)))
- except RuntimeError as exc:
- if attempts > 0:
- self._reconnect(attempts-1)
- else:
- raise exc
+ logger.debug('New SSH to {peer} took {total} seconds: {ssh}'.
+ format(
+ peer=self._ssh.get_transport().getpeername(),
+ total=(time() - start),
+ ssh=self._ssh))
+ except SSHException:
+ raise IOError('Cannot connect to {host}'.
+ format(host=node['host']))
def disconnect(self, node):
"""Close SSH connection to the node.
"""
node_hash = self._node_hash(node)
if node_hash in SSH.__existing_connections:
- logger.debug('Disconnecting peer: {}, {}'.
- format(node['host'], node['port']))
+ logger.debug('Disconnecting peer: {host}, {port}'.
+ format(host=node['host'], port=node['port']))
ssh = SSH.__existing_connections.pop(node_hash)
ssh.close()
def _reconnect(self, attempts=0):
- """Close the SSH connection and open it again."""
+ """Close the SSH connection and open it again.
+ :param attempts: Number of reconnect attempts.
+ :type attempts: int
+ """
node = self._node
self.disconnect(node)
self.connect(node, attempts)
- logger.debug('Reconnecting peer done: {}'.
- format(self._ssh.get_transport().getpeername()))
+ logger.debug('Reconnecting peer done: {host}, {port}'.
+ format(host=node['host'], port=node['port']))
def exec_command(self, cmd, timeout=10):
"""Execute SSH command on a new channel on the connected Node.
def interactive_terminal_open(self, time_out=30):
"""Open interactive terminal on a new channel on the connected Node.
- FIXME: Convert or document other possible exceptions, such as
- socket.error or SSHException.
+ :param time_out: Timeout in seconds.
+ :returns: SSH channel with opened terminal.
.. warning:: Interruptingcow is used here, and it uses
signal(SIGALRM) to let the operating system interrupt program
handlers only apply to the main thread, so you cannot use this
from other threads. You must not use this in a program that
uses SIGALRM itself (this includes certain profilers)
-
- :param time_out: Timeout in seconds.
- :returns: SSH channel with opened terminal.
- :raise IOError: If receive attempt results in socket.timeout.
"""
chan = self._ssh.get_transport().open_session()
chan.get_pty()
break
except socket.timeout:
logger.error('Socket timeout: {0}'.format(buf))
- # TODO: Find out which exception would callers appreciate here.
- raise IOError('Socket timeout: {0}'.format(buf))
+ raise Exception('Socket timeout: {0}'.format(buf))
return chan
def interactive_terminal_exec_command(self, chan, cmd, prompt):
interactive_terminal_open() method has to be called first!
+ :param chan: SSH channel with opened terminal.
+ :param cmd: Command to be executed.
+ :param prompt: Command prompt, sequence of characters used to
+ indicate readiness to accept commands.
+ :returns: Command output.
+
.. warning:: Interruptingcow is used here, and it uses
signal(SIGALRM) to let the operating system interrupt program
execution. This has the following limitations: Python signal
handlers only apply to the main thread, so you cannot use this
from other threads. You must not use this in a program that
uses SIGALRM itself (this includes certain profilers)
-
- :param chan: SSH channel with opened terminal.
- :param cmd: Command to be executed.
- :param prompt: Command prompt, sequence of characters used to
- indicate readiness to accept commands.
- :returns: Command output.
- :raise IOError: If receive attempt results in socket.timeout.
"""
chan.sendall('{c}\n'.format(c=cmd))
buf = ''
except socket.timeout:
logger.error('Socket timeout during execution of command: '
'{0}\nBuffer content:\n{1}'.format(cmd, buf))
- # TODO: Find out which exception would callers appreciate here.
- raise IOError('Socket timeout during execution of command: '
- '{0}\nBuffer content:\n{1}'.format(cmd, buf))
+ raise Exception('Socket timeout during execution of command: '
+ '{0}\nBuffer content:\n{1}'.format(cmd, buf))
tmp = buf.replace(cmd.replace('\n', ''), '')
for item in prompt:
tmp.replace(item, '')
"""
chan.close()
- def scp(self, local_path, remote_path, get=False, timeout=10):
+ def scp(self, local_path, remote_path, get=False, timeout=30):
"""Copy files from local_path to remote_path or vice versa.
connect() method has to be called first!
def exec_cmd(node, cmd, timeout=600, sudo=False):
"""Convenience function to ssh/exec/return rc, out & err.
- FIXME: Document :param, :type, :raise and similar.
Returns (rc, stdout, stderr).
"""
if node is None:
Code Review / csit.git / blobdiff