DaemonSetJob: implement jobTemplate update (AmadeusITGroup#78)

Issue: AmadeusITGroup#71

- Add JobSpec md5 hash in Job annotation.
- Add more unittests for daemonsetjobcontroller.Sync method
- Add end2end test for DaemonSetJob

Signed-off-by: cedric lamoriniere <cedric.lamoriniere@amadeus.com>

Author: clamoriniere1A

app/workflow-controller.go

@@ -129,7 +129,7 @@ func NewWorkflowControllerApp(c *Config) *WorkflowController {
 	// configure readiness and liveness probes
 	health := healthcheck.NewHandler()
 	workflowCtrl.AddHealthCheck(health)
-	cronWorkflowCtrl.AddHealthCheck(health)
+	//cronWorkflowCtrl.AddHealthCheck(health)
 	daemonSetJobCtrl.AddHealthCheck(health)
 
 	return &WorkflowController{

hack/run-test-e2e.sh

@@ -30,4 +30,4 @@ popd
 #  - check if kubeconfig is there
 #  - check if cluster is up
 
-./test/e2e/e2e.test --kubeconfig=$HOME/.kube/config
+./test/e2e/e2e.test --kubeconfig=$HOME/.kube/config --ginkgo.slowSpecThreshold 60

pkg/api/daemonsetjob/v1/default.go

@@ -25,6 +25,5 @@ func DefaultDaemonSetJob(undefaultedDaemonSetJob *DaemonSetJob) *DaemonSetJob {
 	}
 	common.SetDefaults_Job(dummyJob)
 	d.Spec.JobTemplate.Spec = dummyJob.Spec
-
 	return d
 }

pkg/controller/common.go

@@ -9,6 +9,8 @@ import (
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 
 	batchv1listers "k8s.io/client-go/listers/batch/v1"
+
+	batch "k8s.io/api/batch/v1"
 )
 
 func deleteJobsFromLabelSelector(namespace string, jobsSelector labels.Selector, lister batchv1listers.JobLister, controler JobControlInterface) error {
@@ -29,3 +31,10 @@ func deleteJobsFromLabelSelector(namespace string, jobsSelector labels.Selector,
 	}
 	return utilerrors.NewAggregate(errs)
 }
+
+func compareJobSpecMD5Hash(hash string, job *batch.Job) bool {
+	if val, ok := job.Annotations[DaemonSetJobMD5AnnotationKey]; ok && val == hash {
+		return true
+	}
+	return false
+}

pkg/controller/common_test.go

@@ -1,10 +1,11 @@
 package controller
 
 import (
+	"testing"
+
 	batch "k8s.io/api/batch/v1"
 	batchv2 "k8s.io/api/batch/v2alpha1"
 	api "k8s.io/api/core/v1"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -33,3 +34,83 @@ func newJobTemplateSpec() *batchv2.JobTemplateSpec {
 		},
 	}
 }
+
+func Test_compareJobSpecMD5Hash(t *testing.T) {
+	jobSpec1 := batch.JobSpec{Template: api.PodTemplateSpec{Spec: api.PodSpec{Containers: []api.Container{{Name: "foo", Image: "foo:3.0.3"}}}}}
+	jobSpec2 := batch.JobSpec{Template: api.PodTemplateSpec{Spec: api.PodSpec{Containers: []api.Container{{Name: "foo", Image: "foo:4.0.8"}}}}}
+	jobSpec3 := batch.JobSpec{Template: api.PodTemplateSpec{Spec: api.PodSpec{Containers: []api.Container{{Name: "foo", Image: "foo:3.0.3"}, {Name: "bar", Image: "bar:latest"}}}}}
+	hashspec1, _ := generateMD5JobSpec(&jobSpec1)
+	t.Logf("hashspec1:%s", hashspec1)
+	hashspec2, _ := generateMD5JobSpec(&jobSpec2)
+	t.Logf("hashspec2:%s", hashspec2)
+	hashspec3, _ := generateMD5JobSpec(&jobSpec3)
+	t.Logf("hashspec3:%s", hashspec3)
+
+	type args struct {
+		hash string
+		job  *batch.Job
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "similar JobSpec",
+			args: args{
+				hash: hashspec1,
+				job: &batch.Job{
+					ObjectMeta: metav1.ObjectMeta{
+						Annotations: map[string]string{DaemonSetJobMD5AnnotationKey: hashspec1},
+					},
+					Spec: jobSpec1,
+				},
+			},
+			want: true,
+		},
+		{
+			name: "different JobSpec",
+			args: args{
+				hash: hashspec1,
+				job: &batch.Job{
+					ObjectMeta: metav1.ObjectMeta{
+						Annotations: map[string]string{DaemonSetJobMD5AnnotationKey: hashspec2},
+					},
+					Spec: jobSpec2,
+				},
+			},
+			want: false,
+		},
+		{
+			name: "PodTemplate changed",
+			args: args{
+				hash: hashspec1,
+				job: &batch.Job{
+					ObjectMeta: metav1.ObjectMeta{
+						Annotations: map[string]string{DaemonSetJobMD5AnnotationKey: string(hashspec3)},
+					},
+					Spec: jobSpec2,
+				},
+			},
+			want: false,
+		},
+		{
+			name: "missing annotation",
+			args: args{
+				hash: hashspec2,
+				job: &batch.Job{
+					ObjectMeta: metav1.ObjectMeta{},
+					Spec:       jobSpec2,
+				},
+			},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := compareJobSpecMD5Hash(tt.args.hash, tt.args.job); got != tt.want {
+				t.Errorf("compareJobSpecMD5Hash() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

pkg/controller/controller_util.go

@@ -89,7 +89,7 @@ func getJobLabelsSetFromDaemonSetJob(daemonsetjob *dapi.DaemonSetJob, template *
 	// Job should also get template.metadata.labels for monitoring metrics.
 	// These would ordinarily only be applied to the Pod running the Job,
 	// but the CronJob controller does apply them to its Job.
-	for k, v := range template.Spec.Template.ObjectMeta.Labels {
+	for k, v := range template.ObjectMeta.Labels {
 		desiredLabels[k] = v
 	}
 	desiredLabels[DaemonSetJobLabelKey] = daemonsetjob.Name // add daemonsetjob name to the job  labels

pkg/controller/daemonsetjob.go

@@ -43,7 +43,8 @@ type DaemonSetJobControllerConfig struct {
 
 // DaemonSetJobLabelKey defines the key of label to be injected by DaemonSetJob controller
 const (
-	DaemonSetJobLabelKey = "kubernetes.io/DaemonSetJob"
+	DaemonSetJobLabelKey         = "kubernetes.io/DaemonSetJob"
+	DaemonSetJobMD5AnnotationKey = "workflow.k8s.io/jobspec-md5"
 )
 
 // DaemonSetJobController represents the DaemonSetJob controller
@@ -288,6 +289,11 @@ func (d *DaemonSetJobController) manageDaemonSetJob(daemonsetjob *dapi.DaemonSet
 	daemonsetjobComplete := false
 	daemonsetjobFailed := false
 
+	md5JobSpec, err := generateMD5JobSpec(&daemonsetjob.Spec.JobTemplate.Spec)
+	if err != nil {
+		return fmt.Errorf("unable to generates the JobSpec MD5, %v", err)
+	}
+
 	jobSelector := labels.Set{DaemonSetJobLabelKey: daemonsetjob.Name}
 	jobs, err := d.JobLister.List(jobSelector.AsSelectorPreValidated())
 	jobsByNodeName := make(map[string]*batch.Job)
@@ -306,20 +312,40 @@ func (d *DaemonSetJobController) manageDaemonSetJob(daemonsetjob *dapi.DaemonSet
 	}
 	errs := []error{}
 	allJobs := []*batch.Job{}
+	jobToDelete := []*batch.Job{} // corresponding to the all DaemonSetJob Spec
 	for _, node := range nodes {
+		newJobCreation := false
 		job, ok := jobsByNodeName[node.Name]
 		if !ok {
+			glog.V(6).Infof("Job for the node %s not found %s/%s", node.Name, daemonsetjob.Namespace, daemonsetjob.Name)
+			newJobCreation = true
+		} else if job != nil {
+			if !compareJobSpecMD5Hash(md5JobSpec, job) {
+				glog.V(6).Infof("JobTemplateSpec has changed, %s/%s, previousMD5:%s, current:%s", job.Namespace, job.Name, md5JobSpec, job.GetAnnotations()[DaemonSetJobMD5AnnotationKey])
+				jobToDelete = append(jobToDelete, job)
+				newJobCreation = true
+			} else {
+				allJobs = append(allJobs, job)
+			}
+		}
+
+		if newJobCreation {
 			job, err = d.JobControl.CreateJobFromDaemonSetJob(daemonsetjob.Namespace, daemonsetjob.Spec.JobTemplate, daemonsetjob, node.Name)
 			if err != nil {
 				errs = append(errs, err)
 			}
 			daemonsetjobToBeUpdated = true
 		}
+	}
 
-		if job != nil {
-			allJobs = append(allJobs, job)
+	// Delete old jobs
+	for _, job := range jobToDelete {
+		// TODO maybe wait that the job is finished (success or failure) before deleting, need to define policy configuration
+		// Or maybe wait that the job is finished before creating the new job
+		err = d.JobControl.DeleteJob(job.Namespace, job.Name, job)
+		if err != nil {
+			errs = append(errs, err)
 		}
-
 	}
 
 	// build status
@@ -337,7 +363,6 @@ func (d *DaemonSetJobController) manageDaemonSetJob(daemonsetjob *dapi.DaemonSet
 		daemonsetjob.Status.Failed = failedJobs
 		daemonsetjobToBeUpdated = true
 	}
-
 	updateDaemonSetJobStatusConditions(&daemonsetjob.Status, now, daemonsetjobComplete, daemonsetjobFailed)
 	if daemonsetjobComplete {
 		glog.Infof("Workflow %s/%s complete.", daemonsetjob.Namespace, daemonsetjob.Name)
@@ -541,6 +566,13 @@ func IsDaemonSetJobFinished(d *dapi.DaemonSetJob) bool {
 	return false
 }
 
+// InferDaemonSetJobLabelSelectorForJobs returns labels.Selector corresponding to the associated Jobs
+func InferDaemonSetJobLabelSelectorForJobs(daemonsetjob *dapi.DaemonSetJob) labels.Selector {
+	set := fetchLabelsSetFromLabelSelector(daemonsetjob.Spec.Selector)
+	set[DaemonSetJobLabelKey] = daemonsetjob.Name
+	return labels.SelectorFromSet(set)
+}
+
 // get daemonsetjob by key method
 func (d *DaemonSetJobController) getDaemonSetJobByKey(key string) (*dapi.DaemonSetJob, error) {
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
@@ -590,7 +622,7 @@ func (d *DaemonSetJobController) pastActiveDeadline(daemonsetjob *dapi.DaemonSet
 func (d *DaemonSetJobController) deleteDaemonSetJobJobs(daemonsetjob *dapi.DaemonSetJob) error {
 	glog.V(6).Infof("deleting all jobs for daemonsetjob %s/%s", daemonsetjob.Namespace, daemonsetjob.Name)
 
-	jobsSelector := inferDaemonSetJobLabelSelectorForJobs(daemonsetjob)
+	jobsSelector := InferDaemonSetJobLabelSelectorForJobs(daemonsetjob)
 	return deleteJobsFromLabelSelector(daemonsetjob.Namespace, jobsSelector, d.JobLister, d.JobControl)
 }
 
@@ -656,9 +688,3 @@ func newDaemonSetJobStatusCondition(conditionType dapi.DaemonSetJobConditionType
 		Message:            message,
 	}
 }
-
-func inferDaemonSetJobLabelSelectorForJobs(daemonsetjob *dapi.DaemonSetJob) labels.Selector {
-	set := fetchLabelsSetFromLabelSelector(daemonsetjob.Spec.Selector)
-	set[DaemonSetJobLabelKey] = daemonsetjob.Name
-	return labels.SelectorFromSet(set)
-}