Add process logger to bundler

Makefile

@@ -22,6 +22,8 @@ GOFUMPT_VERSION=0.9.2
 JQ_VERSION=1.8.1
 KIND_VERSION=0.31.0
 
+export DHCTL_TESTS_OPENSSH_IMAGE = lscr.io/linuxserver/openssh-server:10.0_p1-r9-ls209
+
 PLATFORM_NAME := $(shell uname -m)
 
 OS_NAME := $(shell uname)
@@ -127,7 +129,10 @@ go-deps/ci/check/no-tidy: go-installed go-deps/tidy
 		exit 1; \
 	fi
 
-test: go-installed docker-installed bin/kind
+test/pull-ssh-image: docker-installed
+	@docker pull $(DHCTL_TESTS_OPENSSH_IMAGE)
+
+test: go-installed docker-installed bin/kind test/pull-ssh-image
 	./hack/run_tests.sh
 	$(MAKE) clean/test
 

pkg/kube/README.md

@@ -0,0 +1,5 @@
+# kube package
+
+Contains kube clients implementations.
+
+Kube clients e2e tests located in `../tests/e2e/kube/*_test.go`.

pkg/provider/README.md

@@ -2,7 +2,7 @@
 
 Contains kube and ssh provider.
 
-SSH provider integration tests located in `../ssh/testssh/provider_test.go`.
+SSH provider e2e tests located in `../tests/e2e/ssh/provider_test.go`.
 
 Kube provider does not contain unit tests only integration.
-Integration tests located in `../tests/provider/kube_test.go`.
+Integration tests located in `../tests/e2e/kube/provider_test.go`.

pkg/ssh.go

@@ -20,6 +20,7 @@ import (
 	"regexp"
 	"time"
 
+	"github.com/deckhouse/lib-dhctl/pkg/log"
 	"github.com/deckhouse/lib-dhctl/pkg/retry"
 
 	"github.com/deckhouse/lib-connection/pkg/ssh/session"
@@ -127,6 +128,7 @@ type BundlerOptions struct {
 	NoLogStepOutOnError  bool
 	StepsDelimiter       string
 	Retries              int
+	ProcessLogger        log.ProcessLogger
 }
 
 func (o *BundlerOptions) IsValid() error {
@@ -179,6 +181,12 @@ func BundlerWithRetries(retries int) BundlerOption {
 	}
 }
 
+func BundlerWithProcessLogger(logger log.ProcessLogger) BundlerOption {
+	return func(opts *BundlerOptions) {
+		opts.ProcessLogger = logger
+	}
+}
+
 type Bundler interface {
 	Execute(ctx context.Context, parentDir, bundleDir string) ([]byte, error)
 }

pkg/ssh/README.md

@@ -0,0 +1,5 @@
+# ssh package
+
+Contains ssh client implementations.
+
+SSH clients e2e tests located in `../tests/e2e/ssh/*/*_test.go`.

pkg/ssh/gossh/common_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/deckhouse/lib-connection/pkg/tests"
+	kind "github.com/deckhouse/lib-connection/pkg/tests/e2e/kube/kind"
 )
 
 func registerStopClient(t *testing.T, sshClient *Client) {
@@ -107,10 +108,10 @@ func registerStopTunnel(t *testing.T, tunnel *Tunnel) {
 func startContainerAndClientAndKind(t *testing.T, test *tests.Test, opts ...tests.TestContainerWrapperSettingsOpts) (*Client, *tests.TestContainerWrapper) {
 	sshClient, container := startContainerAndClientWithContainer(t, test, opts...)
 
-	kindCluster := tests.CreateKINDCluster(t, &tests.KINDClusterCreateParams{
+	kindCluster := kind.CreateKINDCluster(t, &kind.KINDClusterCreateParams{
 		Test:        test,
 		ClusterName: "kube-proxy",
-		Containers: []*tests.SSHContainersForKind{
+		Containers: []*kind.SSHContainersForKind{
 			{
 				Client:    sshClient,
 				Container: container,

pkg/ssh/gossh/kube_proxy_test.go

@@ -17,6 +17,7 @@ package gossh
 import (
 	"context"
 	"fmt"
+	"os"
 	"testing"
 	"time"
 
@@ -29,6 +30,10 @@ import (
 func TestKubeProxy(t *testing.T) {
 	test := tests.ShouldNewIntegrationTest(t, "TestKubeGoProxy")
 
+	if kindBinEnv := os.Getenv("TEST_KIND_BINARY"); kindBinEnv == "" {
+		t.Setenv("TEST_KIND_BINARY", "../../../bin/kind")
+	}
+
 	sshClient, container := prepareContainerForTestKubeProxy(t, test)
 
 	waitRestart := func(op string) {

pkg/ssh/local/script.go

@@ -147,6 +147,16 @@ func (s *Script) WithNoLogStepOutOnError(f bool) {
 	s.noOutError = f
 }
 
+func (s *Script) WithBundleDest(d string) *Script {
+	s.bundleDest = d
+	return s
+}
+
+func (s *Script) WithForceNoSudoForBundle(f bool) *Script {
+	s.forceBundleNoSudo = f
+	return s
+}
+
 func (s *Script) WithExecuteUploadDir(string) {}
 
 func (s *Script) bundleCmdProvider(ctx context.Context, node connection.Interface, parentDir, bundleDir string) (connection.Command, error) {

pkg/ssh/utils/bundle.go

@@ -90,6 +90,12 @@ func BundleWithCommandPreparator(p CommandPreparator) BundleOpt {
 	}
 }
 
+func BundleWithProcessLogger(logger log.ProcessLogger) BundleOpt {
+	return func(b *Bundle) {
+		b.processLogger = logger
+	}
+}
+
 func UserBundleOptsOrBashible(inputOpts ...connection.BundlerOption) ([]BundleOpt, error) {
 	userOpts := make([]connection.BundlerOption, len(inputOpts))
 	copy(userOpts, inputOpts)
@@ -117,6 +123,8 @@ type Bundle struct {
 	commandPreparator   CommandPreparator
 
 	bundleCmdProvider BundleCmdProvider
+
+	processLogger log.ProcessLogger
 }
 
 func NewBundle(sett settings.Settings, client connection.Interface, scriptPath string, args []string, opts ...BundleOpt) (*Bundle, error) {
@@ -161,7 +169,10 @@ func (b *Bundle) Execute(ctx context.Context, parentDir, bundleDir string) ([]by
 	bundleCmd.Sudo(ctx)
 
 	logger := b.sett.Logger()
-	processLogger := logger.ProcessLogger()
+	processLogger := b.processLogger
+	if govalue.Nil(processLogger) {
+		processLogger = logger.ProcessLogger()
+	}
 
 	handler := newOutputHandler(b, bundleCmd, logger, processLogger)
 
@@ -413,5 +424,6 @@ func convertBundleOption(opts ...connection.BundlerOption) ([]BundleOpt, error)
 		BundleWithStepsDelimiter(options.StepsDelimiter),
 		BundleWithNoLogStepOutOnError(options.NoLogStepOutOnError),
 		BundleWithShouldInfoOutChecker(options.ShouldInfoOutChecker),
+		BundleWithProcessLogger(options.ProcessLogger),
 	}, nil
 }

pkg/tests/kube/kube_exec_test.go → pkg/tests/e2e/kube/exec_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package provider
+package kube_test
 
 import (
 	"bytes"
@@ -32,6 +32,7 @@ import (
 	"github.com/deckhouse/lib-connection/pkg/kube"
 	"github.com/deckhouse/lib-connection/pkg/provider"
 	"github.com/deckhouse/lib-connection/pkg/tests"
+	kind "github.com/deckhouse/lib-connection/pkg/tests/e2e/kube/kind"
 )
 
 func TestKubeExec(t *testing.T) {
@@ -41,15 +42,15 @@ func TestKubeExec(t *testing.T) {
 		tests.TestWithParallelRun(false),
 	)
 
-	kindCluster := tests.CreateKINDCluster(t, &tests.KINDClusterCreateParams{
+	kindCluster := kind.CreateKINDCluster(t, &kind.KINDClusterCreateParams{
 		Test:                                execTest,
 		ClusterName:                         "exec-in-pod",
 		NoPrepareLocalKubectlInSSHContainer: true,
 	})
 
 	kindCluster.RegisterCleanup(t)
 
-	kubeProvider := getKubeProvider(t, execTest, kindCluster)
+	kubeProvider := getKubeProviderForExec(t, execTest, kindCluster)
 
 	pythonImage := "registry.deckhouse.io/base_images@sha256:b15f9150f3b51f2e11cd73db39c89eef8a075953659caf802363d4f544335fb5"
 	if envImage := os.Getenv("TEST_KUBE_EXEC_PYTHON_IMAGE"); envImage != "" {
@@ -161,7 +162,7 @@ print(data, end="")
 	})
 }
 
-func getKubeProvider(t *testing.T, test *tests.Test, cluster *tests.KINDCluster) *provider.DefaultKubeProvider {
+func getKubeProviderForExec(t *testing.T, test *tests.Test, cluster *kind.KINDCluster) *provider.DefaultKubeProvider {
 	sett := test.Settings()
 
 	restCfg, err := cluster.RESTConfig()

pkg/tests/kind.go → pkg/tests/e2e/kube/kind/cluster.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"os"
 	"os/exec"
 	"regexp"
 	"strings"
@@ -30,10 +31,10 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 
 	connection "github.com/deckhouse/lib-connection/pkg"
+	"github.com/deckhouse/lib-connection/pkg/tests"
 )
 
 const (
-	KindBinary = "../../../bin/kind"
 	kindConfig = `
 kind: Cluster
 apiVersion: kind.x-k8s.io/v1alpha4
@@ -48,11 +49,11 @@ var (
 
 type SSHContainersForKind struct {
 	Client    connection.SSHClient
-	Container *TestContainerWrapper
+	Container *tests.TestContainerWrapper
 }
 
 type KINDClusterCreateParams struct {
-	Test        *Test
+	Test        *tests.Test
 	ClusterName string
 	Containers  []*SSHContainersForKind
 
@@ -64,7 +65,7 @@ type KINDCluster struct {
 	ControlPlaneIP   string
 	ControlPlanePort string
 
-	test       *Test
+	test       *tests.Test
 	kubeconfig string
 	restConfig *rest.Config
 }
@@ -74,7 +75,7 @@ func (c *KINDCluster) appendClusterNameArg(args []string) []string {
 }
 
 func (c *KINDCluster) runKind(args ...string) (string, error) {
-	cmd := exec.Command(KindBinary, c.appendClusterNameArg(args)...)
+	cmd := exec.Command(getKINDBinary(), c.appendClusterNameArg(args)...)
 	out, err := cmd.CombinedOutput()
 	return string(out), err
 }
@@ -238,7 +239,7 @@ func CreateKINDCluster(t *testing.T, params *KINDClusterCreateParams) *KINDClust
 	test.GetLogger().InfoF("Creating KIND cluster %s...", clusterName)
 
 	out, err := cluster.runKind(args...)
-	require.NoError(t, err, "not create kind cluster: %w:%s\n", out)
+	require.NoError(t, err, "not create kind cluster: %w:%s\n", err, out)
 
 	test.GetLogger().InfoF("KIND cluster %s created:\n%s", clusterName, out)
 
@@ -292,7 +293,7 @@ func runDockerForKINDContainer(cluster *KINDCluster, name string, args ...string
 
 	err := retry.NewLoopWithParams(params).Run(func() error {
 		var err error
-		out, err = RunDockerWithOut(args...)
+		out, err = tests.RunDockerWithOut(args...)
 		out = strings.TrimSpace(out)
 		return err
 	})
@@ -462,3 +463,11 @@ func (p *localKubectlPreparator) prepareLocalKubeCtlInSSHContainer(t *testing.T,
 	)
 	checkErrorDuringCreateCluster(t, cluster, err, "failed to create link to kube config on ssh container %s", containerName)
 }
+
+func getKINDBinary() string {
+	if bin := os.Getenv("TEST_KIND_BINARY"); bin != "" {
+		return bin
+	}
+
+	return "../../../../bin/kind"
+}