csi: move setting csi param to setParams
moving setting csi parameter to setParams function as `setParams` is mostly used to set the csi parameters for the templates. Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>
This commit is contained in:
Madhu Rajanna committed
1 parent
529809ff53
commit
c255e35225
2 files changed
+153
-151
No files matched your search
@@ -17,10 +17,14 @@ limitations under the License.
|
||||
package csi
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
"github.com/rook/rook/pkg/operator/k8sutil"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/version"
|
||||
)
|
||||
|
||||
@@ -30,7 +34,7 @@ func (r *ReconcileCSI) validateAndConfigureDrivers(serverVersion *version.Info,
|
||||
err error
|
||||
)
|
||||
|
||||
if err = r.setParams(); err != nil {
|
||||
if err = r.setParams(serverVersion); err != nil {
|
||||
return errors.Wrapf(err, "failed to configure CSI parameters")
|
||||
}
|
||||
|
||||
@@ -56,7 +60,7 @@ func (r *ReconcileCSI) validateAndConfigureDrivers(serverVersion *version.Info,
|
||||
return r.stopDrivers(serverVersion)
|
||||
}
|
||||
|
||||
func (r *ReconcileCSI) setParams() error {
|
||||
func (r *ReconcileCSI) setParams(ver *version.Info) error {
|
||||
var err error
|
||||
|
||||
if EnableRBD, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_ENABLE_RBD", "true")); err != nil {
|
||||
@@ -78,11 +82,154 @@ func (r *ReconcileCSI) setParams() error {
|
||||
if EnableCSIGRPCMetrics, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_ENABLE_GRPC_METRICS", "false")); err != nil {
|
||||
return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_GRPC_METRICS'")
|
||||
}
|
||||
CSIParam.EnableCSIGRPCMetrics = fmt.Sprintf("%t", EnableCSIGRPCMetrics)
|
||||
|
||||
if CSIParam.EnableCSIHostNetwork, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_HOST_NETWORK", "true")); err != nil {
|
||||
return errors.Wrap(err, "failed to parse value for 'CSI_ENABLE_HOST_NETWORK'")
|
||||
}
|
||||
|
||||
// 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"
|
||||
} else {
|
||||
CSIParam.ForceCephFSKernelClient = "true"
|
||||
}
|
||||
|
||||
// parse RPC timeout
|
||||
timeout := k8sutil.GetValue(r.opConfig.Parameters, grpcTimeout, "150")
|
||||
timeoutSeconds, err := strconv.Atoi(timeout)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to parse %q. Defaulting to %d. %v", grpcTimeout, defaultGRPCTimeout, err)
|
||||
timeoutSeconds = defaultGRPCTimeout
|
||||
}
|
||||
if timeoutSeconds < 120 {
|
||||
logger.Warningf("%s is %q but it should be >= 120, setting the default value %d", grpcTimeout, timeout, defaultGRPCTimeout)
|
||||
timeoutSeconds = defaultGRPCTimeout
|
||||
}
|
||||
CSIParam.GRPCTimeout = time.Duration(timeoutSeconds) * time.Second
|
||||
|
||||
// parse GRPC and Liveness ports
|
||||
CSIParam.CephFSGRPCMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_CEPHFS_GRPC_METRICS_PORT", DefaultCephFSGRPCMerticsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error getting CSI CephFS GRPC metrics port.")
|
||||
}
|
||||
CSIParam.CephFSLivenessMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_CEPHFS_LIVENESS_METRICS_PORT", DefaultCephFSLivenessMerticsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error getting CSI CephFS liveness metrics port.")
|
||||
}
|
||||
|
||||
CSIParam.RBDGRPCMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_RBD_GRPC_METRICS_PORT", DefaultRBDGRPCMerticsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error getting CSI RBD GRPC metrics port.")
|
||||
}
|
||||
CSIParam.CSIAddonsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSIADDONS_PORT", DefaultCSIAddonsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to get CSI Addons port")
|
||||
}
|
||||
CSIParam.RBDLivenessMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_RBD_LIVENESS_METRICS_PORT", DefaultRBDLivenessMerticsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error getting CSI RBD liveness metrics port.")
|
||||
}
|
||||
|
||||
// default value `system-node-critical` is the highest available priority
|
||||
CSIParam.PluginPriorityClassName = k8sutil.GetValue(r.opConfig.Parameters, "CSI_PLUGIN_PRIORITY_CLASSNAME", "")
|
||||
|
||||
// default value `system-cluster-critical` is applied for some
|
||||
// critical pods in cluster but less priority than plugin pods
|
||||
CSIParam.ProvisionerPriorityClassName = k8sutil.GetValue(r.opConfig.Parameters, "CSI_PROVISIONER_PRIORITY_CLASSNAME", "")
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_OMAP_GENERATOR", "false"), "true") {
|
||||
CSIParam.EnableOMAPGenerator = true
|
||||
}
|
||||
|
||||
// SA token projection is stable only from kubernetes version 1.20.
|
||||
if ver.Major == KubeMinMajor && ver.Minor >= KubeMinVerForOIDCTokenProjection {
|
||||
CSIParam.EnableOIDCTokenProjection = true
|
||||
}
|
||||
|
||||
// if k8s >= v1.17 enable RBD and CephFS snapshotter by default
|
||||
if ver.Major == KubeMinMajor && ver.Minor >= kubeMinVerForSnapshot {
|
||||
CSIParam.EnableRBDSnapshotter = true
|
||||
CSIParam.EnableCephFSSnapshotter = true
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_RBD_SNAPSHOTTER", "true"), "false") {
|
||||
CSIParam.EnableRBDSnapshotter = false
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_CEPHFS_SNAPSHOTTER", "true"), "false") {
|
||||
CSIParam.EnableCephFSSnapshotter = false
|
||||
}
|
||||
|
||||
CSIParam.EnableVolumeReplicationSideCar = false
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_VOLUME_REPLICATION", "false"), "true") {
|
||||
CSIParam.EnableVolumeReplicationSideCar = true
|
||||
}
|
||||
|
||||
CSIParam.EnableCSIAddonsSideCar = false
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_CSIADDONS", "false"), "true") {
|
||||
CSIParam.EnableCSIAddonsSideCar = true
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_ENCRYPTION", "false"), "true") {
|
||||
CSIParam.EnableCSIEncryption = true
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
|
||||
CSIParam.CephFSPluginUpdateStrategy = onDelete
|
||||
} else {
|
||||
CSIParam.CephFSPluginUpdateStrategy = rollingUpdate
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_NFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
|
||||
CSIParam.NFSPluginUpdateStrategy = onDelete
|
||||
} else {
|
||||
CSIParam.NFSPluginUpdateStrategy = rollingUpdate
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_RBD_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
|
||||
CSIParam.RBDPluginUpdateStrategy = onDelete
|
||||
} else {
|
||||
CSIParam.RBDPluginUpdateStrategy = rollingUpdate
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_PLUGIN_ENABLE_SELINUX_HOST_MOUNT", "false"), "true") {
|
||||
CSIParam.EnablePluginSelinuxHostMount = true
|
||||
}
|
||||
|
||||
logger.Infof("Kubernetes version is %s.%s", ver.Major, ver.Minor)
|
||||
|
||||
CSIParam.ResizerImage = k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_RESIZER_IMAGE", DefaultResizerImage)
|
||||
|
||||
logLevel := k8sutil.GetValue(r.opConfig.Parameters, "CSI_LOG_LEVEL", "")
|
||||
CSIParam.LogLevel = defaultLogLevel
|
||||
if logLevel != "" {
|
||||
l, err := strconv.ParseUint(logLevel, 10, 8)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to parse CSI_LOG_LEVEL. Defaulting to %d. %v", defaultLogLevel, err)
|
||||
} else {
|
||||
CSIParam.LogLevel = uint8(l)
|
||||
}
|
||||
}
|
||||
|
||||
CSIParam.ProvisionerReplicas = defaultProvisionerReplicas
|
||||
nodes, err := r.context.Clientset.CoreV1().Nodes().List(r.opManagerContext, metav1.ListOptions{})
|
||||
if err == nil {
|
||||
if len(nodes.Items) == 1 {
|
||||
CSIParam.ProvisionerReplicas = 1
|
||||
} else {
|
||||
replicas := k8sutil.GetValue(r.opConfig.Parameters, "CSI_PROVISIONER_REPLICAS", "2")
|
||||
r, err := strconv.ParseInt(replicas, 10, 32)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to parse CSI_PROVISIONER_REPLICAS. Defaulting to %d. %v", defaultProvisionerReplicas, err)
|
||||
} else {
|
||||
CSIParam.ProvisionerReplicas = int32(r)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
logger.Errorf("failed to get nodes. Defaulting the number of replicas of provisioner pods to %d. %v", CSIParam.ProvisionerReplicas, err)
|
||||
}
|
||||
|
||||
CSIParam.CSIPluginImage = k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_CEPH_IMAGE", DefaultCSIPluginImage)
|
||||
CSIParam.NFSPluginImage = k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_NFS_IMAGE", DefaultNFSPluginImage)
|
||||
CSIParam.RegistrarImage = k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_REGISTRAR_IMAGE", DefaultRegistrarImage)
|
||||
|
||||
@@ -20,8 +20,6 @@ import (
|
||||
"context"
|
||||
_ "embed"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/rook/rook/pkg/operator/k8sutil"
|
||||
@@ -260,6 +258,10 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
|
||||
RBDDriverName = tp.DriverNamePrefix + "rbd.csi.ceph.com"
|
||||
NFSDriverName = tp.DriverNamePrefix + "nfs.csi.ceph.com"
|
||||
|
||||
if CustomCSICephConfigExists {
|
||||
CSIParam.MountCustomCephConf = v.SupportsCustomCephConf()
|
||||
}
|
||||
|
||||
csiDriverobj = beta1CsiDriver{}
|
||||
if ver.Major > KubeMinMajor || ver.Major == KubeMinMajor && ver.Minor >= kubeMinVerForV1csiDriver {
|
||||
csiDriverobj = v1CsiDriver{}
|
||||
@@ -282,153 +284,6 @@ func (r *ReconcileCSI) startDrivers(ver *version.Info, ownerInfo *k8sutil.OwnerI
|
||||
}
|
||||
}
|
||||
|
||||
tp.EnableCSIGRPCMetrics = fmt.Sprintf("%t", EnableCSIGRPCMetrics)
|
||||
|
||||
// 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") {
|
||||
tp.ForceCephFSKernelClient = "false"
|
||||
} else {
|
||||
tp.ForceCephFSKernelClient = "true"
|
||||
}
|
||||
|
||||
// parese RPC timeout
|
||||
timeout := k8sutil.GetValue(r.opConfig.Parameters, grpcTimeout, "150")
|
||||
timeoutSeconds, err := strconv.Atoi(timeout)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to parse %q. Defaulting to %d. %v", grpcTimeout, defaultGRPCTimeout, err)
|
||||
timeoutSeconds = defaultGRPCTimeout
|
||||
}
|
||||
if timeoutSeconds < 120 {
|
||||
logger.Warningf("%s is %q but it should be >= 120, setting the default value %d", grpcTimeout, timeout, defaultGRPCTimeout)
|
||||
timeoutSeconds = defaultGRPCTimeout
|
||||
}
|
||||
tp.GRPCTimeout = time.Duration(timeoutSeconds) * time.Second
|
||||
|
||||
// parse GRPC and Liveness ports
|
||||
tp.CephFSGRPCMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_CEPHFS_GRPC_METRICS_PORT", DefaultCephFSGRPCMerticsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error getting CSI CephFS GRPC metrics port.")
|
||||
}
|
||||
tp.CephFSLivenessMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_CEPHFS_LIVENESS_METRICS_PORT", DefaultCephFSLivenessMerticsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error getting CSI CephFS liveness metrics port.")
|
||||
}
|
||||
|
||||
tp.RBDGRPCMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_RBD_GRPC_METRICS_PORT", DefaultRBDGRPCMerticsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error getting CSI RBD GRPC metrics port.")
|
||||
}
|
||||
tp.CSIAddonsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSIADDONS_PORT", DefaultCSIAddonsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to get CSI Addons port")
|
||||
}
|
||||
tp.RBDLivenessMetricsPort, err = getPortFromConfig(r.opConfig.Parameters, "CSI_RBD_LIVENESS_METRICS_PORT", DefaultRBDLivenessMerticsPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "error getting CSI RBD liveness metrics port.")
|
||||
}
|
||||
|
||||
// default value `system-node-critical` is the highest available priority
|
||||
tp.PluginPriorityClassName = k8sutil.GetValue(r.opConfig.Parameters, "CSI_PLUGIN_PRIORITY_CLASSNAME", "")
|
||||
|
||||
// default value `system-cluster-critical` is applied for some
|
||||
// critical pods in cluster but less priority than plugin pods
|
||||
tp.ProvisionerPriorityClassName = k8sutil.GetValue(r.opConfig.Parameters, "CSI_PROVISIONER_PRIORITY_CLASSNAME", "")
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_OMAP_GENERATOR", "false"), "true") {
|
||||
tp.EnableOMAPGenerator = true
|
||||
}
|
||||
|
||||
// SA token projection is stable only from kubernetes version 1.20.
|
||||
if ver.Major == KubeMinMajor && ver.Minor >= KubeMinVerForOIDCTokenProjection {
|
||||
tp.EnableOIDCTokenProjection = true
|
||||
}
|
||||
|
||||
// if k8s >= v1.17 enable RBD and CephFS snapshotter by default
|
||||
if ver.Major == KubeMinMajor && ver.Minor >= kubeMinVerForSnapshot {
|
||||
tp.EnableRBDSnapshotter = true
|
||||
tp.EnableCephFSSnapshotter = true
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_RBD_SNAPSHOTTER", "true"), "false") {
|
||||
tp.EnableRBDSnapshotter = false
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_CEPHFS_SNAPSHOTTER", "true"), "false") {
|
||||
tp.EnableCephFSSnapshotter = false
|
||||
}
|
||||
|
||||
tp.EnableVolumeReplicationSideCar = false
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_VOLUME_REPLICATION", "false"), "true") {
|
||||
tp.EnableVolumeReplicationSideCar = true
|
||||
}
|
||||
|
||||
tp.EnableCSIAddonsSideCar = false
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_CSIADDONS", "false"), "true") {
|
||||
tp.EnableCSIAddonsSideCar = true
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_ENCRYPTION", "false"), "true") {
|
||||
tp.EnableCSIEncryption = true
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_CEPHFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
|
||||
tp.CephFSPluginUpdateStrategy = onDelete
|
||||
} else {
|
||||
tp.CephFSPluginUpdateStrategy = rollingUpdate
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_NFS_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
|
||||
tp.NFSPluginUpdateStrategy = onDelete
|
||||
} else {
|
||||
tp.NFSPluginUpdateStrategy = rollingUpdate
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_RBD_PLUGIN_UPDATE_STRATEGY", rollingUpdate), onDelete) {
|
||||
tp.RBDPluginUpdateStrategy = onDelete
|
||||
} else {
|
||||
tp.RBDPluginUpdateStrategy = rollingUpdate
|
||||
}
|
||||
|
||||
if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_PLUGIN_ENABLE_SELINUX_HOST_MOUNT", "false"), "true") {
|
||||
tp.EnablePluginSelinuxHostMount = true
|
||||
}
|
||||
|
||||
logger.Infof("Kubernetes version is %s.%s", ver.Major, ver.Minor)
|
||||
|
||||
tp.ResizerImage = k8sutil.GetValue(r.opConfig.Parameters, "ROOK_CSI_RESIZER_IMAGE", DefaultResizerImage)
|
||||
|
||||
logLevel := k8sutil.GetValue(r.opConfig.Parameters, "CSI_LOG_LEVEL", "")
|
||||
tp.LogLevel = defaultLogLevel
|
||||
if logLevel != "" {
|
||||
l, err := strconv.ParseUint(logLevel, 10, 8)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to parse CSI_LOG_LEVEL. Defaulting to %d. %v", defaultLogLevel, err)
|
||||
} else {
|
||||
tp.LogLevel = uint8(l)
|
||||
}
|
||||
}
|
||||
|
||||
if CustomCSICephConfigExists {
|
||||
tp.MountCustomCephConf = v.SupportsCustomCephConf()
|
||||
}
|
||||
tp.ProvisionerReplicas = defaultProvisionerReplicas
|
||||
nodes, err := r.context.Clientset.CoreV1().Nodes().List(r.opManagerContext, metav1.ListOptions{})
|
||||
if err == nil {
|
||||
if len(nodes.Items) == 1 {
|
||||
tp.ProvisionerReplicas = 1
|
||||
} else {
|
||||
replicas := k8sutil.GetValue(r.opConfig.Parameters, "CSI_PROVISIONER_REPLICAS", "2")
|
||||
r, err := strconv.ParseInt(replicas, 10, 32)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to parse CSI_PROVISIONER_REPLICAS. Defaulting to %d. %v", defaultProvisionerReplicas, err)
|
||||
} else {
|
||||
tp.ProvisionerReplicas = int32(r)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
logger.Errorf("failed to get nodes. Defaulting the number of replicas of provisioner pods to %d. %v", tp.ProvisionerReplicas, err)
|
||||
}
|
||||
|
||||
if EnableRBD {
|
||||
rbdPlugin, err = templateToDaemonSet("rbdplugin", RBDPluginTemplatePath, tp)
|
||||
if err != nil {
|
||||
|
||||
Reference in new issue
Block a user