Ceph: Fix topologyAware in clusterdisruption package does not respect the rook topolgy prefix.

Allow the clusterdisruption package to recognize node topologies other than zone.

Signed-off-by: Rohan CJ <rohantmp@gmail.com>

cmd/rook/ceph/osd.go

@@ -301,37 +301,14 @@ func getLocation(clientset kubernetes.Interface) (string, error) {
 }
 
 func updateLocationWithNodeLabels(location *[]string, nodeLabels map[string]string) {
-	// get zone
-	zone, ok := nodeLabels[corev1.LabelZoneFailureDomain]
-	if ok {
-		client.UpdateCrushMapValue(location, "zone", client.NormalizeCrushName(zone))
-	}
-	// get region
-	region, ok := nodeLabels[corev1.LabelZoneRegion]
-	if ok {
-		client.UpdateCrushMapValue(location, "region", client.NormalizeCrushName(region))
-	}
 
-	// get the labels corresponding to other levels the CRUSH map such as topology.rook.io/rack
-	for label, value := range nodeLabels {
-		if strings.HasPrefix(label, k8sutil.TopologyLabelPrefix) {
-			keys := strings.Split(label, "/")
-			if len(keys) != 2 {
-				logger.Warningf("ignored invalid node label %q", label)
-				continue
-			}
-			crushType := keys[1]
-			validDomain := false
-			for _, domain := range oposd.CRUSHTopologyLabels {
-				if domain == crushType {
-					validDomain = true
-					client.UpdateCrushMapValue(location, crushType, client.NormalizeCrushName(value))
-					break
-				}
-			}
-			if !validDomain {
-				logger.Warningf("ignored invalid crush type %q in label %q", crushType, label)
-			}
+	topology, invalidLabels := oposd.ExtractRookTopologyFromLabels(nodeLabels)
+	if len(invalidLabels) > 0 {
+		logger.Warningf("ignored invalid node topology labels: %v", invalidLabels)
+	}
+	for topologyType, value := range topology {
+		if topologyType != "host" {
+			client.UpdateCrushMapValue(location, topologyType, value)
 		}
 	}
 }

cmd/rook/ceph/osd_test.go

@@ -106,16 +106,23 @@ func TestDetectCrushLocation(t *testing.T) {
 		"topology.rook.io/rack":                    "rack1",
 		"topology.rook.io/row":                     "row1",
 	}
+	expected := map[string]string{
+		"host":   "foo",
+		"region": "region1",
+		"zone":   "zone1",
+		"rack":   "rack1",
+		"row":    "row1",
+	}
 	updateLocationWithNodeLabels(&location, nodeLabels)
 	assert.Equal(t, 5, len(location))
-	assert.Equal(t, "host=foo", location[0])
-	assert.Equal(t, "zone=zone1", location[1])
-	assert.Equal(t, "region=region1", location[2])
-	if strings.HasPrefix(location[3], "rack") {
-		assert.Equal(t, "rack=rack1", location[3])
-		assert.Equal(t, "row=row1", location[4])
-	} else {
-		assert.Equal(t, "rack=rack1", location[4])
-		assert.Equal(t, "row=row1", location[3])
+	for _, locString := range location {
+		split := strings.Split(locString, "=")
+		assert.Len(t, split, 2)
+		prefix := split[0]
+		value := split[1]
+		expectedValue, ok := expected[prefix]
+		assert.True(t, ok)
+		assert.Equal(t, expectedValue, value)
 	}
+
 }

pkg/operator/ceph/cluster/osd/topology.go

@@ -19,7 +19,18 @@ limitations under the License.
 // flags.
 package osd
 
+import (
+	"fmt"
+	"strings"
+
+	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
 var (
+
 	// The labels that can be specified with the K8s labels such as failure-domain.beta.kubernetes.io/zone
 	// These are all at the top layers of the CRUSH map.
 	KubernetesTopologyLabels = []string{"zone", "region"}
@@ -30,3 +41,46 @@ var (
 	// The list of supported failure domains in the CRUSH map, ordered from lowest to highest
 	CRUSHMapLevelsOrdered = append([]string{"host"}, append(CRUSHTopologyLabels, KubernetesTopologyLabels...)...)
 )
+
+// ExtractRookTopologyFromLabels extracts rook topology from labels and returns a map from topology type to value,
+// and a bool indicating if there any invalid labels with a  topology prefix.
+func ExtractRookTopologyFromLabels(labels map[string]string) (map[string]string, []string) {
+	topology := make(map[string]string)
+
+	// get zone
+	zone, ok := labels[corev1.LabelZoneFailureDomain]
+	if ok {
+		topology["zone"] = client.NormalizeCrushName(zone)
+	}
+	// get region
+	region, ok := labels[corev1.LabelZoneRegion]
+	if ok {
+		topology["region"] = client.NormalizeCrushName(region)
+	}
+
+	// get host
+	host, ok := labels[corev1.LabelHostname]
+	if ok {
+		topology["host"] = client.NormalizeCrushName(host)
+	}
+
+	invalidEncountered := make([]string, 0)
+	for labelKey, labelValue := range labels {
+		for _, validTopologyType := range CRUSHTopologyLabels {
+			if strings.HasPrefix(labelKey, k8sutil.TopologyLabelPrefix) {
+				s := strings.Split(labelKey, "/")
+				if len(s) != 2 {
+					invalidEncountered = append(invalidEncountered, fmt.Sprintf("%s=%s", labelKey, labelValue))
+					continue
+				}
+				topologyType := s[1]
+				if topologyType == validTopologyType {
+					topology[validTopologyType] = client.NormalizeCrushName(labelValue)
+				}
+
+			}
+
+		}
+	}
+	return topology, invalidEncountered
+}

pkg/operator/ceph/disruption/clusterdisruption/location.go

@@ -169,29 +169,25 @@ func getOSDsForNodes(osdDataList []OsdData, nodeList []*corev1.Node, failureDoma
 			logger.Warningf("node in nodelist was nil")
 			continue
 		}
-		topologyLabelMap := map[string]string{
-			"host":   corev1.LabelHostname,
-			"zone":   corev1.LabelZoneFailureDomain,
-			"region": corev1.LabelZoneRegion,
-		}
-		failureDomainLabel, ok := topologyLabelMap[failureDomainType]
-		if !ok {
-			return nil, fmt.Errorf("invalid failure domain %q cannot manage PDBs for OSDs", failureDomainType)
-		}
-		nodeLabels := node.ObjectMeta.GetLabels()
+		nodeTopologyMap, _ := osd.ExtractRookTopologyFromLabels(node.GetLabels())
+
 		for _, osdData := range osdDataList {
-			secondaryCrushHostname := osdData.CrushMeta.Host
+			// get the crush location of the osd
 			crushFailureDomain, ok := osdData.CrushMeta.Location[failureDomainType]
-			if !ok && secondaryCrushHostname == "" {
-				return nil, fmt.Errorf("could not find the CrushFindResult.Location[%q] for %q", failureDomainType, osdData.Deployment.ObjectMeta.Name)
-			}
-			nodeFailureDomain, ok := nodeLabels[failureDomainLabel]
 			if !ok {
-				return nil, fmt.Errorf("could not find the %q label on node %q", failureDomainLabel, node.ObjectMeta.Name)
+				return nil, fmt.Errorf("could not find the CrushFindResult.Location[%q] for %s", failureDomainType, osdData.Deployment.GetName())
 			}
-			if cephClient.IsNormalizedCrushNameEqual(nodeFailureDomain, crushFailureDomain) || cephClient.IsNormalizedCrushNameEqual(secondaryCrushHostname, crushFailureDomain) {
+			// get the crush location of the node
+			nodeFailureDomain, ok := nodeTopologyMap[failureDomainType]
+			if !ok {
+				return nil, fmt.Errorf("could not find the %q failure domain on node %q", failureDomainType, node.GetName())
+			}
+
+			// check if the node and osd have the same crush location value for this particular crush location type
+			if cephClient.IsNormalizedCrushNameEqual(nodeFailureDomain, crushFailureDomain) {
 				nodeOsdDataList = append(nodeOsdDataList, osdData)
 			}
+
 		}
 	}
 	return nodeOsdDataList, nil