Repository: honeydipper/honeydipper
Pull request #298 (2020-07-20 04:02:29)

chore(release): release
 * fix(dipper): map modifier handle nil value gracefully (#295)  
 * fix(drivers): watch API requries calling get and re-authentication (#297)


Patch:
diff --git a/drivers/cmd/kubernetes/main.go b/drivers/cmd/kubernetes/main.go
index 6fb6da9..5a0e1fa 100644
--- a/drivers/cmd/kubernetes/main.go
+++ b/drivers/cmd/kubernetes/main.go
@@ -25,6 +25,7 @@ import (
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 )
@@ -175,62 +176,109 @@ func getJobLog(m *dipper.Message) {
 }
 
 func waitForJob(m *dipper.Message) {
-	k8client := prepareKubeConfig(m)
-
+	m = dipper.DeserializePayload(m)
 	nameSpace, ok := dipper.GetMapDataStr(m.Payload, "namespace")
 	if !ok {
 		nameSpace = DefaultNamespace
 	}
 
 	jobName := dipper.MustGetMapDataStr(m.Payload, "job")
+
 	timeout := time.Duration(10)
-	timeoutStr, ok := m.Labels["timeout"]
-	if ok {
+	if timeoutStr, ok := m.Labels["timeout"]; ok {
 		timeoutInt, _ := strconv.Atoi(timeoutStr)
 		timeout = time.Duration(timeoutInt)
 	}
 
-	jobclient := k8client.BatchV1().Jobs(nameSpace)
-	watchOption := metav1.ListOptions{
-		FieldSelector: "metadata.name==" + jobName,
+	returnJobStatus := func(job *batchv1.Job) {
+		var (
+			jobStatus string = StatusSuccess
+			reason    []string
+		)
+		if job.Status.Failed > 0 {
+			jobStatus = StatusFailure
+			for _, condition := range job.Status.Conditions {
+				reason = append(reason, condition.Reason)
+			}
+		}
+		m.Reply <- dipper.Message{
+			Payload: map[string]interface{}{
+				"status": job.Status,
+			},
+			Labels: map[string]string{
+				"status": jobStatus,
+				"reason": strings.Join(reason, "\n"),
+			},
+		}
 	}
-	watchOption.Kind = "Job"
+
 	ctxWatch, cancelWatch := context.WithTimeout(context.Background(), timeout*time.Second)
 	defer cancelWatch()
-	jobstatus, err := jobclient.Watch(ctxWatch, watchOption)
-	if err != nil {
-		log.Panicf("[%s] unable to watch the job %+v", driver.Service, err)
-	}
-	defer jobstatus.Stop()
-
-	jobStatus := StatusSuccess
-	reason := []string{}
-loop:
-	for {
-		select {
-		case <-ctxWatch.Done():
-			break loop
-		case evt := <-jobstatus.ResultChan():
-			job := evt.Object.(*batchv1.Job)
-			if len(job.Status.Conditions) > 0 && job.Status.Active == 0 {
-				if job.Status.Failed > 0 {
-					jobStatus = StatusFailure
-					for _, condition := range job.Status.Conditions {
-						reason = append(reason, condition.Reason)
+	for EOW := false; !EOW; {
+		var job *batchv1.Job
+		k8client := prepareKubeConfig(m)
+		jobclient := k8client.BatchV1().Jobs(nameSpace)
+
+		watchOption := metav1.ListOptions{
+			FieldSelector: "metadata.name==" + jobName,
+		}
+		watchOption.Kind = "job"
+
+		func() {
+			listCtx, cancelList := context.WithTimeout(context.Background(), driver.APITimeout*time.Second)
+			defer cancelList()
+			joblist, err := jobclient.List(listCtx, watchOption)
+			if err != nil || len(joblist.Items) == 0 {
+				log.Panicf("[%s] job not found [%s]: %+v", driver.Service, jobName, err)
+			}
+			job = &joblist.Items[0]
+			watchOption.ResourceVersion = joblist.ResourceVersion
+		}()
+
+		if len(job.Status.Conditions) > 0 && job.Status.Active == 0 {
+			returnJobStatus(job)
+			break
+		}
+
+		jobstatus, err := jobclient.Watch(ctxWatch, watchOption)
+		if err != nil {
+			log.Panicf("[%s] unable to watch the job %+v", driver.Service, err)
+		}
+
+		func() {
+			defer jobstatus.Stop()
+
+		loop:
+			for {
+				select {
+				case <-ctxWatch.Done():
+					EOW = true
+					break loop
+				case evt := <-jobstatus.ResultChan():
+					switch evt.Type {
+					case watch.Error:
+						e := evt.Object.(*metav1.Status)
+						if e.Code == 410 {
+							log.Warningf("[%s] error from watching channel for job [%s]: %+v", driver.Service, jobName, evt.Object)
+							break loop
+						} else {
+							log.Panicf("[%s] error from watching channel for job [%s]: %+v", driver.Service, jobName, evt.Object)
+						}
+					default:
+						if evt.Object == nil {
+							break loop
+						}
+						job := evt.Object.(*batchv1.Job)
+						log.Debugf("[%s] receiving a event when watching for job [%s] %s: %+v", driver.Service, jobName, evt.Type, job.Status)
+						if len(job.Status.Conditions) > 0 && job.Status.Active == 0 {
+							returnJobStatus(job)
+							EOW = true
+							break loop
+						}
 					}
 				}
-				m.Reply <- dipper.Message{
-					Payload: map[string]interface{}{
-						"status": job.Status,
-					},
-					Labels: map[string]string{
-						"status": jobStatus,
-						"reason": strings.Join(reason, "\n"),
-					},
-				}
-				break loop
 			}
-		}
+		}()
 	}
 }
 
diff --git a/pkg/dipper/mapUtils.go b/pkg/dipper/mapUtils.go
index 7c1dd0b..8c536fb 100644
--- a/pkg/dipper/mapUtils.go
+++ b/pkg/dipper/mapUtils.go
@@ -335,12 +335,12 @@ func mergeModifier(dst map[string]interface{}) {
 		switch {
 		case k[len(k)-1] == '+': // append
 			ev, ok := dst[k[:len(k)-1]]
-			if !ok {
+			if !ok || ev == nil {
 				dst[k[:len(k)-1]] = v
 			} else {
 				if vstr, ok := v.(string); ok {
 					dst[k[:len(k)-1]] = ev.(string) + vstr
-				} else {
+				} else if v != nil {
 					dst[k[:len(k)-1]] = reflect.AppendSlice(reflect.ValueOf(ev), reflect.ValueOf(v)).Interface()
 				}
 			}


Test patch:
diff --git a/pkg/dipper/mapUtils_test.go b/pkg/dipper/mapUtils_test.go
index 6de4471..1ed73ed 100644
--- a/pkg/dipper/mapUtils_test.go
+++ b/pkg/dipper/mapUtils_test.go
@@ -126,3 +126,59 @@ func TestDeepCopy(t *testing.T) {
 	assert.NotEqual(t, src, ret)
 	assert.Nil(t, err)
 }
+
+func TestMerge(t *testing.T) {
+
+	testCases := []map[string]interface{}{
+		map[string]interface{}{
+			"name": "append modifier with nil in src",
+			"dst": map[string]interface{}{
+				"f1": "d1",
+				"f2": []interface{}{"item1", "item2"},
+			},
+			"src": map[string]interface{}{
+				"f2+": nil,
+			},
+			"expect": map[string]interface{}{
+				"f1": "d1",
+				"f2": []interface{}{"item1", "item2"},
+			},
+		},
+		{
+			"name": "append modifier with nil in dst",
+			"dst": map[string]interface{}{
+				"f1": "d1",
+				"f2": nil,
+			},
+			"src": map[string]interface{}{
+				"f2+": []interface{}{"item1", "item2"},
+			},
+			"expect": map[string]interface{}{
+				"f1": "d1",
+				"f2": []interface{}{"item1", "item2"},
+			},
+		},
+		{
+			"name": "append modifier missing key in dst",
+			"dst": map[string]interface{}{
+				"f1": "d1",
+			},
+			"src": map[string]interface{}{
+				"f2+": []interface{}{"item1", "item2"},
+			},
+			"expect": map[string]interface{}{
+				"f1": "d1",
+				"f2": []interface{}{"item1", "item2"},
+			},
+		},
+	}
+	for _, c := range testCases {
+		d, _ := c["dst"].(map[string]interface{})
+		s := c["src"]
+		e := c["expect"]
+		n := c["name"]
+
+		assert.NotPanics(t, func() { MergeMap(d, s) })
+		assert.Equal(t, e, d, n)
+	}
+}