Merge pull request #6759 from travisn/aggregate-affinity

ceph: Merge placement node affinity for stretch clusters

pkg/apis/rook.io/v1/placement.go

@@ -29,7 +29,7 @@ func (p Placement) ApplyToPodSpec(t *v1.PodSpec) {
 		t.Affinity = &v1.Affinity{}
 	}
 	if p.NodeAffinity != nil {
-		t.Affinity.NodeAffinity = p.NodeAffinity.DeepCopy()
+		t.Affinity.NodeAffinity = p.mergeNodeAffinity(t.Affinity.NodeAffinity)
 	}
 	if p.PodAffinity != nil {
 		t.Affinity.PodAffinity = p.PodAffinity.DeepCopy()
@@ -45,6 +45,53 @@ func (p Placement) ApplyToPodSpec(t *v1.PodSpec) {
 	}
 }
 
+func (p Placement) mergeNodeAffinity(nodeAffinity *v1.NodeAffinity) *v1.NodeAffinity {
+	// no node affinity is specified yet, so return the placement's nodeAffinity
+	result := p.NodeAffinity.DeepCopy()
+	if nodeAffinity == nil {
+		return result
+	}
+
+	// merge the preferred node affinity that was already specified, and the placement's nodeAffinity
+	result.PreferredDuringSchedulingIgnoredDuringExecution = append(
+		nodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution,
+		p.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution...)
+
+	// nothing to merge if no affinity was passed in
+	if nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution == nil {
+		return result
+	}
+	// take the desired affinity if there was none on the placement
+	if p.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution == nil {
+		result.RequiredDuringSchedulingIgnoredDuringExecution = nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution
+		return result
+	}
+	// take the desired affinity node selectors without the need to merge
+	if len(nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms) == 0 {
+		return result
+	}
+	// take the placement affinity node selectors without the need to merge
+	if len(p.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms) == 0 {
+		// take the placement from the first option since the second isn't specified
+		result.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms =
+			nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms
+		return result
+	}
+
+	// merge the match expressions together since they are defined in both placements
+	// this will only work if we want an "and" between all the expressions, more complex conditions won't work with this merge
+	var nodeTerm v1.NodeSelectorTerm
+	nodeTerm.MatchExpressions = append(
+		nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions,
+		p.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions...)
+	nodeTerm.MatchFields = append(
+		nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchFields,
+		p.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchFields...)
+	result.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0] = nodeTerm
+
+	return result
+}
+
 // Merge returns a Placement which results from merging the attributes of the
 // original Placement with the attributes of the supplied one. The supplied
 // Placement's attributes will override the original ones if defined.

pkg/apis/rook.io/v1/placement_test.go

@@ -25,7 +25,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func TestPlacement_spec(t *testing.T) {
+func TestPlacementSpec(t *testing.T) {
 	specYaml := []byte(`
 nodeAffinity:
   requiredDuringSchedulingIgnoredDuringExecution:
@@ -91,6 +91,34 @@ topologySpreadConstraints:
 	assert.Equal(t, expected, placement)
 }
 
+func TestMergeNodeAffinity(t *testing.T) {
+	// affinity is nil
+	p := Placement{}
+	result := p.mergeNodeAffinity(nil)
+	assert.Nil(t, result)
+
+	// node affinity is only set on the placement and should remain unchanged
+	p.NodeAffinity = placementTestGenerateNodeAffinity()
+	result = p.mergeNodeAffinity(nil)
+	assert.Equal(t, p.NodeAffinity, result)
+
+	// preferred set, but required not set
+	affinityToMerge := placementTestGenerateNodeAffinity()
+	affinityToMerge.RequiredDuringSchedulingIgnoredDuringExecution = nil
+	result = p.mergeNodeAffinity(affinityToMerge)
+	assert.Equal(t, 2, len(result.PreferredDuringSchedulingIgnoredDuringExecution))
+	assert.Equal(t, p.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution, result.RequiredDuringSchedulingIgnoredDuringExecution)
+
+	// preferred and required expressions set
+	affinityToMerge = placementTestGenerateNodeAffinity()
+	affinityToMerge.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions[0].Key = "baz"
+	result = p.mergeNodeAffinity(affinityToMerge)
+	assert.Equal(t, 2, len(result.PreferredDuringSchedulingIgnoredDuringExecution))
+	assert.Equal(t, 2, len(result.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions))
+	assert.Equal(t, "baz", result.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions[0].Key)
+	assert.Equal(t, "foo", result.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions[1].Key)
+}
+
 func TestPlacementApplyToPodSpec(t *testing.T) {
 	to := placementTestGetTolerations("foo", "bar")
 	na := placementTestGenerateNodeAffinity()
@@ -143,6 +171,7 @@ func TestPlacementApplyToPodSpec(t *testing.T) {
 	p.ApplyToPodSpec(ps)
 	assert.Equal(t, expected, ps)
 
+	// The preferred affinity is merged from both sources to result in two node affinities
 	p = Placement{NodeAffinity: na, PodAntiAffinity: antiaffinity}
 	nap := placementTestGenerateNodeAffinity()
 	nap.PreferredDuringSchedulingIgnoredDuringExecution[0].Weight = 5
@@ -152,7 +181,7 @@ func TestPlacementApplyToPodSpec(t *testing.T) {
 		TopologySpreadConstraints: tc,
 	}
 	p.ApplyToPodSpec(ps)
-	assert.Equal(t, expected, ps)
+	assert.Equal(t, 2, len(ps.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution))
 }
 
 func TestPlacementMerge(t *testing.T) {

pkg/operator/ceph/controller/controller_utils.go

@@ -55,7 +55,7 @@ func CheckForCancelledOrchestration(context *clusterd.Context) error {
 
 	// Check whether we need to cancel the orchestration
 	if context.RequestCancelOrchestration.IsSet() {
-		return errors.New("CANCELLING CURRENT ORCHESTATION")
+		return errors.New("CANCELLING CURRENT ORCHESTRATION")
 	}
 
 	return nil

pkg/operator/k8sutil/node.go

@@ -289,7 +289,6 @@ func ExtractTopologyFromLabels(labels map[string]string, additionalTopologyIDs [
 	}
 
 	// check for the region k8s topology label that is GA in 1.17.
-	// TODO: Replace with a const when we update to the K8s 1.17 go-client.
 	region, ok = labels[corev1.LabelZoneRegionStable]
 	if ok {
 		topology[regionLabel] = region
@@ -303,7 +302,6 @@ func ExtractTopologyFromLabels(labels map[string]string, additionalTopologyIDs [
 	}
 
 	// check for the zone k8s topology label that is GA in 1.17.
-	// TODO: Replace with a const when we update to the K8s 1.17 go-client.
 	zone, ok = labels[corev1.LabelZoneFailureDomainStable]
 	if ok {
 		topology[zoneLabel] = zone