Skip to content

sh

cluv sh: run a raw shell command across every currently-connected cluster via clush.

sh 

sh(command: list[str]) -> None

Runs command via clush on every cluster we currently have an active SSH connection to.

If invoked from a cluster (i.e. current_cluster() is set), also runs command locally first, since get_active_remotes() never includes the current cluster.

Does not sync the project or wrap the command in uv run (see cluv run for that). Does not attempt to connect to clusters we aren't already connected to, so this never triggers a 2FA prompt. If either run exits with a non-zero return code, this exits the process with the first non-zero code encountered (the local run's code takes precedence if both fail).

Parameters:

Name Type Description Default
command list[str]

The command (and its arguments) to run.

 required
Source code in cluv/cli/sh.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
async def sh(command: list[str]) -> None:
    """Runs `command` via `clush` on every cluster we currently have an active SSH connection to.

    If invoked from a cluster (i.e. `current_cluster()` is set), also runs `command` locally
    first, since `get_active_remotes()` never includes the current cluster.

    Does not sync the project or wrap the command in `uv run` (see `cluv run` for that). Does not
    attempt to connect to clusters we aren't already connected to, so this never triggers a 2FA
    prompt. If either run exits with a non-zero return code, this exits the process with the first
    non-zero code encountered (the local run's code takes precedence if both fail).

    Parameters:
        command: The command (and its arguments) to run.
    """
    disabled = get_disabled_clusters()
    print_disabled_clusters(disabled)

    remotes = await get_active_remotes()
    here = current_cluster()

    if not remotes and not here:
        console.print(
            "[yellow]Not currently connected to any cluster.[/yellow] "
            "Use `cluv login` to connect first."
        )
        return

    returncode = 0

    if here:
        returncode = await _run_locally(command)

    if remotes:
        hostnames = [remote.hostname for remote in remotes]
        with tempfile.NamedTemporaryFile("w", suffix=".txt") as hostfile:
            hostfile.write("\n".join(hostnames) + "\n")
            hostfile.flush()
            clush_returncode = await _invoke_clush(Path(hostfile.name), command)
        returncode = returncode or clush_returncode

    if returncode != 0:
        sys.exit(returncode)