review corrections

documentation/description.md

@@ -380,23 +380,23 @@ The `PersistentVolumes` are acquired using a `PersistentVolumeClaim` which is ma
 CassandraCluster fragment of persistent storage definition :
 
 ```
-# ...
+...
   # Global configuration
   dataCapacity: "300Gi"
   dataStorageClass: "local-storage"
   deletePVC: true
-  # ...
+  ...
   topology:
      dc:
        - name: dc1
          # DC level configuration
          dataCapacity: "10Gi"
          dataStorageClass: "test-storage"
-         # ...
-        - name: dc2
-         # ...
-  # ...
-# ...
+         ...
+       - name: dc2
+         ...
+  ...
+...
 
 ```
 
@@ -409,13 +409,13 @@ CassandraCluster fragment of persistent storage definition :
 
 In this example, all statefulsets related to the `dc2` will have the default configuration for the `data` PV :
 
-- `dataCapacity` : "300Gi"
-- `dataStorageClass`: "local-storage"
+- `dataCapacity` : 300Gi
+- `dataStorageClass`: local-storage
 
 All statefulsets related to the `dc1` will have the specific configuration for the `data` PV :
 
-- `dataCapacity` : "10Gi"
-- `dataStorageClass` : "test-storage"
+- `dataCapacity` : 10Gi
+- `dataStorageClass` : test-storage
 
 > **WARNING**: Resizing persistent storage for existing CassandraCluster is not currently supported. You must decide the
 > necessary storage size before deploying the cluster.

pkg/apis/db/v1alpha1/cassandracluster_types.go

@@ -504,70 +504,60 @@ func (cc *CassandraCluster) InitCassandraRackList() int {
 
 // GetDataCapacityForDC sends back the data capacity of cassandra nodes to uses for this dc
 func (cc *CassandraCluster) GetDataCapacityForDC(dcName string) string {
-	dataCapacity := cc.GetDataCapacityFromDCName(dcName)
-	return dataCapacity
+	return cc.GetDataCapacityFromDCName(dcName)
 }
 
 // GetDataCapacityFromDCName send DataCapacity used for the given dcName
 func (cc *CassandraCluster) GetDataCapacityFromDCName(dcName string) string {
-	dcSize := cc.GetDCSize()
-	if dcSize < 1 {
-		return cc.Spec.DataCapacity
-	}
-
-	for dc := 0; dc < dcSize; dc ++ {
-		if dcName == cc.GetDCName(dc) {
-			return cc.getDCDataCapacityFromIndex(dc)
+	dcIndex := cc.GetDCIndexFromDCName(dcName)
+	if dcIndex >= 0 {
+		dc := cc.getDCFromIndex(dcIndex)
+		if dc != nil && dc.DataCapacity != "" {
+			return dc.DataCapacity
 		}
-	}
-
-	return cc.Spec.DataCapacity
-}
-
-// getDCDataCapacityFromIndex send DataCapacity used for the given index
-func (cc *CassandraCluster) getDCDataCapacityFromIndex(dc int) string {
-	if dc >= cc.GetDCSize() {
 		return cc.Spec.DataCapacity
 	}
-	storeDC := cc.Spec.Topology.DC[dc]
-	if storeDC.DataCapacity == "" {
-		return cc.Spec.DataCapacity
-	}
-	return storeDC.DataCapacity
+	return cc.Spec.DataCapacity
 }
 
 // GetDataCapacityForDC sends back the data storage class of cassandra nodes to uses for this dc
 func (cc *CassandraCluster) GetDataStorageClassForDC(dcName string) string {
-	dataCapacity := cc.GetDataStorageClassFromDCName(dcName)
-	return dataCapacity
+	return cc.GetDataStorageClassFromDCName(dcName)
 }
 
 // GetDataCapacityFromDCName send DataStorageClass used for the given dcName
 func (cc *CassandraCluster) GetDataStorageClassFromDCName(dcName string) string {
+	dcIndex := cc.GetDCIndexFromDCName(dcName)
+	if dcIndex >= 0 {
+		dc := cc.getDCFromIndex(dcIndex)
+		if dc != nil && dc.DataCapacity != "" {
+			return dc.DataStorageClass
+		}
+		return cc.Spec.DataStorageClass
+	}
+	return cc.Spec.DataStorageClass
+}
+
+func (cc *CassandraCluster) GetDCIndexFromDCName(dcName string) int {
 	dcSize := cc.GetDCSize()
 	if dcSize < 1 {
-		return cc.Spec.DataStorageClass
+		return -1
 	}
 
 	for dc := 0; dc < dcSize; dc ++ {
 		if dcName == cc.GetDCName(dc) {
-			return cc.getDCDataStorageClassFromIndex(dc)
+			return dc
 		}
 	}
-
-	return cc.Spec.DataCapacity
+	return -1
 }
 
-// getDCDataCapacityFromIndex send DataStorageClass used for the given index
-func (cc *CassandraCluster) getDCDataStorageClassFromIndex(dc int) string {
+// getDCFromIndex send DC for the given index
+func (cc *CassandraCluster) getDCFromIndex(dc int) *DC {
 	if dc >= cc.GetDCSize() {
-		return cc.Spec.DataStorageClass
-	}
-	storeDC := cc.Spec.Topology.DC[dc]
-	if storeDC.DataStorageClass == "" {
-		return cc.Spec.DataStorageClass
+		return nil
 	}
-	return storeDC.DataStorageClass
+	return  &cc.Spec.Topology.DC[dc]
 }
 
 // GetNodesPerRacks sends back the number of cassandra nodes to uses for this dc-rack

pkg/controller/cassandracluster/generator_test.go

@@ -236,7 +236,7 @@ func checkLiveAndReadiNessProbe(t *testing.T, containers []v1.Container,
 
 func checkVolumeClaimTemplates(t *testing.T, expectedlabels map[string]string, pvcs []v1.PersistentVolumeClaim,
 	dataCapacity, dataClassStorage string) {
-	assert.Equal(t, len(pvcs), 3)
+	assert.Equal(t, 3, len(pvcs))
 	for _, pvc := range pvcs {
 		switch pvc.Name {
 		case "data":

pkg/controller/cassandracluster/testdata/cassandracluster-2DC.yaml

@@ -32,8 +32,8 @@ spec:
       - key: my_custom_taint
         operator: Exists
         effect: NoSchedule
-  dataCapacity: "3Gi"
-  dataStorageClass: "local-storage"
+  dataCapacity: 3Gi
+  dataStorageClass: local-storage
   hardAntiAffinity: false
   deletePVC: true
   storageConfigs:
@@ -89,8 +89,8 @@ spec:
   topology:
     dc:
       - name: dc1
-        dataCapacity: "10Gi"
-        dataStorageClass: "test-storage"
+        dataCapacity: 10Gi
+        dataStorageClass: test-storage
         labels:
           location.dfy.orange.com/site : mts
         rack: