Re-adds DR Optimized feature

This change re-adds the isDROptimized feature
as it's now required in 4.17 which will be using Ceph 7.
Skipping the tests as there's a
 known bug that will be resolved in followup PR.
Fix for BZ-2311468.

Signed-off-by: raaizik <132667934+raaizik@users.noreply.github.com>

api/v1alpha1/odfinfoconfig.go

@@ -29,6 +29,7 @@ type InfoStorageCluster struct {
 	NamespacedName          types.NamespacedName `yaml:"namespacedName"`
 	StorageProviderEndpoint string               `yaml:"storageProviderEndpoint"`
 	CephClusterFSID         string               `yaml:"cephClusterFSID"`
+	IsDrOptimized           string               `yaml:"isDrOptimized"`
 }
 
 // OdfInfoData describes odf-info CM's data

controllers/storagecluster/initialization_reconciler_test.go

@@ -271,6 +271,7 @@ func initStorageClusterResourceCreateUpdateTestProviderMode(t *testing.T, runtim
 func initStorageClusterResourceCreateUpdateTest(t *testing.T, runtimeObjs []client.Object,
 	customSpec *api.StorageClusterSpec) (*testing.T, StorageClusterReconciler,
 	*api.StorageCluster, reconcile.Request) {
+	testSkipIsDrOptimized = true
 	cr := createDefaultStorageCluster()
 	if customSpec != nil {
 		_ = mergo.Merge(&cr.Spec, customSpec)
@@ -314,6 +315,7 @@ func initStorageClusterResourceCreateUpdateTest(t *testing.T, runtimeObjs []clie
 }
 
 func createFakeInitializationStorageClusterReconciler(t *testing.T, obj ...runtime.Object) StorageClusterReconciler {
+	testSkipIsDrOptimized = true
 	sc := &api.StorageCluster{}
 	scheme := createFakeScheme(t)
 	cfs := &cephv1.CephFilesystem{

controllers/storagecluster/odfinfoconfig.go

@@ -2,6 +2,8 @@ package storagecluster
 
 import (
 	"fmt"
+	rookCephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"strings"
 	"sync"
 
@@ -36,6 +38,8 @@ var _ resourceManager = &odfInfoConfig{}
 
 var mutex sync.RWMutex
 
+var testSkipIsDrOptimized = false
+
 // ensureCreated ensures that a ConfigMap resource exists with its Spec in
 // the desired state.
 func (obj *odfInfoConfig) ensureCreated(r *StorageClusterReconciler, storageCluster *ocsv1.StorageCluster) (reconcile.Result, error) {
@@ -142,6 +146,17 @@ func getOdfInfoData(r *StorageClusterReconciler, storageCluster *ocsv1.StorageCl
 		return "", err
 	}
 
+	var isDROptimized = "false"
+	// Set isDROptmized to "false" in case of external clusters as we currently don't have to way to determine
+	// if external cluster OSDs are using bluestore-rdr
+	if !storageCluster.Spec.ExternalStorage.Enable && !testSkipIsDrOptimized {
+		isDROptimized, err = getIsDROptimized(r, storageCluster)
+		if err != nil {
+			r.Log.Error(err, "failed to get cephcluster status. retrying again")
+			return "", err
+		}
+	}
+
 	data := ocsv1a1.OdfInfoData{
 		Version:           ocsVersion,
 		DeploymentType:    odfDeploymentType,
@@ -151,6 +166,7 @@ func getOdfInfoData(r *StorageClusterReconciler, storageCluster *ocsv1.StorageCl
 			NamespacedName:          client.ObjectKeyFromObject(storageCluster),
 			StorageProviderEndpoint: storageCluster.Status.StorageProviderEndpoint,
 			CephClusterFSID:         cephFSId,
+			IsDrOptimized:           fmt.Sprintf("%v", isDROptimized),
 		},
 	}
 	yamlData, err := yaml.Marshal(data)
@@ -234,3 +250,23 @@ func getCephFsid(r *StorageClusterReconciler, storageCluster *ocsv1.StorageClust
 
 	return string(val), nil
 }
+
+func getIsDROptimized(r *StorageClusterReconciler, storageCluster *ocsv1.StorageCluster) (string, error) {
+	var cephCluster rookCephv1.CephCluster
+	err := r.Client.Get(r.ctx, types.NamespacedName{Name: generateNameForCephClusterFromString(storageCluster.Name), Namespace: storageCluster.Namespace}, &cephCluster)
+	if err != nil {
+		return "false", err
+	}
+	if cephCluster.Status.CephStorage == nil || cephCluster.Status.CephStorage.OSD.StoreType == nil {
+		return "false", fmt.Errorf("cephcluster status does not have OSD store information")
+	}
+	bluestorerdr, ok := cephCluster.Status.CephStorage.OSD.StoreType["bluestore-rdr"]
+	if !ok {
+		return "false", nil
+	}
+	total := getOsdCount(storageCluster)
+	if bluestorerdr < total {
+		return "false", nil
+	}
+	return "true", nil
+}