feat: create init container for job

cmd/tools/commands/release.go

@@ -33,7 +33,7 @@ func NewReleaseCmd() *cobra.Command {
 			pushVersionTag(nextAppVersion)
 
 			// Let's checkout helm chart repo and put changes to particular app
-			dir, err := git.PartialCheckout("https://github.com/kubeshop/helm-charts.git", appName, "main")
+			dir, err := git.PartialCheckout("https://github.com/kubeshop/helm-charts.git", appName, "main", "")
 			ui.ExitOnError("checking out "+appName+" chart to "+dir, err)
 
 			chart, path, err := helm.GetChart(dir)
@@ -52,7 +52,7 @@ func NewReleaseCmd() *cobra.Command {
 			gitAddCommitAndPush(dir, "updating "+appName+" chart version to "+nextAppVersion)
 
 			// Checkout main testkube chart and bump main chart with next version
-			dir, err = git.PartialCheckout("https://github.com/kubeshop/helm-charts.git", "testkube", "main")
+			dir, err = git.PartialCheckout("https://github.com/kubeshop/helm-charts.git", "testkube", "main", "")
 			ui.ExitOnError("checking out testkube chart to "+dir, err)
 
 			chart, path, err = helm.GetChart(dir)

internal/app/api/v1/server.go

@@ -43,27 +43,27 @@ func NewServer(
 	var httpConfig server.Config
 	envconfig.Process("APISERVER", &httpConfig)
 
-	executor, err := client.NewJobExecutor(executionsResults)
-	if err != nil {
-		panic(err)
-	}
-
 	s := TestkubeAPI{
 		HTTPServer:           server.NewServer(httpConfig),
 		TestExecutionResults: testExecutionsResults,
 		ExecutionResults:     executionsResults,
-		Executor:             executor,
 		TestsClient:          testsClient,
 		ExecutorsClient:      executorsClient,
 		SecretClient:         secretClient,
 		TestsSuitesClient:    testsuitesClient,
 		Metrics:              NewMetrics(),
 	}
 
-	if err = s.loadDefaultExecutors(os.Getenv("TESTKUBE_NAMESPACE"), os.Getenv("TESTKUBE_DEFAULT_EXECUTORS")); err != nil {
+	initImage, err := s.loadDefaultExecutors(os.Getenv("TESTKUBE_NAMESPACE"), os.Getenv("TESTKUBE_DEFAULT_EXECUTORS"))
+	if err != nil {
 		s.Log.Warnf("load default executors %w", err)
 	}
 
+	s.Executor, err = client.NewJobExecutor(executionsResults, initImage)
+	if err != nil {
+		panic(err)
+	}
+
 	s.Init()
 	return s
 }
@@ -239,27 +239,32 @@ func getFilterFromRequest(c *fiber.Ctx) result.Filter {
 }
 
 // loadDefaultExecutors loads default executors
-func (s TestkubeAPI) loadDefaultExecutors(namespace, data string) error {
+func (s TestkubeAPI) loadDefaultExecutors(namespace, data string) (initImage string, err error) {
 	var executors []testkube.ExecutorDetails
 
 	if data == "" {
-		return nil
+		return "", nil
 	}
 
 	dataDecoded, err := base64.StdEncoding.DecodeString(data)
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	if err := json.Unmarshal([]byte(dataDecoded), &executors); err != nil {
-		return err
+		return "", err
 	}
 
 	for _, executor := range executors {
 		if executor.Executor == nil {
 			continue
 		}
 
+		if executor.Name == "executor-init" {
+			initImage = executor.Executor.Image
+			continue
+		}
+
 		obj := &executorv1.Executor{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      executor.Name,
@@ -274,20 +279,20 @@ func (s TestkubeAPI) loadDefaultExecutors(namespace, data string) error {
 
 		result, err := s.ExecutorsClient.Get(namespace, executor.Name)
 		if err != nil && !errors.IsNotFound(err) {
-			return err
+			return "", err
 		}
 
 		if err != nil {
 			if _, err = s.ExecutorsClient.Create(obj); err != nil {
-				return err
+				return "", err
 			}
 		} else {
 			result.Spec = obj.Spec
 			if _, err = s.ExecutorsClient.Update(result); err != nil {
-				return err
+				return "", err
 			}
 		}
 	}
 
-	return nil
+	return initImage, nil
 }

pkg/executor/client/job.go

@@ -13,8 +13,8 @@ import (
 	"go.uber.org/zap"
 )
 
-func NewJobExecutor(repo result.Repository) (client JobExecutor, err error) {
-	jobClient, err := jobs.NewJobClient()
+func NewJobExecutor(repo result.Repository, initImage string) (client JobExecutor, err error) {
+	jobClient, err := jobs.NewJobClient(initImage)
 	if err != nil {
 		return client, fmt.Errorf("can't get k8s jobs client: %w", err)
 	}

pkg/executor/content/fetcher.go

@@ -6,18 +6,22 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/url"
+	"os"
 	"path/filepath"
 	"strings"
 
 	"github.com/kubeshop/testkube/pkg/api/v1/testkube"
 	"github.com/kubeshop/testkube/pkg/git"
 )
 
-func NewFetcher() Fetcher {
-	return Fetcher{}
+func NewFetcher(dir string) Fetcher {
+	return Fetcher{
+		dir: dir,
+	}
 }
 
 type Fetcher struct {
+	dir string
 }
 
 func (f Fetcher) Fetch(content *testkube.TestContent) (path string, err error) {
@@ -63,10 +67,10 @@ func (f Fetcher) FetchGitDir(repo *testkube.Repository) (path string, err error)
 
 	// if path not set make full repo checkout
 	if repo.Path == "" {
-		return git.Checkout(uri, repo.Branch)
+		return git.Checkout(uri, repo.Branch, f.dir)
 	}
 
-	return git.PartialCheckout(uri, repo.Path, repo.Branch)
+	return git.PartialCheckout(uri, repo.Path, repo.Branch, f.dir)
 }
 
 // FetchGitFile returns path to git based file saved in local temp directory
@@ -76,7 +80,7 @@ func (f Fetcher) FetchGitFile(repo *testkube.Repository) (path string, err error
 		return path, err
 	}
 
-	repoPath, err := git.Checkout(uri, repo.Branch)
+	repoPath, err := git.Checkout(uri, repo.Branch, f.dir)
 	if err != nil {
 		return path, err
 	}
@@ -101,7 +105,13 @@ func (f Fetcher) gitURI(repo *testkube.Repository) (uri string, err error) {
 }
 
 func (f Fetcher) saveTempFile(reader io.Reader) (path string, err error) {
-	tmpFile, err := ioutil.TempFile("", "fetcher-save-temp-file")
+	var tmpFile *os.File
+	filename := "fetcher-save-temp-file"
+	if f.dir == "" {
+		tmpFile, err = ioutil.TempFile("", filename)
+	} else {
+		tmpFile, err = os.Create(filepath.Join(f.dir, filename))
+	}
 	if err != nil {
 		return "", err
 	}

pkg/executor/content/fetcher_test.go

@@ -19,7 +19,7 @@ const fileContent = `{"some":"json","file":"with content"}
 `
 
 func TestFetcher(t *testing.T) {
-	f := NewFetcher()
+	f := NewFetcher("")
 
 	t.Run("test fetch uri", func(t *testing.T) {
 

pkg/git/checkout.go

@@ -10,11 +10,13 @@ import (
 )
 
 // Checkout will checkout directory from Git repository
-func Checkout(uri, branch string) (outputDir string, err error) {
-
-	tmpDir, err := ioutil.TempDir("", "git-checkout")
-	if err != nil {
-		return tmpDir, err
+func Checkout(uri, branch, dir string) (outputDir string, err error) {
+	tmpDir := dir
+	if tmpDir == "" {
+		tmpDir, err = ioutil.TempDir("", "git-checkout")
+		if err != nil {
+			return "", err
+		}
 	}
 
 	_, err = process.ExecuteInDir(
@@ -32,12 +34,14 @@ func Checkout(uri, branch string) (outputDir string, err error) {
 	return tmpDir + "/repo/", nil
 }
 
-// Partial checkout will checkout only given directory from Git repository
-func PartialCheckout(uri, path, branch string) (outputDir string, err error) {
-
-	tmpDir, err := ioutil.TempDir("", "git-sparse-checkout")
-	if err != nil {
-		return tmpDir, err
+// PartialCheckout will checkout only given directory from Git repository
+func PartialCheckout(uri, path, branch, dir string) (outputDir string, err error) {
+	tmpDir := dir
+	if tmpDir == "" {
+		tmpDir, err = ioutil.TempDir("", "git-sparse-checkout")
+		if err != nil {
+			return "", err
+		}
 	}
 
 	_, err = process.ExecuteInDir(

pkg/git/checkout_test.go

@@ -11,7 +11,7 @@ func TestCheckOut(t *testing.T) {
 
 	repo := "https://github.com/cirosantilli/test-git-partial-clone-big-small"
 
-	dir, err := PartialCheckout(repo, "small", "main")
+	dir, err := PartialCheckout(repo, "small", "main", "")
 	t.Logf("partial repo checkedout to dir: %s", dir)
 	assert.NoError(t, err)
 	t.Fail()

pkg/jobs/jobclient.go

@@ -38,6 +38,9 @@ const (
 
 	pollTimeout  = 24 * time.Hour
 	pollInterval = 200 * time.Millisecond
+
+	volumeName = "data-volume"
+	volumeDir  = "/data"
 )
 
 type JobClient struct {
@@ -46,9 +49,10 @@ type JobClient struct {
 	Namespace  string
 	Cmd        string
 	Log        *zap.SugaredLogger
+	initImage  string
 }
 
-func NewJobClient() (*JobClient, error) {
+func NewJobClient(initImage string) (*JobClient, error) {
 	clientSet, err := k8sclient.ConnectToK8s()
 	if err != nil {
 		return nil, err
@@ -58,6 +62,7 @@ func NewJobClient() (*JobClient, error) {
 		ClientSet: clientSet,
 		Namespace: "testkube",
 		Log:       log.DefaultLogger,
+		initImage: initImage,
 	}, nil
 }
 
@@ -77,7 +82,7 @@ func (c *JobClient) LaunchK8sJobSync(image string, repo result.Repository, execu
 		return result.Err(err), err
 	}
 
-	jobSpec := NewJobSpec(execution.Id, c.Namespace, image, string(jsn), execution.TestName, hasSecrets)
+	jobSpec := NewJobSpec(execution.Id, c.Namespace, image, string(jsn), execution.TestName, c.initImage, hasSecrets)
 
 	_, err = jobs.Create(ctx, jobSpec, metav1.CreateOptions{})
 	if err != nil {
@@ -151,7 +156,7 @@ func (c *JobClient) LaunchK8sJob(image string, repo result.Repository, execution
 		return result.Err(err), err
 	}
 
-	jobSpec := NewJobSpec(execution.Id, c.Namespace, image, string(jsn), execution.TestName, hasSecrets)
+	jobSpec := NewJobSpec(execution.Id, c.Namespace, image, string(jsn), execution.TestName, c.initImage, hasSecrets)
 
 	_, err = jobs.Create(ctx, jobSpec, metav1.CreateOptions{})
 	if err != nil {
@@ -434,7 +439,7 @@ func (c *JobClient) CreatePersistentVolumeClaim(name string) error {
 }
 
 // NewJobSpec is a method to create new job spec
-func NewJobSpec(id, namespace, image, jsn, testName string, hasSecrets bool) *batchv1.Job {
+func NewJobSpec(id, namespace, image, jsn, testName, executorInitImage string, hasSecrets bool) *batchv1.Job {
 	var TTLSecondsAfterFinished int32 = 180
 	// TODO backOff need to be handled correctly by Logs and by Running job spec - currently we can get unexpected results
 	var backOffLimit int32 = 0
@@ -467,6 +472,25 @@ func NewJobSpec(id, namespace, image, jsn, testName string, hasSecrets bool) *ba
 		}
 	}
 
+	var initContainers []corev1.Container
+	if executorInitImage != "" {
+		initContainers = []corev1.Container{
+			{
+				Name:            id + "-init",
+				Image:           executorInitImage,
+				Command:         []string{"/bin/runner", jsn},
+				ImagePullPolicy: corev1.PullAlways,
+				Env:             append(envVars, secretEnvVars...),
+				VolumeMounts: []corev1.VolumeMount{
+					{
+						Name:      volumeName,
+						MountPath: volumeDir,
+					},
+				},
+			},
+		}
+	}
+
 	return &batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      id,
@@ -476,13 +500,25 @@ func NewJobSpec(id, namespace, image, jsn, testName string, hasSecrets bool) *ba
 			TTLSecondsAfterFinished: &TTLSecondsAfterFinished,
 			Template: corev1.PodTemplateSpec{
 				Spec: corev1.PodSpec{
+					InitContainers: initContainers,
 					Containers: []corev1.Container{
 						{
 							Name:            id,
 							Image:           image,
 							Command:         []string{"/bin/runner", jsn},
 							ImagePullPolicy: corev1.PullAlways,
 							Env:             append(envVars, secretEnvVars...),
+							VolumeMounts: []corev1.VolumeMount{
+								{
+									Name:      volumeName,
+									MountPath: volumeDir,
+								},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: volumeName,
 						},
 					},
 					RestartPolicy: corev1.RestartPolicyNever,
@@ -526,6 +562,10 @@ var envVars = []corev1.EnvVar{
 		Name:  "RUNNER_SCRAPPERENABLED",
 		Value: os.Getenv("SCRAPPERENABLED"),
 	},
+	{
+		Name:  "RUNNER_DATADIR",
+		Value: volumeDir,
+	},
 }
 
 // IsPodReady defines if pod is ready or failed for logs scrapping