Merge pull request red-hat-storage#303 from egafford/persistent-failu…

…re-domain

Preserve rack FailureDomain if previously assigned

pkg/controller/storagecluster/reconcile.go

@@ -190,7 +190,7 @@ func (r *ReconcileStorageCluster) Reconcile(request reconcile.Request) (reconcil
 			instance.Status.CephBlockPoolsCreated = false
 			instance.Status.CephObjectStoreUsersCreated = false
 			instance.Status.CephFilesystemsCreated = false
-			instance.Status.FailureDomain = determineFailureDomain(instance.Status.NodeTopologies)
+			instance.Status.FailureDomain = determineFailureDomain(instance)
 			err = r.client.Status().Update(context.TODO(), instance)
 			if err != nil {
 				return reconcile.Result{}, err
@@ -371,17 +371,17 @@ func (r *ReconcileStorageCluster) reconcileNodeTopologyMap(sc *ocsv1.StorageClus
 						updated = true
 					}
 				}
-				if strings.Contains(label, "rack") {
-					if !nodeRacks.Contains(value, node.Name) {
-						nodeRacks.Add(value, node.Name)
-					}
+			}
+			if strings.Contains(label, "rack") {
+				if !nodeRacks.Contains(value, node.Name) {
+					nodeRacks.Add(value, node.Name)
 				}
 			}
 		}
 
 	}
 
-	if determineFailureDomain(topologyMap) == "rack" {
+	if determineFailureDomain(sc) == "rack" {
 		err = r.ensureNodeRacks(nodes, minNodes, nodeRacks, topologyMap, reqLogger)
 		if err != nil {
 			return err
@@ -664,18 +664,19 @@ func (r *ReconcileStorageCluster) ensureCephCluster(sc *ocsv1.StorageCluster, re
 
 // determineFailureDomain determines the appropriate Ceph failure domain based
 // on the storage cluster's topology map
-func determineFailureDomain(topologyMap *ocsv1.NodeTopologyMap) string {
+func determineFailureDomain(sc *ocsv1.StorageCluster) string {
+	if sc.Status.FailureDomain != "" {
+		return sc.Status.FailureDomain
+	}
+	topologyMap := sc.Status.NodeTopologies
 	failureDomain := "rack"
-
 	for label, labelValues := range topologyMap.Labels {
 		if strings.Contains(label, "zone") {
 			if len(labelValues) >= 3 {
 				failureDomain = "zone"
-				break
 			}
 		}
 	}
-
 	return failureDomain
 }
 
@@ -722,7 +723,7 @@ func newCephCluster(sc *ocsv1.StorageCluster, cephImage string) *cephv1.CephClus
 				RulesNamespace: "openshift-storage",
 			},
 			Storage: rook.StorageScopeSpec{
-				StorageClassDeviceSets: newStorageClassDeviceSets(sc.Spec.StorageDeviceSets, sc.Status.NodeTopologies),
+				StorageClassDeviceSets: newStorageClassDeviceSets(sc),
 				TopologyAware:          true,
 			},
 			Placement: rook.PlacementSpec{
@@ -770,7 +771,10 @@ func newCephDaemonResources(custom map[string]corev1.ResourceRequirements) map[s
 }
 
 // newStorageClassDeviceSets converts a list of StorageDeviceSets into a list of Rook StorageClassDeviceSets
-func newStorageClassDeviceSets(storageDeviceSets []ocsv1.StorageDeviceSet, topologyMap *ocsv1.NodeTopologyMap) []rook.StorageClassDeviceSet {
+func newStorageClassDeviceSets(sc *ocsv1.StorageCluster) []rook.StorageClassDeviceSet {
+	storageDeviceSets := sc.Spec.StorageDeviceSets
+	topologyMap := sc.Status.NodeTopologies
+
 	var storageClassDeviceSets []rook.StorageClassDeviceSet
 
 	for _, ds := range storageDeviceSets {
@@ -785,7 +789,7 @@ func newStorageClassDeviceSets(storageDeviceSets []ocsv1.StorageDeviceSet, topol
 
 		if noPlacement {
 			if topologyKey == "" {
-				topologyKey = determineFailureDomain(topologyMap)
+				topologyKey = determineFailureDomain(sc)
 			}
 			if topologyMap != nil {
 				topologyKey, topologyKeyValues = topologyMap.GetKeyValues(topologyKey)

pkg/controller/storagecluster/storagecluster_controller_test.go

@@ -220,6 +220,38 @@ func TestNodeTopologyMapNoNodes(t *testing.T) {
 	assert.Equal(t, err, fmt.Errorf("Not enough nodes found: Expected %d, found %d", defaults.DeviceSetReplica, len(nodeList.Items)))
 }
 
+func TestNodeTopologyMapPreexistingRack(t *testing.T) {
+	sc := &api.StorageCluster{}
+	mockStorageCluster.DeepCopyInto(sc)
+	nodeList := &corev1.NodeList{}
+	mockNodeList.DeepCopyInto(nodeList)
+	sc.Status.FailureDomain = "rack"
+
+	nodeTopologyMap := &api.NodeTopologyMap{
+		Labels: map[string]api.TopologyLabelValues{
+			zoneTopologyLabel: []string{
+				"zone1",
+				"zone2",
+				"zone3",
+			},
+			defaults.RackTopologyKey: []string{
+				"rack0",
+				"rack1",
+				"rack2",
+			},
+		},
+	}
+
+	reconciler := createFakeStorageClusterReconciler(t, sc, nodeList)
+	err := reconciler.reconcileNodeTopologyMap(sc, reconciler.reqLogger)
+	assert.NoError(t, err)
+
+	actual := &api.StorageCluster{}
+	err = reconciler.client.Get(nil, mockStorageClusterRequest.NamespacedName, actual)
+	assert.NoError(t, err)
+	assert.Equal(t, nodeTopologyMap, actual.Status.NodeTopologies)
+}
+
 func TestNodeTopologyMapTwoAZ(t *testing.T) {
 	sc := &api.StorageCluster{}
 	mockStorageCluster.DeepCopyInto(sc)
@@ -277,11 +309,13 @@ func TestNodeTopologyMapThreeAZ(t *testing.T) {
 }
 
 func TestFailureDomain(t *testing.T) {
+	sc := &api.StorageCluster{}
 	nodeTopologyMap := &api.NodeTopologyMap{
 		Labels: map[string]api.TopologyLabelValues{},
 	}
+	sc.Status.NodeTopologies = nodeTopologyMap
 
-	failureDomain := determineFailureDomain(nodeTopologyMap)
+	failureDomain := determineFailureDomain(sc)
 	assert.Equal(t, "rack", failureDomain)
 
 	nodeTopologyMap.Labels[zoneTopologyLabel] = []string{
@@ -290,7 +324,7 @@ func TestFailureDomain(t *testing.T) {
 		"zone3",
 	}
 
-	failureDomain = determineFailureDomain(nodeTopologyMap)
+	failureDomain = determineFailureDomain(sc)
 	assert.Equal(t, "zone", failureDomain)
 }
 
@@ -369,7 +403,8 @@ func TestStorageClusterCephClusterCreation(t *testing.T) {
 }
 
 func TestStorageClassDeviceSetCreation(t *testing.T) {
-	deviceSet := mockDeviceSets[0]
+	sc := &api.StorageCluster{}
+	sc.Spec.StorageDeviceSets = mockDeviceSets
 
 	nodeTopologyMap := &api.NodeTopologyMap{
 		Labels: map[string]api.TopologyLabelValues{
@@ -379,10 +414,12 @@ func TestStorageClassDeviceSetCreation(t *testing.T) {
 			},
 		},
 	}
+	sc.Status.NodeTopologies = nodeTopologyMap
 
-	actual := newStorageClassDeviceSets(mockDeviceSets, nodeTopologyMap)
+	actual := newStorageClassDeviceSets(sc)
 	assert.Equal(t, defaults.DeviceSetReplica, len(actual))
 
+	deviceSet := sc.Spec.StorageDeviceSets[0]
 	for i, scds := range actual {
 		assert.Equal(t, fmt.Sprintf("%s-%d", deviceSet.Name, i), scds.Name)
 		// TODO: Change this when OCP console is updated
@@ -395,7 +432,7 @@ func TestStorageClassDeviceSetCreation(t *testing.T) {
 
 	nodeTopologyMap.Labels[zoneTopologyLabel] = append(nodeTopologyMap.Labels[zoneTopologyLabel], "zone3")
 
-	actual = newStorageClassDeviceSets(mockDeviceSets, nodeTopologyMap)
+	actual = newStorageClassDeviceSets(sc)
 	assert.Equal(t, defaults.DeviceSetReplica, len(actual))
 
 	for i, scds := range actual {