fix for duplicate applying nodes (#133)

Fixes #132

Additionally, address a logic bug introduced in #126 that could also
spin out of control.

See
https://github.com/rancher/system-upgrade-controller/commit/90e3d082a41fefaf09d14286b880c54a28af6c21



Signed-off-by: Jacob Blain Christen <jacob@rancher.com>

pkg/upgrade/handle_upgrade.go

@@ -7,6 +7,7 @@ import (
 	upgradeapiv1 "github.com/rancher/system-upgrade-controller/pkg/apis/upgrade.cattle.io/v1"
 	upgradectlv1 "github.com/rancher/system-upgrade-controller/pkg/generated/controllers/upgrade.cattle.io/v1"
 	upgradejob "github.com/rancher/system-upgrade-controller/pkg/upgrade/job"
+	upgradenode "github.com/rancher/system-upgrade-controller/pkg/upgrade/node"
 	upgradeplan "github.com/rancher/system-upgrade-controller/pkg/upgrade/plan"
 	"github.com/rancher/wrangler/pkg/generic"
 	"github.com/sirupsen/logrus"
@@ -72,10 +73,13 @@ func (ctl *Controller) handlePlans(ctx context.Context) error {
 			if err != nil {
 				return objects, status, err
 			}
-			for _, node := range concurrentNodes {
+			concurrentNodeNames := make([]string, len(concurrentNodes))
+			for i := range concurrentNodes {
+				node := concurrentNodes[i]
 				objects = append(objects, upgradejob.New(obj, node, ctl.Name))
-				obj.Status.Applying = append(obj.Status.Applying, node.Name)
+				concurrentNodeNames[i] = upgradenode.Hostname(node)
 			}
+			obj.Status.Applying = concurrentNodeNames[:]
 			return objects, obj.Status, nil
 		},
 		&generic.GeneratingHandlerOptions{

pkg/upgrade/job/job.go

@@ -8,6 +8,7 @@ import (
 	upgradeapi "github.com/rancher/system-upgrade-controller/pkg/apis/upgrade.cattle.io"
 	upgradeapiv1 "github.com/rancher/system-upgrade-controller/pkg/apis/upgrade.cattle.io/v1"
 	upgradectr "github.com/rancher/system-upgrade-controller/pkg/upgrade/container"
+	upgradenode "github.com/rancher/system-upgrade-controller/pkg/upgrade/node"
 	"github.com/rancher/wrangler/pkg/name"
 	"github.com/sirupsen/logrus"
 	batchv1 "k8s.io/api/batch/v1"
@@ -93,12 +94,6 @@ var (
 func New(plan *upgradeapiv1.Plan, node *corev1.Node, controllerName string) *batchv1.Job {
 	hostPathDirectory := corev1.HostPathDirectory
 	labelPlanName := upgradeapi.LabelPlanName(plan.Name)
-	nodeHostname := node.Name
-	if node.Labels != nil {
-		if hostname, ok := node.Labels[corev1.LabelHostname]; ok {
-			nodeHostname = hostname
-		}
-	}
 	job := &batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name.SafeConcatName("apply", plan.Name, "on", node.Name, "with", plan.Status.LatestHash),
@@ -141,7 +136,7 @@ func New(plan *upgradeapiv1.Plan, node *corev1.Node, controllerName string) *bat
 										Key:      corev1.LabelHostname,
 										Operator: corev1.NodeSelectorOpIn,
 										Values: []string{
-											nodeHostname,
+											upgradenode.Hostname(node),
 										},
 									}},
 								}},

pkg/upgrade/node/node.go

+package node
+
+import corev1 "k8s.io/api/core/v1"
+
+func Hostname(node *corev1.Node) string {
+	if node.Labels != nil {
+		if hostname, ok := node.Labels[corev1.LabelHostname]; ok {
+			return hostname
+		}
+	}
+	return node.Name
+}

pkg/upgrade/plan/plan.go

@@ -123,12 +123,24 @@ func ResolveChannel(ctx context.Context, url, latestVersion, clusterID string) (
 	return "", fmt.Errorf("unexpected response: %s %s", response.Proto, response.Status)
 }
 
+func NodeSelector(plan *upgradeapiv1.Plan) (labels.Selector, error) {
+	nodeSelector, err := metav1.LabelSelectorAsSelector(plan.Spec.NodeSelector)
+	if err != nil {
+		return nil, err
+	}
+	requireHostname, err := labels.NewRequirement(corev1.LabelHostname, selection.Exists, []string{})
+	if err != nil {
+		return nil, err
+	}
+	return nodeSelector.Add(*requireHostname), nil
+}
+
 func SelectConcurrentNodes(plan *upgradeapiv1.Plan, nodeCache corectlv1.NodeCache) ([]*corev1.Node, error) {
 	var (
 		applying = plan.Status.Applying
 		selected []*corev1.Node
 	)
-	nodeSelector, err := metav1.LabelSelectorAsSelector(plan.Spec.NodeSelector)
+	nodeSelector, err := NodeSelector(plan)
 	if err != nil {
 		return nil, err
 	}
@@ -177,7 +189,7 @@ func SelectConcurrentNodes(plan *upgradeapiv1.Plan, nodeCache corectlv1.NodeCach
 		}
 	}
 	sort.Slice(selected, func(i, j int) bool {
-		return selected[i].Name < selected[i].Name
+		return selected[i].Name < selected[j].Name
 	})
 	return selected, nil
 }