Merge pull request #2965 from OdedViner/strech_cluster_zones

 Restrict Node Label Matching for Supported Topology Keys in NodeTopologyMap

api/v1/topologymap.go

@@ -17,7 +17,7 @@ limitations under the License.
 package v1
 
 import (
-	"strings"
+	corev1 "k8s.io/api/core/v1"
 )
 
 // NewNodeTopologyMap returns an initialized NodeTopologyMap
@@ -61,18 +61,31 @@ func (m *NodeTopologyMap) Add(topologyKey string, value string) {
 	m.Labels[topologyKey] = append(m.Labels[topologyKey], value)
 }
 
-// GetKeyValues returns a node label matching the topologyKey and all values
-// for that label across all storage nodes
+// GetKeyValues returns a node label matching the supported topologyKey and all values
+// for that label across all storage nodes.
 func (m *NodeTopologyMap) GetKeyValues(topologyKey string) (string, []string) {
 	values := []string{}
 
+	// Supported failure domain labels
+	supportedLabels := map[string]string{
+		"rack":     "topology.rook.io/rack",
+		"hostname": corev1.LabelHostname,
+		"zone":     corev1.LabelZoneFailureDomainStable,
+	}
+
+	// Get the specific label based on the topologyKey
+	expectedLabel, exists := supportedLabels[topologyKey]
+	if !exists {
+		return "", values // Return empty if the topologyKey is unsupported
+	}
+
+	// Match the expected label and fetch the values
 	for label, labelValues := range m.Labels {
-		if strings.Contains(label, topologyKey) {
-			topologyKey = label
+		if label == expectedLabel {
 			values = labelValues
-			break
+			return label, values
 		}
 	}
 
-	return topologyKey, values
+	return "", values // Return empty if no match is found
 }

controllers/storagecluster/cephcluster_test.go

@@ -489,7 +489,7 @@ func TestStorageClassDeviceSetCreation(t *testing.T) {
 	sc1.Spec.StorageDeviceSets = mockDeviceSets
 	sc1.Status.NodeTopologies = &ocsv1.NodeTopologyMap{
 		Labels: map[string]ocsv1.TopologyLabelValues{
-			zoneTopologyLabel: []string{
+			corev1.LabelZoneFailureDomainStable: []string{
 				"zone1",
 				"zone2",
 			},
@@ -501,7 +501,7 @@ func TestStorageClassDeviceSetCreation(t *testing.T) {
 	sc2.Spec.StorageDeviceSets = mockDeviceSets
 	sc2.Status.NodeTopologies = &ocsv1.NodeTopologyMap{
 		Labels: map[string]ocsv1.TopologyLabelValues{
-			zoneTopologyLabel: []string{
+			corev1.LabelZoneFailureDomainStable: []string{
 				"zone1",
 				"zone2",
 				"zone3",
@@ -514,7 +514,7 @@ func TestStorageClassDeviceSetCreation(t *testing.T) {
 	sc3.Spec.StorageDeviceSets = mockDeviceSets
 	sc3.Status.NodeTopologies = &ocsv1.NodeTopologyMap{
 		Labels: map[string]ocsv1.TopologyLabelValues{
-			zoneTopologyLabel: []string{
+			corev1.LabelZoneFailureDomainStable: []string{
 				"zone1",
 				"zone2",
 				"zone3",
@@ -572,6 +572,7 @@ func TestStorageClassDeviceSetCreation(t *testing.T) {
 
 	for _, c := range cases {
 		t.Logf("Case: %s\n", c.label)
+		setFailureDomain(c.sc)
 		actual := newStorageClassDeviceSets(c.sc)
 		assert.Equal(t, defaults.DeviceSetReplica, len(actual))
 		deviceSet := c.sc.Spec.StorageDeviceSets[0]
@@ -607,8 +608,8 @@ func TestStorageClassDeviceSetCreation(t *testing.T) {
 
 			if c.topologyKey == "rack" {
 				assert.Equal(t, defaults.RackTopologyKey, topologyKey)
-			} else if len(c.sc.Status.NodeTopologies.Labels[zoneTopologyLabel]) >= defaults.DeviceSetReplica {
-				assert.Equal(t, zoneTopologyLabel, topologyKey)
+			} else if len(c.sc.Status.NodeTopologies.Labels[corev1.LabelZoneFailureDomainStable]) >= defaults.DeviceSetReplica {
+				assert.Equal(t, corev1.LabelZoneFailureDomainStable, topologyKey)
 			} else {
 				assert.Equal(t, hostnameLabel, topologyKey)
 			}
@@ -765,15 +766,15 @@ func TestStorageClassDeviceSetCreationForArbiter(t *testing.T) {
 	sc1.Spec.StorageDeviceSets = getMockDeviceSets("mock", 1, 4, true)
 	sc1.Status.NodeTopologies = &ocsv1.NodeTopologyMap{
 		Labels: map[string]ocsv1.TopologyLabelValues{
-			zoneTopologyLabel: []string{
+			corev1.LabelZoneFailureDomainStable: []string{
 				"zone1",
 				"zone2",
 			},
 		},
 		ArbiterLocation: "zone3",
 	}
 	sc1.Status.FailureDomain = "zone"
-	sc1.Status.FailureDomainKey = zoneTopologyLabel
+	sc1.Status.FailureDomainKey = corev1.LabelZoneFailureDomainStable
 	sc1.Status.FailureDomainValues = []string{"zone1", "zone2"}
 
 	sc2 := &ocsv1.StorageCluster{}
@@ -783,11 +784,11 @@ func TestStorageClassDeviceSetCreationForArbiter(t *testing.T) {
 	}
 	sc2.Spec.StorageDeviceSets = getMockDeviceSets("mock", 1, 6, true)
 	sc2.Status.NodeTopologies = &ocsv1.NodeTopologyMap{
-		Labels:          map[string]ocsv1.TopologyLabelValues{zoneTopologyLabel: []string{"zone1", "zone2"}},
+		Labels:          map[string]ocsv1.TopologyLabelValues{corev1.LabelZoneFailureDomainStable: []string{"zone1", "zone2"}},
 		ArbiterLocation: "zone3",
 	}
 	sc2.Status.FailureDomain = "zone"
-	sc2.Status.FailureDomainKey = zoneTopologyLabel
+	sc2.Status.FailureDomainKey = corev1.LabelZoneFailureDomainStable
 	sc2.Status.FailureDomainValues = []string{"zone1", "zone2"}
 
 	cases := []struct {
@@ -798,12 +799,12 @@ func TestStorageClassDeviceSetCreationForArbiter(t *testing.T) {
 		{
 			label:       "case 1",
 			sc:          sc1,
-			topologyKey: zoneTopologyLabel,
+			topologyKey: corev1.LabelZoneFailureDomainStable,
 		},
 		{
 			label:       "case 2",
 			sc:          sc2,
-			topologyKey: zoneTopologyLabel,
+			topologyKey: corev1.LabelZoneFailureDomainStable,
 		},
 	}
 

controllers/storagecluster/placement_test.go

@@ -381,7 +381,7 @@ func TestGetPlacement(t *testing.T) {
 											},
 										},
 									},
-									TopologyKey: zoneTopologyLabel,
+									TopologyKey: corev1.LabelZoneFailureDomainStable,
 								},
 							},
 						},
@@ -402,15 +402,15 @@ func TestGetPlacement(t *testing.T) {
 										},
 									},
 								},
-								TopologyKey: zoneTopologyLabel,
+								TopologyKey: corev1.LabelZoneFailureDomainStable,
 							},
 						},
 					},
 				},
 			},
 			topologyMap: &ocsv1.NodeTopologyMap{
 				Labels: map[string]ocsv1.TopologyLabelValues{
-					zoneTopologyLabel: []string{
+					corev1.LabelZoneFailureDomainStable: []string{
 						"zone1",
 						"zone2",
 						"zone3",
@@ -497,7 +497,9 @@ func TestGetPlacement(t *testing.T) {
 		sc.Spec.Placement = c.placements
 		sc.Spec.LabelSelector = c.labelSelector
 		sc.Status.NodeTopologies = c.topologyMap
-
+		if c.topologyMap != nil {
+			setFailureDomain(sc)
+		}
 		expectedPlacement := c.expectedPlacements["all"]
 		actualPlacement = getPlacement(sc, "all")
 		assert.Equal(t, expectedPlacement, actualPlacement, c.label)