SSH is connected to the remote server from the local server to open the command window and enter in the order of 1) helm repo update → 2) helm list as a Python script.

The output of 1) is normally returned, but the output of 2) is returned as None.
I tried connecting ssh to the remote server through moboxterm in my local pc, and then entered 1) 2), it is confirmed that they are output normally. (not None)

The SSH connection is normal, but only the command of 2) does not seem to be performed properly.

When 2) is performed, an error log called “Error: Kubernetes cluster unreachable” appears.

I have confirmed that the ssh connection is normal, and it is normal to perform the problematic command manually. There is a problem only when running with python script.

Here is code and log below.

1. lib/ssh.py

class SSH:

def __init__(self, cfg: dict, verbose=False) -> None:
    self._host = cfg.get("host")
    self._port = cfg.get("port")
    self._username = cfg.get("username")
    self._password = cfg.get("password")
    if not all([self._host, self._port, self._username, self._password]):
        raise ValueError(
            f"URL must include a value for the host[{self._host}],port[{self._port}],username[{self._username}],password[{self._password}] to use. None is not allowed!!"
        )
    self._client = None
    info(f"init done [cfg={cfg}]" if verbose else "init done")

### public
def connect(self):
    if self._client is None:
        self._client = paramiko.SSHClient()
        self._client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self._client.connect(
            hostname=self._host,
            port=self._port,
            username=self._username,
            password=self._password,
        )
        _transport = self._client.get_transport()
        _transport.set_keepalive(60)
        info("connected!")


def disconnect(self):
    self._client = self._client.close() if not self._client is None else None
    info("disconnected!")


def command(self, cmd: str, wait_for=None, timeout=30):
    try:
        self._auto_connect()
        _, _out, _err = self._client.exec_command(cmd)
        start_time = time.time()  # 1717034168
        while True:
            _temp_out = deepcopy(_out.read().decode())
            _temp_err = deepcopy(_err.read().decode())
            if not wait_for:
                break
            if wait_for in _temp_out or wait_for in _temp_err:
                raise AssertionError(f"can not found in output: {wait_for}")
            if time.time() - start_time > timeout:
                break
        return _temp_out  # if len(_err) == 0 else f"[err] {_err}"
    except Exception as e:
        error(f"command error as {e}")
        return None

2. lib/helm.py

from lib.ssh import SSH

class SSH_HELM:


def __init__(self, cfg: dict, verbose=True):
    self._cfg = {"host": cfg.get("host")}
    self._cfg["port"] = cfg.get("port", "22")
    self._cfg["username"] = cfg.get("username", "user")
    self._cfg["password"] = cfg.get("password", "1234")
    if not all(self._cfg.values()):
        raise ValueError(
            f"URL must include a value for the cfg={self._cfg} to use. None is not allowed!!"
        )
    debug(f"cfg error : {str(self._cfg)}")
    self._ssh = SSH(self._cfg)
    debug(f"cfg error22 :{str(self._ssh)}")
    info(f"init done [cfg={self._cfg}]" if verbose else "init done")



############# have some troubles at here #############
def command(self, cmd: str = None):
    # 1st
    self._ssh.connect()
    debug("connect 1st time - helm repo update")
    rr = self._ssh.command("/usr/sbin/helm repo update --debug | tee output1.json | cat")
    debug("connect 1st time - before disconnect")
    self._ssh.disconnect()
    debug(f"ssh helm information1 : {rr}")

    # 2nd
    self._ssh.connect()
    debug("connect 2nd time - helm repo update")        
    self._ssh.command("/usr/sbin/helm list -n default -o json > output.json 2> error.log")
    r = self._ssh.command("/usr/sbin/helm list -n default -o json --debug | tee output2.json | cat")  # connection ok. command nok.
    debug("connect 2nd time - before disconnect")
    self._ssh.disconnect()
    debug(f"ssh helm information2 : {r}")
    return json.loads(r)

3. remote server

~$ cat output.json
~$ cat output1.json
Hang tight while we grab the latest from your chart repositories...
...Successfully got an update from the "starlingx" chart repository
...Successfully got an update from the "stx-platform" chart repository
...Successfully got an update from the "main-ran" chart repository
...Successfully got an update from the "rel-ran" chart repository
Update Complete. ⎈ Happy Helming!⎈
~$ cat output2.json
~$ cat error.log
Error: Kubernetes cluster unreachable
~$