ssh
transport remote hpc · maintainer machinable-org · upstream
Run an execution on a remote machine over SSH.
Install
Declare the remote in your project's provider; the module is fetched on first use:
python
# interface/project.py — your project's provider
from machinable import Project
class MyProject(Project):
def on_resolve_remotes(self):
return {
"ssh": "url+https://raw.githubusercontent.com/machinable-org/machinable/6c6cb57c17808ac093ec2f954d3b790d5a5ae502/integrations/ssh/ssh.py",
}The module is fetched once and cached in interface/remotes/; machinable fetch ssh downloads it without importing so you can inspect the code first. See Resolving remotes for how pulling, pinning, and updating work.
ssh
Runs an execution on another machine while API server, the index, etc. stay local (see Transport).
bash
machinable get ssh slurm my_interface --launchConfiguration
python
from machinable import get
with get("ssh", {
"host": "gpu.lan", # anything ssh takes, including a config alias
"directory": "/home/me/project", # the project, on the far side
"storage": "/scratch/me/storage", # the storage root, on the far side
}):
get("my_interface").launch()Monitoring
Nothing runs on the far side, so watching a run is a poll:
python
status = transport.wait(interface, timeout=3600) # markers only, cheaply
transport.sync(interface) # the whole record, oncewait returns when the run finishes or when its heartbeat goes stale.
The poll fetches the *_at markers, the run's output.log, and the scheduler's job.out so following a job's log does not cost much. Artifacts come back on an explicit sync.
What it needs on the far side
Each command opens a connection and closes it again, with a short ControlPersist window so a burst shares one connection without holding it open. The far side needs the scheduler, and an interpreter for the job itself (e.g. Slurm's node_local + venv_tar)
Configuration
| Option | Type | Default |
|---|---|---|
host | str | required |
storage | str | None | None |
directory | str | None | None |
options | list[str] | ['-o', 'BatchMode=yes', '-o', 'ControlMaster=auto', '-o', 'ControlPersist=60'] |
rsync_options | list[str] | ['-a'] |
Source
ssh.py
py
"""SSH transport: run an execution on another machine."""
import os
import posixpath
import shlex
from pydantic import BaseModel, ConfigDict
from machinable import Project, Storage, Transport
class Ssh(Transport):
"""Reaches another machine over SSH."""
class Config(BaseModel):
model_config = ConfigDict(extra="forbid")
host: str
"""``[user@]hostname`` as ssh would take it (including an ssh_config alias)."""
storage: str | None = None
"""Storage root on the far side; mirrors the local one record for record."""
directory: str | None = None
"""Project directory on the far side."""
options: list[str] = [
"-o",
"BatchMode=yes",
"-o",
"ControlMaster=auto",
"-o",
"ControlPersist=60",
]
"""Passed to ssh."""
rsync_options: list[str] = ["-a"]
"""Passed to rsync in both directions."""
# -- addressing ---------------------------------------------------------
def roots(self) -> list[tuple[str, str]]:
"""``(local, remote)`` directory pairs, longest local prefix first."""
pairs = []
if self.config.storage:
pairs.append((os.path.abspath(Storage.get().root()), self.config.storage))
if self.config.directory:
pairs.append((os.path.abspath(Project.get().path()), self.config.directory))
return sorted(pairs, key=lambda pair: -len(pair[0]))
def path(self, local: str) -> str:
"""This machine's path as the far side sees it."""
local = os.path.abspath(local)
for base, remote in self.roots():
if local == base:
return remote
if local.startswith(base + os.sep):
relative = os.path.relpath(local, base).replace(os.sep, "/")
return posixpath.join(remote, relative)
raise ValueError(
f"'{local}' is not under any directory {self} maps to the far side "
f"({', '.join(base for base, _ in self.roots()) or 'none configured'}); "
"set the transport's `storage` and `directory`."
)
def ssh_command(self) -> list[str]:
"""The ssh invocation prefix, without a remote command."""
return ["ssh", *self.config.options, self.config.host]
def run(self, cmd, **kwargs):
"""Run ``cmd`` on the far side."""
remote = " ".join(shlex.quote(str(argument)) for argument in cmd)
return super().run([*self.ssh_command(), remote], **kwargs)
def push(self, local: str, remote: str | None = None, include=None) -> str:
"""Mirror a local directory to the far side."""
remote = remote or self.path(local)
self.run(["mkdir", "-p", remote], check=True)
self._rsync(
_as_directory(local),
f"{self.config.host}:{_as_directory(remote)}",
# a running job's markers are newer than the copies here; a
# re-push (a resubmission, say) must not roll them back
options=["--update", *_filter(include)],
)
return remote
def pull(self, remote: str, local: str, include=None) -> bool:
"""Mirror a directory back from the far side."""
os.makedirs(local, exist_ok=True)
result = self._rsync(
f"{self.config.host}:{_as_directory(remote)}",
_as_directory(local),
options=_filter(include),
check=False,
)
return result.returncode == 0
def _rsync(self, source: str, destination: str, *, options=None, check=True):
transport = " ".join(shlex.quote(part) for part in self.ssh_command()[:-1])
return super().run(
[
"rsync",
*self.config.rsync_options,
*(options or []),
"-e",
transport,
source,
destination,
],
check=check,
)
def _filter(include) -> list[str]:
"""rsync arguments selecting only ``include``, or everything when empty."""
if not include:
return []
options = []
for pattern in include:
options += ["--include", pattern]
# recurse everywhere, take only the named files
return ["--include", "*/", *options, "--exclude", "*"]
def _as_directory(path: str) -> str:
"""Trailing slash: rsync copies a directory's contents, not the directory."""
return path if path.endswith("/") else path + "/"