Merge pull request #498 from lminzhw/volumes

refresh volumes logic

Author: volcano-sh-bot

pkg/admission/admission_controller.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 
+	"k8s.io/kubernetes/pkg/apis/core/validation"
 	"volcano.sh/volcano/pkg/apis/batch/v1alpha1"
 	vcver "volcano.sh/volcano/pkg/client/clientset/versioned"
 )
@@ -241,6 +242,14 @@ func ValidateIO(volumes []v1alpha1.VolumeSpec) (string, bool) {
 		if _, found := volumeMap[volume.MountPath]; found {
 			return fmt.Sprintf(" duplicated mountPath: %s;", volume.MountPath), true
 		}
+		if len(volume.VolumeClaimName) != 0 {
+			if volume.VolumeClaim != nil {
+				return fmt.Sprintf("Confilct: If you want to use an existing PVC, just specify VolumeClaimName. If you want to create a new PVC, you do not need to specify VolumeClaimName."), true
+			}
+			if errMsgs := validation.ValidatePersistentVolumeName(volume.VolumeClaimName, false); len(errMsgs) > 0 {
+				return fmt.Sprintf("Illegal VolumeClaimName %s : %v", volume.VolumeClaimName, errMsgs), true
+			}
+		}
 		volumeMap[volume.MountPath] = true
 	}
 	return "", false

pkg/controllers/job/helpers/helpers.go

@@ -49,7 +49,8 @@ func MakePodName(jobName string, taskName string, index int) string {
 	return fmt.Sprintf(PodNameFmt, jobName, taskName, index)
 }
 
-func genRandomStr(l int) string {
+// GenRandomStr generate random str with specified length l
+func GenRandomStr(l int) string {
 	str := "0123456789abcdefghijklmnopqrstuvwxyz"
 	bytes := []byte(str)
 	var result []byte
@@ -62,7 +63,7 @@ func genRandomStr(l int) string {
 
 // MakeVolumeClaimName creates volume claim name
 func MakeVolumeClaimName(jobName string) string {
-	return fmt.Sprintf(VolumeClaimFmt, jobName, genRandomStr(12))
+	return fmt.Sprintf(VolumeClaimFmt, jobName, GenRandomStr(12))
 }
 
 // GetJobKeyByReq gets the key for the job request

pkg/controllers/job/job_controller_actions.go

@@ -17,6 +17,7 @@ limitations under the License.
 package job
 
 import (
+	"errors"
 	"fmt"
 	"sort"
 	"sync"
@@ -157,19 +158,19 @@ func (cc *Controller) createJob(job *vkv1.Job) (*vkv1.Job, error) {
 		return nil, err
 	}
 
-	if err := cc.createPodGroupIfNotExist(job); err != nil {
-		cc.recorder.Event(job, v1.EventTypeWarning, string(vkv1.PodGroupError),
-			fmt.Sprintf("Failed to create PodGroup, err: %v", err))
-		return nil, err
-	}
-
 	newJob, err := cc.createJobIOIfNotExist(job)
 	if err != nil {
 		cc.recorder.Event(job, v1.EventTypeWarning, string(vkv1.PVCError),
 			fmt.Sprintf("Failed to create PVC, err: %v", err))
 		return nil, err
 	}
 
+	if err := cc.createPodGroupIfNotExist(newJob); err != nil {
+		cc.recorder.Event(job, v1.EventTypeWarning, string(vkv1.PodGroupError),
+			fmt.Sprintf("Failed to create PodGroup, err: %v", err))
+		return nil, err
+	}
+
 	return newJob, nil
 }
 
@@ -336,18 +337,20 @@ func (cc *Controller) syncJob(jobInfo *apis.JobInfo, updateStatus state.UpdateSt
 
 func (cc *Controller) createJobIOIfNotExist(job *vkv1.Job) (*vkv1.Job, error) {
 	// If PVC does not exist, create them for Job.
-	var needUpdate, nameExist bool
+	var needUpdate bool
 	volumes := job.Spec.Volumes
+	if job.Status.ControlledResources == nil {
+		job.Status.ControlledResources = make(map[string]string)
+	}
 	for index, volume := range volumes {
-		nameExist = false
 		vcName := volume.VolumeClaimName
 		if len(vcName) == 0 {
 			//NOTE(k82cn): Ensure never have duplicated generated names.
 			for {
 				vcName = vkjobhelpers.MakeVolumeClaimName(job.Name)
 				exist, err := cc.checkPVCExist(job, vcName)
 				if err != nil {
-					return nil, err
+					return job, err
 				}
 				if exist {
 					continue
@@ -356,37 +359,40 @@ func (cc *Controller) createJobIOIfNotExist(job *vkv1.Job) (*vkv1.Job, error) {
 				needUpdate = true
 				break
 			}
-		} else {
-			exist, err := cc.checkPVCExist(job, vcName)
-			if err != nil {
-				return nil, err
-			}
-			nameExist = exist
-		}
-
-		if !nameExist {
-			if job.Status.ControlledResources == nil {
-				job.Status.ControlledResources = make(map[string]string)
-			}
 			if volume.VolumeClaim != nil {
 				if err := cc.createPVC(job, vcName, volume.VolumeClaim); err != nil {
-					return nil, err
+					return job, err
 				}
 				job.Status.ControlledResources["volume-pvc-"+vcName] = vcName
 			} else {
 				job.Status.ControlledResources["volume-emptyDir-"+vcName] = vcName
 			}
+		} else {
+			if job.Status.ControlledResources["volume-emptyDir-"+vcName] == vcName || job.Status.ControlledResources["volume-pvc-"+vcName] == vcName {
+				continue
+			}
+			exist, err := cc.checkPVCExist(job, vcName)
+			if err != nil {
+				return job, err
+			}
+			if exist {
+				job.Status.ControlledResources["volume-pvc-"+vcName] = vcName
+			} else {
+				msg := fmt.Sprintf("pvc %s is not found, the job will be in the Pending state until the PVC is created", vcName)
+				glog.Error(msg)
+				return job, errors.New(msg)
+			}
 		}
 	}
 	if needUpdate {
 		newJob, err := cc.vkClients.BatchV1alpha1().Jobs(job.Namespace).Update(job)
 		if err != nil {
 			glog.Errorf("Failed to update Job %v/%v for volume claim name: %v ",
 				job.Namespace, job.Name, err)
-			return nil, err
+			return job, err
 		}
-		newJob.Status = job.Status
 
+		newJob.Status = job.Status
 		return newJob, err
 	}
 	return job, nil

pkg/controllers/job/job_controller_actions_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package job
 
 import (
+	"errors"
 	"fmt"
 	"testing"
 
@@ -295,7 +296,7 @@ func TestCreateJobIOIfNotExistFunc(t *testing.T) {
 		ExpextVal error
 	}{
 		{
-			Name: "Create Job IO success case",
+			Name: "Create Job IO case",
 			Job: &v1alpha1.Job{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "job1",
@@ -309,7 +310,7 @@ func TestCreateJobIOIfNotExistFunc(t *testing.T) {
 					},
 				},
 			},
-			ExpextVal: nil,
+			ExpextVal: errors.New("pvc pvc1 is not found, the job will be in the Pending state until the PVC is created"),
 		},
 	}
 
@@ -319,8 +320,14 @@ func TestCreateJobIOIfNotExistFunc(t *testing.T) {
 			fakeController := newFakeController()
 
 			job, err := fakeController.createJobIOIfNotExist(testcase.Job)
-			if err != testcase.ExpextVal {
-				t.Errorf("Expected Return value to be : %s, but got: %s in testcase %d", testcase.ExpextVal, err, i)
+			if testcase.ExpextVal == nil {
+				if err != nil {
+					t.Errorf("Expected Return value to be : %v, but got: %v in testcase %d", testcase.ExpextVal, err, i)
+				}
+			} else {
+				if err == nil || err.Error() != testcase.ExpextVal.Error() {
+					t.Errorf("Expected Return value to be : %v, but got: %v in testcase %d", testcase.ExpextVal.Error(), err.Error(), i)
+				}
 			}
 
 			if len(job.Spec.Volumes) == 0 {

pkg/controllers/job/job_controller_util.go

@@ -59,32 +59,35 @@ func createJobPod(job *vkv1.Job, template *v1.PodTemplateSpec, ix int) *v1.Pod {
 		pod.Spec.SchedulerName = job.Spec.SchedulerName
 	}
 
-	volumeMap := make(map[string]bool)
+	volumeMap := make(map[string]string)
 	for _, volume := range job.Spec.Volumes {
 		vcName := volume.VolumeClaimName
+		name := fmt.Sprintf("%s-%s", job.Name, vkjobhelpers.GenRandomStr(12))
 		if _, ok := volumeMap[vcName]; !ok {
-			if _, ok := job.Status.ControlledResources["volume-emptyDir-"+vcName]; ok && volume.VolumeClaim == nil {
+			if _, ok := job.Status.ControlledResources["volume-emptyDir-"+vcName]; ok {
 				volume := v1.Volume{
-					Name: vcName,
+					Name: name,
 				}
 				volume.EmptyDir = &v1.EmptyDirVolumeSource{}
 				pod.Spec.Volumes = append(pod.Spec.Volumes, volume)
 			} else {
 				volume := v1.Volume{
-					Name: vcName,
+					Name: name,
 				}
 				volume.PersistentVolumeClaim = &v1.PersistentVolumeClaimVolumeSource{
 					ClaimName: vcName,
 				}
 				pod.Spec.Volumes = append(pod.Spec.Volumes, volume)
 			}
-			volumeMap[vcName] = true
+			volumeMap[vcName] = name
+		} else {
+			name = volumeMap[vcName]
 		}
 
 		for i, c := range pod.Spec.Containers {
 			vm := v1.VolumeMount{
 				MountPath: volume.MountPath,
-				Name:      vcName,
+				Name:      name,
 			}
 			pod.Spec.Containers[i].VolumeMounts = append(c.VolumeMounts, vm)
 		}

test/e2e/job_controlled_resource.go

@@ -125,6 +125,17 @@ var _ = Describe("Job E2E Test: Test Job PVCs", func() {
 
 		_, err1 := context.kubeclient.CoreV1().PersistentVolumeClaims(namespace).Create(&pvc)
 		Expect(err1).NotTo(HaveOccurred(), "pvc creation")
+
+		pvSpec := &v12.PersistentVolumeClaimSpec{
+			Resources: v12.ResourceRequirements{
+				Requests: v12.ResourceList{
+					v12.ResourceName(v12.ResourceStorage): resource.MustParse("1Gi"),
+				},
+			},
+			AccessModes: []v12.PersistentVolumeAccessMode{
+				v12.ReadWriteOnce,
+			},
+		}
 		job := createJob(context, &jobSpec{
 			namespace: namespace,
 			name:      jobName,
@@ -139,9 +150,12 @@ var _ = Describe("Job E2E Test: Test Job PVCs", func() {
 			},
 			volumes: []v1alpha1.VolumeSpec{
 				{
-					MountPath:       "/mounttwo",
+					MountPath:       "/mountone",
 					VolumeClaimName: pvcName,
-					VolumeClaim:     &pvc.Spec,
+				},
+				{
+					MountPath:   "/mounttwo",
+					VolumeClaim: pvSpec,
 				},
 			},
 		})
@@ -152,12 +166,12 @@ var _ = Describe("Job E2E Test: Test Job PVCs", func() {
 		job, err = context.vcclient.BatchV1alpha1().Jobs(namespace).Get(jobName, metav1.GetOptions{})
 		Expect(err).NotTo(HaveOccurred())
 
-		Expect(len(job.Spec.Volumes)).To(Equal(1),
+		Expect(len(job.Spec.Volumes)).To(Equal(2),
 			" volume should be created")
-		for _, volume := range job.Spec.Volumes {
-			Expect(volume.VolumeClaimName).Should((Equal(pvcName)),
-				"PVC name should not be generated .")
-		}
+		Expect(job.Spec.Volumes[0].VolumeClaimName).Should(Equal(pvcName),
+			"volume 1 PVC name should not be generated .")
+		Expect(job.Spec.Volumes[1].VolumeClaimName).Should(Not(Equal("")),
+			"volume 0 PVC name should be generated.")
 	})
 
 	It("Generate PodGroup and valid minResource when creating job", func() {