feat(commander): run node commands over SSH so pkg/e2e accepts the cluster

The previous commit returned a Cluster with a nil NodeExecutor, and pkg/e2e
rejects that outright:

  provider "commander" returned a cluster without a NodeExecutor

That check is deliberate and covered by
TestConnectWithProviderValidatesCluster/missing_node_executor, so the answer is
to supply an executor rather than to loosen the contract. The NodeExecutor doc
comment already describes the shape this should take: "the node InternalIP
behind the master for Commander".

commanderNodeExecutor resolves the node name to its InternalIP through the
Kubernetes API and SSHes to it over the connector's existing route - the bastion
when one is configured - as the same user the master is reached with. No
infrastructure access beyond what Commander already gives us, and no new SSH
plumbing: buildSSHClient was already parameterised by host and user.

It opens a client per Exec rather than holding one per node: node commands are
occasional in these suites, and a long-lived connection per node touched would
keep sessions open for the whole run.

Exit codes follow the interface contract - a command that ran and exited non-zero
is reported through ExecResult.ExitCode with a nil error; only transport failures
are errors.

Disks stays nil, which pkg/e2e already handles by substituting a stub that
explains itself: Commander hands out a cluster, not the machines under it.

Tested: InternalIP selection (preferred over ExternalIP/Hostname), the
no-InternalIP and unknown-node errors, and that a resolve failure surfaces before
any SSH is attempted. The SSH path itself needs a live cluster and is exercised by
the consuming suite.

Signed-off-by: v.oleynikov <vasily.oleynikov@flant.com>

Author: duckhawk

internal/provisioning/commander/connect_test_cluster.go

@@ -20,6 +20,8 @@ import (
 	"context"
 	"fmt"
 
+	"k8s.io/client-go/kubernetes"
+
 	"github.com/deckhouse/storage-e2e/pkg/clusterprovider"
 )
 
@@ -31,24 +33,49 @@ var _ clusterprovider.Provider = (*commanderProvider)(nil)
 //
 // The connection itself is the one the legacy pkg/cluster path already used
 // through the Connector interface - SSH to the master via the bastion, kubeconfig
-// fetched off the master, in-process API tunnel - so this delegates to Connect
-// rather than duplicating it.
+// fetched off the master, in-process API tunnel - so this reuses the connector
+// rather than duplicating any of it.
 //
-// Nodes and Disks are nil: Commander hands out a cluster, not the infrastructure
-// under it, so there is no node-exec transport and no way to attach block
-// devices. Suites that need either must run on a provider that offers them (dvp);
-// suites that only talk to the Kubernetes API - the common case for a storage
-// module whose backend is external - are fully served here.
+// Disks stays nil: Commander hands out a cluster, not the infrastructure under
+// it, so there is no way to attach block devices. Cluster documents Disks as
+// nillable and pkg/e2e substitutes a stub that says so when a suite tries.
 func (p *commanderProvider) ConnectTestCluster(ctx context.Context) (*clusterprovider.Cluster, error) {
-	restConfig, cleanup, err := p.Connect(ctx)
+	// Detach cancellation: the tunnel must outlive the caller's connect ctx (the
+	// suite keeps the connection for its whole run); Cleanup tears it down.
+	ctx = context.WithoutCancel(ctx)
+
+	creds, err := p.conf.Resolve()
+	if err != nil {
+		return nil, fmt.Errorf("resolve commander credentials: %w", err)
+	}
+	conn := newConnector(p.client, p.conf, creds, p.logger)
+
+	// The node executor SSHes to nodes as the same user the master is reached
+	// with, over the same hops.
+	_, sshUser, err := conn.resolveMaster(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	restConfig, cleanup, err := conn.Connect(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("connect to the commander cluster: %w", err)
 	}
 
+	clientset, err := kubernetes.NewForConfig(restConfig)
+	if err != nil {
+		cleanup()
+		return nil, fmt.Errorf("build clientset for node address lookups: %w", err)
+	}
+
 	return &clusterprovider.Cluster{
 		RESTConfig: restConfig,
-		Nodes:      nil,
-		Disks:      nil,
-		Cleanup:    cleanup,
+		Nodes: &commanderNodeExecutor{
+			conn:     conn,
+			resolver: &internalIPResolver{clientset: clientset},
+			user:     sshUser,
+		},
+		Disks:   nil,
+		Cleanup: cleanup,
 	}, nil
 }

internal/provisioning/commander/node_executor.go

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2026 Flant JSC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * 	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package commander
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	cryptossh "golang.org/x/crypto/ssh"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/deckhouse/storage-e2e/pkg/clusterprovider"
+)
+
+// nodeAddressResolver maps a Kubernetes node name to an address reachable over
+// SSH. Commander does not own the infrastructure, so the only address it can
+// offer is the one the node reports itself.
+type nodeAddressResolver interface {
+	Resolve(ctx context.Context, nodeName string) (string, error)
+}
+
+// internalIPResolver reads the node's InternalIP from the Kubernetes API.
+type internalIPResolver struct {
+	clientset kubernetes.Interface
+}
+
+func (r *internalIPResolver) Resolve(ctx context.Context, nodeName string) (string, error) {
+	node, err := r.clientset.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
+	if err != nil {
+		return "", fmt.Errorf("get node %s: %w", nodeName, err)
+	}
+	for _, addr := range node.Status.Addresses {
+		if addr.Type == corev1.NodeInternalIP && addr.Address != "" {
+			return addr.Address, nil
+		}
+	}
+	return "", fmt.Errorf("node %s reports no InternalIP", nodeName)
+}
+
+// commanderNodeExecutor runs commands on test cluster nodes over SSH. The node
+// name is resolved to its InternalIP through the Kubernetes API, and the
+// connection reuses the connector's route - the bastion when one is configured,
+// then the node itself - so it needs no infrastructure access Commander does not
+// already give us.
+//
+// A fresh client per Exec: node commands are occasional in these suites, and
+// holding one connection per node for the whole run would keep sessions open
+// against every node a scenario ever touched.
+type commanderNodeExecutor struct {
+	conn     *connector
+	resolver nodeAddressResolver
+	user     string
+}
+
+var _ clusterprovider.NodeExecutor = (*commanderNodeExecutor)(nil)
+
+func (e *commanderNodeExecutor) Exec(ctx context.Context, nodeName, command string) (clusterprovider.ExecResult, error) {
+	ip, err := e.resolver.Resolve(ctx, nodeName)
+	if err != nil {
+		return clusterprovider.ExecResult{}, err
+	}
+
+	client, err := e.conn.buildSSHClient(ctx, ip, e.user)
+	if err != nil {
+		return clusterprovider.ExecResult{}, fmt.Errorf("connect to node %s (%s): %w", nodeName, ip, err)
+	}
+	defer func() { _ = client.Close() }()
+
+	res, err := client.Exec(ctx, command)
+	out := clusterprovider.ExecResult{
+		Stdout:   res.Stdout,
+		Stderr:   res.Stderr,
+		ExitCode: res.ExitCode,
+	}
+	// Per the NodeExecutor contract a command that ran and exited non-zero is
+	// not an error: the exit code carries that. Only transport failures are.
+	var exitErr *cryptossh.ExitError
+	if errors.As(err, &exitErr) {
+		return out, nil
+	}
+	if err != nil {
+		return out, fmt.Errorf("exec on node %s (%s): %w", nodeName, ip, err)
+	}
+	return out, nil
+}

internal/provisioning/commander/node_executor_test.go

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2026 Flant JSC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * 	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package commander
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func nodeWithAddresses(name string, addrs ...corev1.NodeAddress) *corev1.Node {
+	return &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: name},
+		Status:     corev1.NodeStatus{Addresses: addrs},
+	}
+}
+
+func TestInternalIPResolver_PicksInternalIP(t *testing.T) {
+	node := nodeWithAddresses("worker-1",
+		corev1.NodeAddress{Type: corev1.NodeHostName, Address: "worker-1"},
+		corev1.NodeAddress{Type: corev1.NodeExternalIP, Address: "203.0.113.7"},
+		corev1.NodeAddress{Type: corev1.NodeInternalIP, Address: "10.12.4.151"},
+	)
+	r := &internalIPResolver{clientset: fake.NewSimpleClientset(node)}
+
+	got, err := r.Resolve(context.Background(), "worker-1")
+	if err != nil {
+		t.Fatalf("Resolve returned unexpected error: %v", err)
+	}
+	if want := "10.12.4.151"; got != want {
+		t.Errorf("Resolve() = %q, want %q", got, want)
+	}
+}
+
+func TestInternalIPResolver_ErrorsWithoutInternalIP(t *testing.T) {
+	node := nodeWithAddresses("worker-1",
+		corev1.NodeAddress{Type: corev1.NodeExternalIP, Address: "203.0.113.7"},
+	)
+	r := &internalIPResolver{clientset: fake.NewSimpleClientset(node)}
+
+	if _, err := r.Resolve(context.Background(), "worker-1"); err == nil {
+		t.Fatal("expected an error for a node without an InternalIP, got nil")
+	} else if !strings.Contains(err.Error(), "no InternalIP") {
+		t.Errorf("error = %v, want it to mention the missing InternalIP", err)
+	}
+}
+
+func TestInternalIPResolver_ErrorsOnMissingNode(t *testing.T) {
+	r := &internalIPResolver{clientset: fake.NewSimpleClientset()}
+
+	if _, err := r.Resolve(context.Background(), "absent"); err == nil {
+		t.Fatal("expected an error for an unknown node, got nil")
+	}
+}
+
+// resolverFunc lets a test stub address resolution.
+type resolverFunc func(ctx context.Context, nodeName string) (string, error)
+
+func (f resolverFunc) Resolve(ctx context.Context, nodeName string) (string, error) {
+	return f(ctx, nodeName)
+}
+
+// A resolution failure must surface before any SSH is attempted, so the executor
+// is usable (and its errors legible) without a reachable node.
+func TestCommanderNodeExecutor_SurfacesResolveError(t *testing.T) {
+	e := &commanderNodeExecutor{
+		resolver: resolverFunc(func(context.Context, string) (string, error) {
+			return "", context.DeadlineExceeded
+		}),
+	}
+
+	if _, err := e.Exec(context.Background(), "worker-1", "true"); err == nil {
+		t.Fatal("expected the resolve error to surface, got nil")
+	}
+}