diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go index 938313523..806a6ac93 100644 --- a/pkg/operator/ceph/cluster/osd/osd.go +++ b/pkg/operator/ceph/cluster/osd/osd.go @@ -26,22 +26,23 @@ import ( "strings" "time" - "k8s.io/client-go/kubernetes" - - "github.com/coreos/pkg/capnslog" - "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/clusterd" cephclient "github.com/rook/rook/pkg/daemon/ceph/client" osdconfig "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config" + "github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology" "github.com/rook/rook/pkg/operator/ceph/controller" cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" + + "github.com/coreos/pkg/capnslog" + "github.com/pkg/errors" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/client-go/kubernetes" ) var ( @@ -662,7 +663,7 @@ func getTopologyFromNode(ctx context.Context, clientset kubernetes.Interface, d if err != nil { return "", errors.Wrap(err, "failed to get the node for topology affinity") } - _, topologyAffinity := ExtractOSDTopologyFromLabels(node.Labels) + _, topologyAffinity := topology.ExtractOSDTopologyFromLabels(node.Labels) logger.Infof("found osd %d topology affinity at %q", osd.ID, topologyAffinity) return topologyAffinity, nil } @@ -722,7 +723,7 @@ func getNode(ctx context.Context, clientset kubernetes.Interface, nodeName strin } func updateLocationWithNodeLabels(location *[]string, nodeLabels map[string]string) string { - topology, topologyAffinity := ExtractOSDTopologyFromLabels(nodeLabels) + topology, topologyAffinity := topology.ExtractOSDTopologyFromLabels(nodeLabels) keys := make([]string, 0, len(topology)) for k := range topology { diff --git a/pkg/operator/ceph/cluster/osd/topology.go b/pkg/operator/ceph/cluster/osd/topology/topology.go similarity index 91% rename from pkg/operator/ceph/cluster/osd/topology.go rename to pkg/operator/ceph/cluster/osd/topology/topology.go index 548dd272b..c1128b951 100644 --- a/pkg/operator/ceph/cluster/osd/topology.go +++ b/pkg/operator/ceph/cluster/osd/topology/topology.go @@ -17,10 +17,11 @@ limitations under the License. // Package config provides methods for generating the Ceph config for a Ceph cluster and for // producing a "ceph.conf" compatible file from the config as well as Ceph command line-compatible // flags. -package osd +package topology import ( "fmt" + "strings" "github.com/rook/rook/pkg/daemon/ceph/client" corev1 "k8s.io/api/core/v1" @@ -121,3 +122,13 @@ func extractTopologyFromLabels(labels map[string]string) (map[string]string, str func formatTopologyAffinity(label, value string) string { return fmt.Sprintf("%s=%s", label, value) } + +// GetDefaultTopologyLabels returns the supported default topology labels. +func GetDefaultTopologyLabels() string { + Labels := []string{corev1.LabelHostname, corev1.LabelZoneRegionStable, corev1.LabelZoneFailureDomainStable} + for _, label := range CRUSHTopologyLabels { + Labels = append(Labels, topologyLabelPrefix+label) + } + + return strings.Join(Labels, ",") +} diff --git a/pkg/operator/ceph/cluster/osd/topology_test.go b/pkg/operator/ceph/cluster/osd/topology/topology_test.go similarity index 92% rename from pkg/operator/ceph/cluster/osd/topology_test.go rename to pkg/operator/ceph/cluster/osd/topology/topology_test.go index e0327d501..720414f15 100644 --- a/pkg/operator/ceph/cluster/osd/topology_test.go +++ b/pkg/operator/ceph/cluster/osd/topology/topology_test.go @@ -17,7 +17,7 @@ limitations under the License. // Package config provides methods for generating the Ceph config for a Ceph cluster and for // producing a "ceph.conf" compatible file from the config as well as Ceph command line-compatible // flags. -package osd +package topology import ( "testing" @@ -145,3 +145,17 @@ func TestTopologyLabels(t *testing.T) { assert.Equal(t, "zone", topology["zone"]) assert.Equal(t, "topology.kubernetes.io/zone=zone", affinity) } + +func TestGetDefaultTopologyLabels(t *testing.T) { + expectedLabels := "kubernetes.io/hostname," + + "topology.kubernetes.io/region," + + "topology.kubernetes.io/zone," + + "topology.rook.io/chassis," + + "topology.rook.io/rack," + + "topology.rook.io/row," + + "topology.rook.io/pdu," + + "topology.rook.io/pod," + + "topology.rook.io/room," + + "topology.rook.io/datacenter" + assert.Equal(t, expectedLabels, GetDefaultTopologyLabels()) +} diff --git a/pkg/operator/ceph/csi/csi.go b/pkg/operator/ceph/csi/csi.go index e0179401b..2e441037c 100644 --- a/pkg/operator/ceph/csi/csi.go +++ b/pkg/operator/ceph/csi/csi.go @@ -22,8 +22,10 @@ import ( "strings" "time" - "github.com/pkg/errors" + "github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology" "github.com/rook/rook/pkg/operator/k8sutil" + + "github.com/pkg/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/version" ) @@ -88,6 +90,11 @@ func (r *ReconcileCSI) setParams(ver *version.Info) error { return errors.Wrap(err, "failed to parse value for 'CSI_ENABLE_HOST_NETWORK'") } + if CSIParam.EnableReadAffinity, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_READ_AFFINITY", "false")); err != nil { + return errors.Wrap(err, "failed to parse value for 'CSI_ENABLE_READ_AFFINITY'") + } + CSIParam.CrushLocationLabels = k8sutil.GetValue(r.opConfig.Parameters, "CSI_CRUSH_LOCATION_LABELS", topology.GetDefaultTopologyLabels()) + // If not set or set to anything but "false", the kernel client will be enabled if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_FORCE_CEPHFS_KERNEL_CLIENT", "true"), "false") { CSIParam.ForceCephFSKernelClient = "false" diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go index 226da1bfc..35cd93ccb 100644 --- a/pkg/operator/ceph/csi/spec.go +++ b/pkg/operator/ceph/csi/spec.go @@ -62,6 +62,7 @@ type Param struct { ImagePullPolicy string CSIClusterName string CSIDomainLabels string + CrushLocationLabels string GRPCTimeout time.Duration CSIEnableMetadata bool EnablePluginSelinuxHostMount bool @@ -76,6 +77,7 @@ type Param struct { EnableCSIEncryption bool EnableCSITopology bool EnableLiveness bool + EnableReadAffinity bool LogLevel uint8 SidecarLogLevel uint8 CephFSGRPCMetricsPort uint16 diff --git a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml b/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml index 6f70adaf7..1db1a5c80 100644 --- a/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml +++ b/pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml @@ -79,6 +79,10 @@ spec: {{- if .EnableCSITopology }} - "--domainlabels={{ .CSIDomainLabels }}" {{- end }} + {{- if .EnableReadAffinity }} + - "--enable-read-affinity=true" + - "--crush-location-labels={{ .CrushLocationLabels }}" + {{- end }} env: - name: POD_IP valueFrom: diff --git a/pkg/operator/ceph/disruption/clusterdisruption/pools.go b/pkg/operator/ceph/disruption/clusterdisruption/pools.go index 3126702ab..2b55d42f5 100644 --- a/pkg/operator/ceph/disruption/clusterdisruption/pools.go +++ b/pkg/operator/ceph/disruption/clusterdisruption/pools.go @@ -19,19 +19,18 @@ package clusterdisruption import ( "fmt" - "sigs.k8s.io/controller-runtime/pkg/client" - "sigs.k8s.io/controller-runtime/pkg/reconcile" + cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" + "github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology" + "github.com/rook/rook/pkg/operator/k8sutil" "github.com/pkg/errors" - "github.com/rook/rook/pkg/operator/ceph/cluster/osd" - - cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/pkg/operator/k8sutil" policyv1 "k8s.io/api/policy/v1" policyv1beta1 "k8s.io/api/policy/v1beta1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/intstr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" ) func (r *ReconcileClusterDisruption) processPools(request reconcile.Request) (*cephv1.CephObjectStoreList, *cephv1.CephFilesystemList, string, int, error) { @@ -85,11 +84,11 @@ func getMinimumFailureDomain(poolList []cephv1.PoolSpec) string { } //start with max as the min - minfailureDomainIndex := len(osd.CRUSHMapLevelsOrdered) - 1 + minfailureDomainIndex := len(topology.CRUSHMapLevelsOrdered) - 1 matched := false for _, pool := range poolList { - for index, failureDomain := range osd.CRUSHMapLevelsOrdered { + for index, failureDomain := range topology.CRUSHMapLevelsOrdered { if index == minfailureDomainIndex { // index is higher-than/equal-to the min break @@ -105,7 +104,7 @@ func getMinimumFailureDomain(poolList []cephv1.PoolSpec) string { logger.Debugf("could not match failure domain. defaulting to %q", cephv1.DefaultFailureDomain) return cephv1.DefaultFailureDomain } - return osd.CRUSHMapLevelsOrdered[minfailureDomainIndex] + return topology.CRUSHMapLevelsOrdered[minfailureDomainIndex] } // Setting naive minAvailable for RGW at: n - 1