Use sshiofs

Signed-off-by: Artiom Diomin <[email protected]>

pkg/etcdutil/etcd.go

@@ -20,6 +20,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
+	"io/fs"
 	"time"
 
 	"github.com/pkg/errors"
@@ -28,6 +29,7 @@ import (
 
 	"k8c.io/kubeone/pkg/apis/kubeone"
 	"k8c.io/kubeone/pkg/ssh"
+	"k8c.io/kubeone/pkg/ssh/sshiofs"
 	"k8c.io/kubeone/pkg/ssh/sshtunnel"
 	"k8c.io/kubeone/pkg/state"
 )
@@ -66,33 +68,34 @@ func NewClientConfig(s *state.State, host kubeone.HostConfig) (*clientv3.Config,
 // certificates and key are downloaded over SSH from the
 // /etc/kubernetes/pki/etcd/ directory.
 func LoadTLSConfig(conn ssh.Connection) (*tls.Config, error) {
+	sshfs := sshiofs.New(conn)
 	// Download CA
-	caCertPem, _, _, err := conn.Exec("sudo cat /etc/kubernetes/pki/etcd/ca.crt")
+	caCertPem, err := fs.ReadFile(sshfs, "/etc/kubernetes/pki/etcd/ca.crt")
 	if err != nil {
 		return nil, err
 	}
 
 	// Download cert
-	certPem, _, _, err := conn.Exec("sudo cat /etc/kubernetes/pki/etcd/server.crt")
+	certPem, err := fs.ReadFile(sshfs, "/etc/kubernetes/pki/etcd/server.crt")
 	if err != nil {
 		return nil, err
 	}
 
 	// Download key
-	keyPem, _, _, err := conn.Exec("sudo cat /etc/kubernetes/pki/etcd/server.key")
+	keyPem, err := fs.ReadFile(sshfs, "/etc/kubernetes/pki/etcd/server.key")
 	if err != nil {
 		return nil, err
 	}
 
 	// Add certificate and key to the TLS config
-	cert, err := tls.X509KeyPair([]byte(certPem), []byte(keyPem))
+	cert, err := tls.X509KeyPair(certPem, keyPem)
 	if err != nil {
 		return nil, err
 	}
 
 	// Add CA certificate to the TLS config
 	caCertPool := x509.NewCertPool()
-	caCertPool.AppendCertsFromPEM([]byte(caCertPem))
+	caCertPool.AppendCertsFromPEM(caCertPem)
 
 	return &tls.Config{
 		MinVersion:   tls.VersionTLS12,

pkg/kubeconfig/kubeconfig.go

@@ -17,12 +17,15 @@ limitations under the License.
 package kubeconfig
 
 import (
+	"io/fs"
+
 	"github.com/pkg/errors"
 
 	"k8s.io/client-go/tools/clientcmd"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"k8c.io/kubeone/pkg/ssh"
+	"k8c.io/kubeone/pkg/ssh/sshiofs"
 	"k8c.io/kubeone/pkg/state"
 )
 
@@ -43,12 +46,7 @@ func Download(s *state.State) ([]byte, error) {
 }
 
 func CatKubernetesAdminConf(conn ssh.Connection) ([]byte, error) {
-	konfig, _, _, err := conn.Exec("sudo cat /etc/kubernetes/admin.conf")
-	if err != nil {
-		return nil, err
-	}
-
-	return []byte(konfig), nil
+	return fs.ReadFile(sshiofs.New(conn), "/etc/kubernetes/admin.conf")
 }
 
 // BuildKubernetesClientset builds core kubernetes and apiextensions clientsets

pkg/tasks/certs.go

@@ -17,12 +17,10 @@ limitations under the License.
 package tasks
 
 import (
-	"bytes"
 	"crypto/x509"
 	"encoding/pem"
 	"errors"
-	"fmt"
-	"io"
+	"io/fs"
 	"time"
 
 	"github.com/Masterminds/semver/v3"
@@ -32,6 +30,7 @@ import (
 	"k8c.io/kubeone/pkg/clientutil"
 	"k8c.io/kubeone/pkg/kubeconfig"
 	"k8c.io/kubeone/pkg/ssh"
+	"k8c.io/kubeone/pkg/ssh/sshiofs"
 	"k8c.io/kubeone/pkg/state"
 )
 
@@ -66,7 +65,7 @@ func renewControlPlaneCerts(s *state.State) error {
 
 	err := s.RunTaskOnControlPlane(
 		func(s *state.State, node *kubeoneapi.HostConfig, conn ssh.Connection) error {
-			_, _, _, err := conn.Exec(renewCmd)
+			_, _, err := s.Runner.RunRaw(renewCmd)
 			return err
 		},
 		state.RunParallel,
@@ -78,18 +77,13 @@ func renewControlPlaneCerts(s *state.State) error {
 	return kubeconfig.BuildKubernetesClientset(s)
 }
 
-func fetchCert(conn ssh.Connection, filename string) (*x509.Certificate, error) {
-	var stdoutBuf bytes.Buffer
-
-	exitCode, err := conn.Stream(fmt.Sprintf("sudo cat %s", filename), &stdoutBuf, io.Discard)
+func fetchCert(sshfs fs.FS, filename string) (*x509.Certificate, error) {
+	buf, err := fs.ReadFile(sshfs, filename)
 	if err != nil {
 		return nil, err
 	}
-	if exitCode != 0 {
-		return nil, errors.New("non zero exit code")
-	}
 
-	pemBlock, rest := pem.Decode(stdoutBuf.Bytes())
+	pemBlock, rest := pem.Decode(buf)
 	if len(rest) != 0 {
 		return nil, errors.New("returned non-zero rest")
 	}
@@ -128,8 +122,9 @@ func earliestCertExpiry(conn ssh.Connection) (time.Time, error) {
 		}
 	)
 
+	sshfs := sshiofs.New(conn)
 	for _, certName := range certsToCheck {
-		cert, err := fetchCert(conn, certName)
+		cert, err := fetchCert(sshfs, certName)
 		if err != nil {
 			return earliestCertExpirationTime, err
 		}

pkg/tasks/encryption_providers.go

@@ -18,13 +18,15 @@ package tasks
 
 import (
 	"context"
-	"fmt"
+	"io/fs"
+	"path"
 
 	"github.com/pkg/errors"
 
 	kubeoneapi "k8c.io/kubeone/pkg/apis/kubeone"
 	"k8c.io/kubeone/pkg/scripts"
 	"k8c.io/kubeone/pkg/ssh"
+	"k8c.io/kubeone/pkg/ssh/sshiofs"
 	"k8c.io/kubeone/pkg/state"
 	"k8c.io/kubeone/pkg/templates"
 	encryptionproviders "k8c.io/kubeone/pkg/templates/encryptionproviders"
@@ -50,15 +52,20 @@ func fetchEncryptionProvidersFile(s *state.State) error {
 	if err != nil {
 		return err
 	}
+
+	sshfs := sshiofs.New(conn)
 	fileName := s.GetEncryptionProviderConfigName()
-	config, _, _, err := conn.Exec(fmt.Sprintf("sudo cat /etc/kubernetes/encryption-providers/%s", fileName))
+
+	config, err := fs.ReadFile(sshfs, path.Join("/etc/kubernetes/encryption-providers", fileName))
 	if err != nil {
 		return err
 	}
+
 	s.LiveCluster.Lock.Lock()
 	s.LiveCluster.EncryptionConfiguration.Config = &apiserverconfigv1.EncryptionConfiguration{}
-	err = kyaml.UnmarshalStrict([]byte(config), s.LiveCluster.EncryptionConfiguration.Config)
+	err = kyaml.UnmarshalStrict(config, s.LiveCluster.EncryptionConfiguration.Config)
 	s.LiveCluster.Lock.Unlock()
+
 	return err
 }
 

pkg/tasks/nodes.go

@@ -18,6 +18,7 @@ package tasks
 
 import (
 	"bytes"
+	"fmt"
 	"io"
 	"path/filepath"
 
@@ -187,30 +188,25 @@ func labelNodeOSes(s *state.State) error {
 	return nil
 }
 
-const (
-	getControllerManagerScript = `sudo dd if=/etc/kubernetes/manifests/kube-controller-manager.yaml`
-	putControllerManagerScript = `cat | sudo dd of=/etc/kubernetes/manifests/kube-controller-manager.yaml`
-)
-
 func patchStaticPods(s *state.State) error {
+	const controllerManagerManifestPath = `/etc/kubernetes/manifests/kube-controller-manager.yaml`
+
 	return s.RunTaskOnControlPlane(func(s *state.State, node *kubeoneapi.HostConfig, conn ssh.Connection) error {
 		s.Logger.Infoln("Patching static pods...")
 
-		popenConn, ok := conn.(interface {
-			POpen(cmd string, stdin io.Reader, stdout io.Writer, stderr io.Writer) (int, error)
-		})
-		if !ok {
-			return errors.New("ssh.Connection does not implement POpen()")
+		sshfs := s.Runner.NewFS()
+		file, err := sshfs.Open(controllerManagerManifestPath)
+		if err != nil {
+			return err
 		}
 
-		var stdout, stderr bytes.Buffer
-		_, err := popenConn.POpen(getControllerManagerScript, nil, &stdout, &stderr)
+		buf, err := io.ReadAll(file)
 		if err != nil {
-			return errors.Wrapf(err, "failed to get kube-controller-manager.yaml, stderr: %s", stderr.String())
+			return err
 		}
 
 		pod := corev1.Pod{}
-		if err = yaml.Unmarshal(stdout.Bytes(), &pod); err != nil {
+		if err = yaml.Unmarshal(buf, &pod); err != nil {
 			return errors.Wrap(err, "failed to unmarshal kube-controller-manager.yaml")
 		}
 
@@ -243,15 +239,22 @@ func patchStaticPods(s *state.State) error {
 			pod.Spec.Containers[0].Env = append(pod.Spec.Containers[0].Env, envVar)
 		}
 
-		buf, err := yaml.Marshal(&pod)
+		buf, err = yaml.Marshal(&pod)
 		if err != nil {
 			return errors.Wrap(err, "failed to marshal kube-controller-manager.yaml")
 		}
 
-		stdin := bytes.NewBuffer(buf)
-		stderr.Reset()
-		stdout.Reset()
-		_, err = popenConn.POpen(putControllerManagerScript, stdin, &stdout, &stderr)
-		return errors.Wrapf(err, "failed to write kube-controller-manager.yaml\nstdout: %s\nstderr: %s", stdout.String(), stderr.String())
+		wfile, ok := file.(io.WriteSeeker)
+		if !ok {
+			return fmt.Errorf("file handler %v doesn't implement writer", file)
+		}
+
+		_, err = wfile.Seek(0, io.SeekStart)
+		if err != nil {
+			return err
+		}
+
+		_, err = io.Copy(wfile, bytes.NewBuffer(buf))
+		return errors.Wrapf(err, "failed to write")
 	}, state.RunParallel)
 }

pkg/tasks/tasks.go

@@ -131,11 +131,12 @@ func WithResources(t Tasks) Tasks {
 		Tasks{
 			{
 				Fn: func(s *state.State) error {
-					return s.RunTaskOnControlPlane(func(ctx *state.State, node *kubeone.HostConfig, conn ssh.Connection) error {
+					return s.RunTaskOnControlPlane(func(ctx *state.State, node *kubeone.HostConfig, _ ssh.Connection) error {
 						cmd, err := scripts.SaveCABundle(ctx.WorkDir)
 						if err != nil {
 							return err
 						}
+
 						_, _, err = ctx.Runner.RunRaw(cmd)
 						return err
 					}, state.RunParallel)