diff --git a/CHANGELOG.md b/CHANGELOG.md index 6fe4cff0..5a79f24d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,14 @@ Sections with "(Upcoming)" describe changes on the roadmap for CIJOE. Changes on the `master` branch, from the latest version tag up to and including HEAD can be subject to a git rebase. +## 0.9.61 (Upcoming) + +The SSH transport keeps one connection for its run()/get()/put() instead of +connecting and disconnecting around each, which cost a key exchange, about 0.4 s, +per command. The connection has a 30 second keepalive, is made again +on next use when it has dropped, e.g. as the remote rebooted, and is closed by +the new `Cijoe.close()`, at the end of a pytest session, or as Python exits. + ## 0.9.59 Renamed the "workflow" concept to "task" across the codebase, docs, and example diff --git a/docs/source/configs/index.rst b/docs/source/configs/index.rst index 89de5389..88ec3788 100644 --- a/docs/source/configs/index.rst +++ b/docs/source/configs/index.rst @@ -76,6 +76,11 @@ password login or key-based login. You can configure it as follows: key_filename = "path/to/private_key" passphrase = "baz" +The connection is made on first use and kept for the commands and transfers +that follow, with a keepalive every 30 seconds. When it has dropped, for example +because the remote rebooted, it is made again on next use. ``cijoe.close()`` +closes it; otherwise it is closed as Python exits. + Configuring multiple transports ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/src/cijoe/core/command.py b/src/cijoe/core/command.py index 6778ba6f..c6b2610b 100644 --- a/src/cijoe/core/command.py +++ b/src/cijoe/core/command.py @@ -263,6 +263,19 @@ def get(self, src, dst, transport_name=None): return False + def close(self, transport_name=None): + """ + Close the connections the transports keep, or only the one of the named + transport; a later call connects again + """ + + if transport_name is not None: + self._get_transport(transport_name).close() + return + + for tport in self.transports.values(): + tport.close() + def getconf(self, key: str, default: Any = None): """ Return value for given key, and return default if no value is found. diff --git a/src/cijoe/core/scripts/wait_for_transport.py b/src/cijoe/core/scripts/wait_for_transport.py index 8df115eb..c3a6250b 100644 --- a/src/cijoe/core/scripts/wait_for_transport.py +++ b/src/cijoe/core/scripts/wait_for_transport.py @@ -130,7 +130,10 @@ def main(args, cijoe): continue # TCP is open - verify with the trivial command to distinguish usable vs. unusable SSH. + # The transport keeps its connection between commands; closing it first makes the probe + # a fresh connect, which is what tells a host that went down or came up. try: + cijoe.close(transport_name) err, _ = cijoe.run("hostname", transport_name=transport_name) ssh_up = err == 0 except Exception as exc: diff --git a/src/cijoe/core/transport.py b/src/cijoe/core/transport.py index 6d365390..2c037563 100644 --- a/src/cijoe/core/transport.py +++ b/src/cijoe/core/transport.py @@ -3,6 +3,7 @@ import os import shutil import subprocess +import weakref from abc import ABC, abstractmethod from pathlib import Path @@ -25,6 +26,9 @@ def get(self, src, dst=None): def put(self, src, dst=None): pass + def close(self): + """Release what the transport holds open; it may be used again afterwards""" + class Local(Transport): """Provide cmd/push/pull locally""" @@ -83,6 +87,10 @@ def get(self, src, dst=None): return self.put(src, dst) +# Keeps the NAT and firewall state of an idle kept connection alive +KEEPALIVE_SECONDS = 30 + + class SSH(Transport): """Provide cmd/push/pull over SSH""" @@ -109,14 +117,11 @@ def __init__(self, config, output_path, transport_name): self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.scp = None - # Not connecting at this point, since the remote side might be ready at the time - # the transport is initialized. For example, when a qemu-guest is booted, then - # cijoe.run_local() is used, and not until the guest is up will the cij.run() be - # used. Also, when a system reboots etc. then dropped connections must be - # handled, lastly connection close must happen as Python terminates, dangling - # connections will make it hang. - # Thus, __connect()/__disconnect() is called for each call to run()/get()/put(). - # Current short-coming is of course that then these cannot happen in parallel. + # Not connecting here, the remote might not be up yet, e.g. a qemu guest that + # run_local() boots. The connection is made on first use and kept, rather than + # paying a key exchange per command. One that dropped, e.g. on a reboot of the + # remote, is made again on next use. Calls on one transport cannot overlap. + weakref.finalize(self, self.ssh.close) log.getLogger("paramiko.transport").setLevel(log.CRITICAL) paramiko.util.log_to_file( @@ -124,11 +129,30 @@ def __init__(self, config, output_path, transport_name): ) def __connect(self): - self.ssh.connect(**self.ssh_params) - self.scp = SCPClient(self.ssh.get_transport()) + """Connect, unless the connection made before is still up""" - def __disconnect(self): - self.scp.close() + transport = self.ssh.get_transport() + if transport is not None and transport.is_active(): + return + + self.close() + try: + self.ssh.connect(**self.ssh_params) + transport = self.ssh.get_transport() + transport.set_keepalive(KEEPALIVE_SECONDS) + self.scp = SCPClient(transport) + except BaseException: + # paramiko starts the transport before it authenticates, so a failed + # connect can leave one active, which the next use must not take + self.close() + raise + + def close(self): + """Close the connection; the next run()/get()/put() connects again""" + + if self.scp is not None: + self.scp.close() + self.scp = None self.ssh.close() def run(self, cmd, cwd, env, cmd_output): @@ -173,9 +197,9 @@ def run(self, cmd, cwd, env, cmd_output): cmd_output.flush() err = stdout.channel.recv_exit_status() - - self.__disconnect() + stdout.channel.close() except paramiko.ssh_exception.SSHException as exc: + self.close() err = ( exc.errno if hasattr(exc, "errno") @@ -201,9 +225,11 @@ def put(self, src, dst=None): if not os.path.isabs(src): src = os.path.join(self.output_path, self.output_ident, src) - self.scp.put(src, dst, recursive=True) - - self.__disconnect() + try: + self.scp.put(src, dst, recursive=True) + except Exception: + self.close() + raise return True @@ -217,8 +243,10 @@ def get(self, src, dst=None): if not os.path.isabs(src): dst = os.path.join(self.output_path, self.output_ident, dst) - self.scp.get(src, dst, recursive=True) - - self.__disconnect() + try: + self.scp.get(src, dst, recursive=True) + except Exception: + self.close() + raise return True diff --git a/src/cijoe/pytest_plugin/hooks_and_fixtures.py b/src/cijoe/pytest_plugin/hooks_and_fixtures.py index bcb43866..7c8ca409 100644 --- a/src/cijoe/pytest_plugin/hooks_and_fixtures.py +++ b/src/cijoe/pytest_plugin/hooks_and_fixtures.py @@ -81,6 +81,13 @@ def pytest_configure(config): raise Exception("Failed instantiating cijoe") +def pytest_unconfigure(config): + """Closes the connections of the cijoe instance made by ``pytest_configure``""" + + if pytest.cijoe_instance is not None: + pytest.cijoe_instance.close() + + def pytest_terminal_summary(terminalreporter, exitstatus, config): """ Prints out a notice that cijoe is beeing used along with the values of the diff --git a/tests/core/test_transport_ssh.py b/tests/core/test_transport_ssh.py new file mode 100644 index 00000000..01b8df24 --- /dev/null +++ b/tests/core/test_transport_ssh.py @@ -0,0 +1,274 @@ +""" +The SSH transport over a fake paramiko client, thus without a remote host: one +connection kept for many commands and transfers, made again after it dropped, +and closed by Cijoe.close() +""" + +import io +from pathlib import Path + +import pytest +from scp import SCPException + +from cijoe.core import transport +from cijoe.core.command import Cijoe +from cijoe.core.resources import Config + + +class FakeChannel: + def __init__(self): + self.closed = False + + def set_combine_stderr(self, combine): + pass + + def exit_status_ready(self): + return True + + def recv_exit_status(self): + return 0 + + def close(self): + self.closed = True + + +class FakeStdout: + def __init__(self, output): + self.channel = FakeChannel() + self.output = io.BytesIO(output) + + def read(self, size=-1): + return self.output.read(size) + + +class FakeTransport: + def __init__(self): + self.active = True + self.keepalive = 0 + + def is_active(self): + return self.active + + def set_keepalive(self, interval): + self.keepalive = interval + + +class FakeSSHClient: + """The parts of paramiko.SSHClient that the transport uses, counting connects""" + + def __init__(self): + self.transport = None + self.connects = 0 + self.commands = [] + self.channels = [] + self.fail_auth = 0 + self.fail_exec = 0 + + def set_missing_host_key_policy(self, policy): + pass + + def connect(self, **params): + self.connects += 1 + # As paramiko does: the transport is up before authentication fails + self.transport = FakeTransport() + if self.fail_auth: + self.fail_auth -= 1 + raise transport.paramiko.AuthenticationException("fake auth failure") + + def get_transport(self): + return self.transport + + def close(self): + self.transport = None + + def exec_command(self, cmd, environment=None): + assert self.transport and self.transport.active, "command without connection" + if self.fail_exec: + self.fail_exec -= 1 + raise transport.paramiko.SSHException("fake channel failure") + self.commands.append(cmd) + stdout = FakeStdout(b"output\n") + self.channels.append(stdout.channel) + return None, stdout, None + + +class FakeSCPClient: + fail_transfer = 0 + + def __init__(self, ssh_transport): + self.ssh_transport = ssh_transport + self.transfers = [] + + def put(self, src, dst, recursive=False): + assert self.ssh_transport.active, "transfer without connection" + if FakeSCPClient.fail_transfer: + FakeSCPClient.fail_transfer -= 1 + raise SCPException("fake transfer failure") + self.transfers.append(("put", src, dst)) + + def get(self, src, dst, recursive=False): + assert self.ssh_transport.active, "transfer without connection" + self.transfers.append(("get", src, dst)) + + def close(self): + pass + + +@pytest.fixture +def remote_cijoe(tmp_path, monkeypatch): + """A Cijoe instance with one SSH transport over FakeSSHClient""" + + monkeypatch.setattr(transport.paramiko, "SSHClient", FakeSSHClient) + monkeypatch.setattr(transport, "SCPClient", FakeSCPClient) + + config = Config(Path("fake.toml")) + config.options = { + "cijoe": { + "transport": { + "remote": {"hostname": "remote"}, + "other": {"hostname": "other"}, + } + } + } + + instance = Cijoe(config, tmp_path, False) + instance.set_output_ident("test") + yield instance + instance.close() + + +def test_one_connection(remote_cijoe): + """Commands and transfers share the connection made by the first""" + + commands = ["true", "id -u", "hostname"] + for cmd in commands: + err, _ = remote_cijoe.run(cmd) + assert err == 0 + + assert remote_cijoe.put("/tmp/src", "dst") + assert remote_cijoe.get("src", "/tmp/dst") + + ssh = remote_cijoe.transports["remote"].ssh + assert ssh.connects == 1 + assert ssh.commands == commands + assert ssh.transport.keepalive == transport.KEEPALIVE_SECONDS + assert all(channel.closed for channel in ssh.channels) + assert [kind for kind, _, _ in remote_cijoe.transports["remote"].scp.transfers] == [ + "put", + "get", + ] + + +def test_reconnect_after_drop(remote_cijoe): + """A connection that dropped is made again on next use, for commands and transfers""" + + ssh = remote_cijoe.transports["remote"].ssh + + err, _ = remote_cijoe.run("true") + assert err == 0 and ssh.connects == 1 + + ssh.transport.active = False + err, _ = remote_cijoe.run("true") + assert err == 0 and ssh.connects == 2 + + ssh.transport.active = False + assert remote_cijoe.put("/tmp/src", "dst") + assert ssh.connects == 3 + + +def test_closed(remote_cijoe): + """Cijoe.close() closes the connection, and a later command connects again""" + + ssh = remote_cijoe.transports["remote"].ssh + + err, _ = remote_cijoe.run("true") + assert err == 0 and ssh.transport is not None + + remote_cijoe.close() + assert ssh.transport is None + assert remote_cijoe.transports["remote"].scp is None + + err, _ = remote_cijoe.run("true") + assert err == 0 and ssh.connects == 2 + + +def test_wait_for_transport_probes_a_fresh_connection(remote_cijoe, monkeypatch): + """ + wait_for_transport's probe connects afresh, rather than riding the kept + connection, which may outlive the remote accepting new ones + """ + import argparse + + from cijoe.core.scripts import wait_for_transport + + monkeypatch.setattr(wait_for_transport.time, "sleep", lambda secs: None) + monkeypatch.setattr( + wait_for_transport.socket, "create_connection", lambda *a, **k: io.BytesIO() + ) + + err, _ = remote_cijoe.run("true", transport_name="remote") + assert not err + ssh = remote_cijoe.transports["remote"].ssh + assert ssh.connects == 1 + + args = argparse.Namespace(transport_name="remote", state="up", timeout=10) + assert wait_for_transport.main(args, remote_cijoe) == 0 + assert ssh.connects == 2, "the probe reused the kept connection" + + +def test_failed_connect_is_not_kept(remote_cijoe): + """A connect that fails after the transport started is not kept for next use""" + + ssh = remote_cijoe.transports["remote"].ssh + + ssh.fail_auth = 1 + assert not remote_cijoe.put("/tmp/src", "dst") + assert ssh.transport is None + + assert remote_cijoe.put("/tmp/src", "dst") + assert ssh.connects == 2 + + +def test_command_failure_drops_the_connection(remote_cijoe): + """An SSH error in a command closes the connection, so the next connects again""" + + ssh = remote_cijoe.transports["remote"].ssh + + err, _ = remote_cijoe.run("true") + assert err == 0 and ssh.connects == 1 + + ssh.fail_exec = 1 + err, _ = remote_cijoe.run("true") + assert err != 0 + + err, _ = remote_cijoe.run("true") + assert err == 0 and ssh.connects == 2 + + +def test_transfer_failure_drops_the_connection(remote_cijoe, monkeypatch): + """A failed transfer closes the connection, so the next transfer connects again""" + + ssh = remote_cijoe.transports["remote"].ssh + + assert remote_cijoe.put("/tmp/src", "dst") + assert ssh.connects == 1 + + monkeypatch.setattr(FakeSCPClient, "fail_transfer", 1) + assert not remote_cijoe.put("/tmp/src", "dst") + assert ssh.transport is None + + assert remote_cijoe.put("/tmp/src", "dst") + assert ssh.connects == 2 + + +def test_close_one_transport(remote_cijoe): + """Cijoe.close(name) closes only the named transport's connection""" + + err, _ = remote_cijoe.run("true", transport_name="remote") + assert not err + err, _ = remote_cijoe.run("true", transport_name="other") + assert not err + + remote_cijoe.close("remote") + assert remote_cijoe.transports["remote"].ssh.transport is None + assert remote_cijoe.transports["other"].ssh.transport is not None