feat: [TKC-2895] pvc config

pkg/testworkflows/testworkflowprocessor/bundle.go

@@ -27,6 +27,7 @@ type BundleOptions struct {
 type Bundle struct {
 	Secrets       []corev1.Secret
 	ConfigMaps    []corev1.ConfigMap
+	Pvcs          []corev1.PersistentVolumeClaim
 	Job           batchv1.Job
 	Signature     []stage.Signature
 	FullSignature []stage.Signature
@@ -50,6 +51,9 @@ func (b *Bundle) SetGroupId(groupId string) {
 	for i := range b.Secrets {
 		AnnotateGroupId(&b.Secrets[i], groupId)
 	}
+	for i := range b.Pvcs {
+		AnnotateGroupId(&b.Pvcs[i], groupId)
+	}
 }
 
 func (b *Bundle) Deploy(ctx context.Context, clientSet kubernetes.Interface, namespace string) (err error) {
@@ -68,6 +72,13 @@ func (b *Bundle) Deploy(ctx context.Context, clientSet kubernetes.Interface, nam
 			return errors.Wrap(err, "failed to deploy config maps")
 		}
 	}
+	for _, item := range b.Pvcs {
+		_, err = clientSet.CoreV1().PersistentVolumeClaims(namespace).Create(ctx, &item, metav1.CreateOptions{})
+		if err != nil {
+			return errors.Wrap(err, "failed to deploy pvcs")
+		}
+	}
+
 	_, err = clientSet.BatchV1().Jobs(namespace).Create(ctx, &b.Job, metav1.CreateOptions{})
 	return errors.Wrap(err, "failed to deploy job")
 }

pkg/testworkflows/testworkflowprocessor/intermediate.go

@@ -21,13 +21,15 @@ type Intermediate interface {
 	ConfigMaps() []corev1.ConfigMap
 	Secrets() []corev1.Secret
 	Volumes() []corev1.Volume
+	Pvcs() []corev1.PersistentVolumeClaim
 
 	AppendJobConfig(cfg *testworkflowsv1.JobConfig) Intermediate
 	AppendPodConfig(cfg *testworkflowsv1.PodConfig) Intermediate
 
 	AddConfigMap(configMap corev1.ConfigMap) Intermediate
 	AddSecret(secret corev1.Secret) Intermediate
 	AddVolume(volume corev1.Volume) Intermediate
+	AddPvc(pvc corev1.PersistentVolumeClaim) Intermediate
 
 	AddEmptyDirVolume(source *corev1.EmptyDirVolumeSource, mountPath string) corev1.VolumeMount
 
@@ -47,8 +49,9 @@ type intermediate struct {
 	Job testworkflowsv1.JobConfig `expr:"include"`
 
 	// Actual Kubernetes resources to use
-	Secs []corev1.Secret    `expr:"force"`
-	Cfgs []corev1.ConfigMap `expr:"force"`
+	Secs []corev1.Secret                `expr:"force"`
+	Cfgs []corev1.ConfigMap             `expr:"force"`
+	Ps   []corev1.PersistentVolumeClaim `expr:"force"`
 
 	// Storing files
 	Files ConfigMapFiles `expr:"include"`
@@ -87,6 +90,10 @@ func (s *intermediate) Volumes() []corev1.Volume {
 	return append(s.Pod.Volumes, s.Files.Volumes()...)
 }
 
+func (s *intermediate) Pvcs() []corev1.PersistentVolumeClaim {
+	return s.Ps
+}
+
 func (s *intermediate) AppendJobConfig(cfg *testworkflowsv1.JobConfig) Intermediate {
 	s.Job = *testworkflowresolver.MergeJobConfig(&s.Job, cfg)
 	return s
@@ -102,6 +109,11 @@ func (s *intermediate) AddVolume(volume corev1.Volume) Intermediate {
 	return s
 }
 
+func (s *intermediate) AddPvc(pvc corev1.PersistentVolumeClaim) Intermediate {
+	s.Ps = append(s.Ps, pvc)
+	return s
+}
+
 func (s *intermediate) AddConfigMap(configMap corev1.ConfigMap) Intermediate {
 	s.Cfgs = append(s.Cfgs, configMap)
 	return s

pkg/testworkflows/testworkflowprocessor/processor.go

@@ -162,6 +162,16 @@ func (p *processor) Bundle(ctx context.Context, workflow *testworkflowsv1.TestWo
 		}
 	}
 
+	// Finalize Pvcs
+	pvcs := layer.Pvcs()
+	for i := range pvcs {
+		AnnotateControlledBy(&pvcs[i], options.Config.Resource.RootId, options.Config.Resource.Id)
+		err = expressions.FinalizeForce(&pvcs[i], machines...)
+		if err != nil {
+			return nil, errors.Wrap(err, "finalizing Pvc")
+		}
+	}
+
 	// Finalize Secrets
 	secrets := append(layer.Secrets(), options.Secrets...)
 	for i := range secrets {
@@ -420,6 +430,7 @@ func (p *processor) Bundle(ctx context.Context, workflow *testworkflowsv1.TestWo
 	bundle = &Bundle{
 		ConfigMaps:    configMaps,
 		Secrets:       secrets,
+		Pvcs:          pvcs,
 		Job:           jobSpec,
 		Signature:     sig,
 		FullSignature: fullSig,

pkg/testworkflows/testworkflowresolver/apply.go

@@ -69,6 +69,7 @@ func injectTemplateToSpec(spec *testworkflowsv1.TestWorkflowSpec, template testw
 	spec.Services = MergeMap(common.MapMap(template.Spec.Services, ConvertIndependentServiceToService), spec.Services)
 	spec.Container = MergeContainerConfig(template.Spec.Container, spec.Container)
 	spec.System = MergeSystem(template.Spec.System, spec.System)
+	spec.Pvcs = MergeMap(template.Spec.Pvcs, spec.Pvcs)
 
 	// Include the steps from the template
 	setup := common.MapSlice(template.Spec.Setup, ConvertIndependentStepToStep)

pkg/testworkflows/testworkflowresolver/merge.go

@@ -12,6 +12,7 @@ import (
 	"maps"
 
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	testworkflowsv1 "github.com/kubeshop/testkube-operator/api/testworkflows/v1"
 	"github.com/kubeshop/testkube/internal/common"
@@ -414,3 +415,35 @@ func MergeTags(dst, src map[string]string) map[string]string {
 
 	return dst
 }
+
+func MergePvcConfig(dst, include *testworkflowsv1.TestWorkflowPvcConfig) *testworkflowsv1.TestWorkflowPvcConfig {
+	if dst == nil {
+		return include
+	} else if include == nil {
+		return dst
+	}
+	if include.Shared {
+		dst.Shared = include.Shared
+	}
+	dst.AccessModes = append(dst.AccessModes, include.AccessModes...)
+	if include.VolumeMode != "" {
+		dst.VolumeMode = include.VolumeMode
+	}
+	dst.Resources = MergeResources(dst.Resources, include.Resources)
+	if include.StorageClassName != "" {
+		dst.StorageClassName = include.StorageClassName
+	}
+	dst.Selector = MergeLabelSelector(dst.Selector, include.Selector)
+	return dst
+}
+
+func MergeLabelSelector(dst, include *metav1.LabelSelector) *metav1.LabelSelector {
+	if dst == nil {
+		return include
+	} else if include == nil {
+		return dst
+	}
+	dst.MatchLabels = common.MergeMaps(dst.MatchLabels, include.MatchLabels)
+	dst.MatchExpressions = append(dst.MatchExpressions, include.MatchExpressions...)
+	return dst
+}