Refactor upgrade logic to exclude WORKERS in patch upgrades

controllers/spec.go

@@ -560,22 +560,23 @@ func (s *DeploymentPlanSpec) toString() (string, error) {
 }
 
 type VersionUpgradeJobSpec struct {
-	Image              string            `json:"image,omitempty"`
-	JobName            string            `json:"jobName,omitempty"`
-	Labels             map[string]string `json:"labels,omitempty"`
-	HostName           string            `json:"hostName,omitempty"`
-	BackoffLimit       int32             `json:"backoffLimit,omitempty"`
-	ReferentName       string            `json:"referentName,omitempty"`
-	ReferentKind       string            `json:"referentKind,omitempty"`
-	ReferentApiVersion string            `json:"referentApiVersion,omitempty"`
-	ReferentUID        types.UID         `json:"referentUID,omitempty"`
-	SecuritySecret     string            `json:"securitySecret,omitempty"`
-	StartTimeMs        int64             `json:"startTimeMs,omitempty"`
-	Namespace          string            `json:"namespace,omitempty"`
-	CConf              string            `json:"cdapConf,omitempty"`
-	HConf              string            `json:"hadoopConf,omitempty"`
-	PreUpgrade         bool              `json:"preUpgrade,omitempty"`
-	PostUpgrade        bool              `json:"postUpgrade,omitempty"`
+	Image               string            `json:"image,omitempty"`
+	JobName             string            `json:"jobName,omitempty"`
+	Labels              map[string]string `json:"labels,omitempty"`
+	HostName            string            `json:"hostName,omitempty"`
+	BackoffLimit        int32             `json:"backoffLimit,omitempty"`
+	ReferentName        string            `json:"referentName,omitempty"`
+	ReferentKind        string            `json:"referentKind,omitempty"`
+	ReferentApiVersion  string            `json:"referentApiVersion,omitempty"`
+	ReferentUID         types.UID         `json:"referentUID,omitempty"`
+	SecuritySecret      string            `json:"securitySecret,omitempty"`
+	StartTimeMs         int64             `json:"startTimeMs,omitempty"`
+	Namespace           string            `json:"namespace,omitempty"`
+	CConf               string            `json:"cdapConf,omitempty"`
+	HConf               string            `json:"hadoopConf,omitempty"`
+	PreUpgrade          bool              `json:"preUpgrade,omitempty"`
+	PostUpgrade         bool              `json:"postUpgrade,omitempty"`
+	ExcludeProgramTypes []string          `json:"excludeProgramTypes,omitempty"`
 }
 
 func newUpgradeJobSpec(master *v1alpha1.CDAPMaster, name string, labels map[string]string, startTimeMs int64, cconf, hconf string) *VersionUpgradeJobSpec {
@@ -594,6 +595,8 @@ func newUpgradeJobSpec(master *v1alpha1.CDAPMaster, name string, labels map[stri
 	s.StartTimeMs = startTimeMs
 	s.CConf = cconf
 	s.HConf = hconf
+	s.ExcludeProgramTypes = []string{} // Default to empty list
+
 	return s
 }
 
@@ -606,3 +609,8 @@ func (s *VersionUpgradeJobSpec) SetPostUpgrade(isPostUpgrade bool) *VersionUpgra
 	s.PostUpgrade = isPostUpgrade
 	return s
 }
+
+func (s *VersionUpgradeJobSpec) SetExcludeProgramTypes(excludeProgramTypes []string) *VersionUpgradeJobSpec {
+	s.ExcludeProgramTypes = excludeProgramTypes
+	return s
+}

controllers/version_update.go

@@ -32,10 +32,21 @@ func init() {
 /////////////////////////////////////////////////////////////
 
 func handleVersionUpdate(master *v1alpha1.CDAPMaster, labels map[string]string, observed []reconciler.Object) ([]reconciler.Object, error) {
+	curVersion, err := getCurrentImageVersion(master)
+	if err != nil {
+		return nil, err
+	}
+	newVersion, err := getNewImageVersion(master)
+	if err != nil {
+		return nil, err
+	}
+	versionComparison := compareVersion(curVersion, newVersion)
+	isPatchUpgrade := versionComparison == -2
+
 	// Let the current update complete if there is any
 	if isConditionTrue(master, updateStatus.Inprogress) {
 		log.Printf("Version update ingress. Continue... ")
-		return upgradeForBackend(master, labels, observed)
+		return upgradeForBackend(master, labels, observed, isPatchUpgrade)
 	}
 
 	if objs, versionUpdated, err := updateForUserInterface(master); err != nil {
@@ -45,21 +56,13 @@ func handleVersionUpdate(master *v1alpha1.CDAPMaster, labels map[string]string,
 	}
 
 	// Update backend service image version
-	curVersion, err := getCurrentImageVersion(master)
-	if err != nil {
-		return nil, err
-	}
-	newVersion, err := getNewImageVersion(master)
-	if err != nil {
-		return nil, err
-	}
 	if len(curVersion.rawString) == 0 {
 		setImageToUse(master)
 		return []reconciler.Object{}, nil
 	}
 
-	switch compareVersion(curVersion, newVersion) {
-	case -1:
+	switch versionComparison {
+	case -2, -1:
 		// Upgrade case
 
 		// Don't retry upgrade if it failed.
@@ -73,7 +76,7 @@ func handleVersionUpdate(master *v1alpha1.CDAPMaster, labels map[string]string,
 		setCondition(master, updateStatus.Inprogress)
 		master.Status.UpgradeStartTimeMillis = getCurrentTimeMs()
 		log.Printf("Version update: start upgrading %s -> %s ", curVersion.rawString, newVersion.rawString)
-		return upgradeForBackend(master, labels, observed)
+		return upgradeForBackend(master, labels, observed, isPatchUpgrade)
 	case 0:
 		// Reset all condition so that failed upgraded/downgrade can be retried later if needed.
 		// This is needed when last upgrade failed and user has reset the version in spec.
@@ -120,7 +123,7 @@ func downgradeForBackend(master *v1alpha1.CDAPMaster) ([]reconciler.Object, erro
 	return []reconciler.Object{}, nil
 }
 
-func upgradeForBackend(master *v1alpha1.CDAPMaster, labels map[string]string, observed []reconciler.Object) ([]reconciler.Object, error) {
+func upgradeForBackend(master *v1alpha1.CDAPMaster, labels map[string]string, observed []reconciler.Object, isPatchUpgrade bool) ([]reconciler.Object, error) {
 	// Find either pre- or post- upgrade job
 	findJob := func(jobName string) *batchv1.Job {
 		var job *batchv1.Job = nil
@@ -163,7 +166,7 @@ func upgradeForBackend(master *v1alpha1.CDAPMaster, labels map[string]string, ob
 	if !isConditionTrue(master, updateStatus.PreUpgradeSucceeded) {
 		log.Printf("Version update: pre-upgrade job not completed")
 		preJobName := getPreUpgradeJobName(master.Status.UpgradeStartTimeMillis)
-		preJobSpec := buildPreUpgradeJobSpec(getPreUpgradeJobName(master.Status.UpgradeStartTimeMillis), master, labels)
+		preJobSpec := buildPreUpgradeJobSpec(getPreUpgradeJobName(master.Status.UpgradeStartTimeMillis), master, labels, isPatchUpgrade)
 		job := findJob(preJobName)
 		if job == nil {
 			obj, err := createJob(preJobSpec)
@@ -406,6 +409,7 @@ func parseImageString(imageString string) (*Version, error) {
 }
 
 // compare two parsed versions
+// -2: left < right, patch upgrade
 // -1: left < right
 // 0: left = right
 // 1: left > right
@@ -418,9 +422,24 @@ func compareVersion(l, r *Version) int {
 		return -1
 	}
 
+  lenL, lenR := len(l.components), len(r.components)
+  // Check if it only a patch upgrade
+  if lenL == lenR && lenL > 0 && l.components[lenL-1] < r.components[lenL-1] {
+    allEqual := true
+    for i := 0; i < lenL-1; i++ {
+      if l.components[i] != r.components[i] {
+        allEqual = false
+        break
+      }
+    }
+    if allEqual {
+      return -2
+    }
+  }
+
 	i := 0
 	j := 0
-	for i < len(l.components) && j < len(r.components) {
+	for i < lenL && j < lenR {
 		if l.components[i] > r.components[j] {
 			return 1
 		} else if l.components[i] < r.components[j] {
@@ -429,13 +448,13 @@ func compareVersion(l, r *Version) int {
 		i++
 		j++
 	}
-	for i < len(l.components) {
+	for i < lenL {
 		if l.components[i] > 0 {
 			return 1
 		}
 		i++
 	}
-	for j < len(r.components) {
+	for j < lenR {
 		if r.components[j] > 0 {
 			return 1
 		}
@@ -513,12 +532,19 @@ func getPostUpgradeJobName(startTimeMs int64) string {
 }
 
 // Return pre-upgrade job spec
-func buildPreUpgradeJobSpec(jobName string, master *v1alpha1.CDAPMaster, labels map[string]string) *VersionUpgradeJobSpec {
+func buildPreUpgradeJobSpec(jobName string, master *v1alpha1.CDAPMaster, labels map[string]string, isPatchUpgrade bool) *VersionUpgradeJobSpec {
 	startTimeMs := master.Status.UpgradeStartTimeMillis
 	cconf := getObjName(master, configMapCConf)
 	hconf := getObjName(master, configMapHConf)
 	name := getObjName(master, jobName)
-	return newUpgradeJobSpec(master, name, labels, startTimeMs, cconf, hconf).SetPreUpgrade(true)
+
+  excludeProgramTypes := []string{}
+  if isPatchUpgrade {
+    // Add "WORKERS" to ExcludeProgramTypes for patch upgrade
+    excludeProgramTypes = append(excludeProgramTypes, "WORKERS")
+  }
+
+	return newUpgradeJobSpec(master, name, labels, startTimeMs, cconf, hconf).SetPreUpgrade(true).SetExcludeProgramTypes(excludeProgramTypes)
 }
 
 // Return post-upgrade job spec

controllers/version_update_test.go

@@ -42,7 +42,7 @@ var _ = Describe("Controller Suite", func() {
 		It("Compare image versions", func() {
 			imagePairs := []Pair{
 				Pair{"gcr.io/cdapio/cdap:6.0.0.0", "gcr.io/cdapio/cdap:latest"},
-				Pair{"gcr.io/cdapio/cdap:6.0.0.0", "gcr.io/cdapio/cdap:6.0.0.1"},
+				Pair{"gcr.io/cdapio/cdap:6.0.0.0", "gcr.io/cdapio/cdap:6.0.1.0"},
 				Pair{"gcr.io/cdapio/cdap:6.0.0.0", "gcr.io/cdapio/cdap:6.1.0"},
 				Pair{"gcr.io/cdapio/cdap:6.0.0.0", "gcr.io/cdapio/cdap:7"},
 			}
@@ -55,6 +55,21 @@ var _ = Describe("Controller Suite", func() {
 				Expect(compareVersion(high, low)).To(Equal(1))
 			}
 		})
+		It("Compare image versions in patch upgrade", func() {
+			imagePairs := []Pair{
+				Pair{"gcr.io/cdapio/cdap:6.0.0.0", "gcr.io/cdapio/cdap:6.0.0.1"},
+				Pair{"gcr.io/cdapio/cdap:6.0.0.0", "gcr.io/cdapio/cdap:6.0.0.3"},
+				Pair{"gcr.io/cdapio/cdap:6.0.0.0", "gcr.io/cdapio/cdap:6.0.0.9"},
+			}
+			for _, imagePair := range imagePairs {
+				low, err := parseImageString(imagePair.first.(string))
+				Expect(err).To(BeNil())
+				high, err := parseImageString(imagePair.second.(string))
+				Expect(err).To(BeNil())
+				Expect(compareVersion(low, high)).To(Equal(-2))
+				Expect(compareVersion(high, low)).To(Equal(1))
+			}
+		})
 		It("Compare same image versions", func() {
 			imagePairs := []Pair{
 				Pair{"gcr.io/cdapio/cdap:latest", "gcr.io/cdapio/cdap:latest"},
@@ -199,7 +214,8 @@ var _ = Describe("Controller Suite", func() {
 					ImageToUse:             curUIImage,
 				},
 			}
-			postJobSpec := buildPreUpgradeJobSpec(getPreUpgradeJobName(master.Status.UpgradeStartTimeMillis), master, emptyLabels)
+			// Force the pre upgrade job configurations
+			postJobSpec := buildPreUpgradeJobSpec(getPreUpgradeJobName(master.Status.UpgradeStartTimeMillis), master, emptyLabels, false)
 			object, err := buildUpgradeJobObject(postJobSpec)
 			Expect(err).To(BeNil())
 

templates/upgrade-job.yaml

@@ -45,7 +45,13 @@ spec:
         {{end}}
         {{if .PreUpgrade}}
         - name: pre-upgrade
-          args: ["io.cdap.cdap.master.upgrade.UpgradeJobMain", "{{.HostName}}", "11015"]
+          args:
+          - "io.cdap.cdap.master.upgrade.UpgradeJobMain"
+          - "{{.HostName}}"
+          - "11015"
+          {{range .ExcludeProgramTypes}}
+          - "--exclude-program-type={{.}}"
+          {{end}}
         {{end}}
           image: {{.Image}}
           volumeMounts: