forked from rook/rook
284 lines
12 KiB
Go
284 lines
12 KiB
Go
/*
|
|
Copyright 2020 The Rook Authors. All rights reserved.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
package controller
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"reflect"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
|
|
"github.com/rook/rook/pkg/operator/k8sutil"
|
|
"github.com/rook/rook/pkg/util/exec"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
|
"sigs.k8s.io/controller-runtime/pkg/reconcile"
|
|
)
|
|
|
|
// OperatorConfig represents the configuration of the operator
|
|
type OperatorConfig struct {
|
|
OperatorNamespace string
|
|
Image string
|
|
ServiceAccount string
|
|
NamespaceToWatch string
|
|
}
|
|
|
|
// ClusterHealth is passed to the various monitoring go routines to stop them when the context is cancelled
|
|
type ClusterHealth struct {
|
|
InternalCtx context.Context
|
|
InternalCancel context.CancelFunc
|
|
}
|
|
|
|
const (
|
|
// OperatorSettingConfigMapName refers to ConfigMap that configures rook ceph operator
|
|
OperatorSettingConfigMapName string = "rook-ceph-operator-config"
|
|
enforceHostNetworkSettingName string = "ROOK_ENFORCE_HOST_NETWORK"
|
|
enforceHostNetworkDefaultValue string = "false"
|
|
|
|
obcAllowAdditionalConfigFieldsSettingName string = "ROOK_OBC_ALLOW_ADDITIONAL_CONFIG_FIELDS"
|
|
obcAllowAdditionalConfigFieldsDefaultValue string = "maxObjects,maxSize"
|
|
|
|
revisionHistoryLimitSettingName string = "ROOK_REVISION_HISTORY_LIMIT"
|
|
|
|
// UninitializedCephConfigError refers to the error message printed by the Ceph CLI when there is no ceph configuration file
|
|
// This typically is raised when the operator has not finished initializing
|
|
UninitializedCephConfigError = "error calling conf_read_file"
|
|
|
|
// OperatorNotInitializedMessage is the message we print when the Operator is not ready to reconcile, typically the ceph.conf has not been generated yet
|
|
OperatorNotInitializedMessage = "skipping reconcile since operator is still initializing"
|
|
)
|
|
|
|
var (
|
|
// ImmediateRetryResult Return this for a immediate retry of the reconciliation loop with the same request object.
|
|
ImmediateRetryResult = reconcile.Result{Requeue: true}
|
|
|
|
// ImmediateRetryResultNoBackoff Return this for a immediate retry of the reconciliation loop with the same request object.
|
|
// Override the exponential backoff behavior by setting the RequeueAfter time explicitly.
|
|
ImmediateRetryResultNoBackoff = reconcile.Result{Requeue: true, RequeueAfter: time.Second}
|
|
|
|
// WaitForRequeueIfCephClusterNotReady waits for the CephCluster to be ready
|
|
WaitForRequeueIfCephClusterNotReady = reconcile.Result{Requeue: true, RequeueAfter: 10 * time.Second}
|
|
|
|
// WaitForRequeueIfCephClusterIsUpgrading waits until the upgrade is complete
|
|
WaitForRequeueIfCephClusterIsUpgrading = reconcile.Result{Requeue: true, RequeueAfter: time.Minute}
|
|
|
|
// WaitForRequeueIfFinalizerBlocked waits for resources to be cleaned up before the finalizer can be removed
|
|
WaitForRequeueIfFinalizerBlocked = reconcile.Result{Requeue: true, RequeueAfter: 10 * time.Second}
|
|
|
|
// WaitForRequeueIfOperatorNotInitialized waits for resources to be cleaned up before the finalizer can be removed
|
|
WaitForRequeueIfOperatorNotInitialized = reconcile.Result{Requeue: true, RequeueAfter: 10 * time.Second}
|
|
|
|
// OperatorCephBaseImageVersion is the ceph version in the operator image
|
|
OperatorCephBaseImageVersion string
|
|
|
|
// loopDevicesAllowed indicates whether loop devices are allowed to be used
|
|
loopDevicesAllowed = false
|
|
revisionHistoryLimit *int32 = nil
|
|
|
|
// allowed OBC additional config fields
|
|
obcAllowAdditionalConfigFields = strings.Split(obcAllowAdditionalConfigFieldsDefaultValue, ",")
|
|
)
|
|
|
|
func DiscoveryDaemonEnabled() bool {
|
|
return k8sutil.GetOperatorSetting("ROOK_ENABLE_DISCOVERY_DAEMON", "false") == "true"
|
|
}
|
|
|
|
// SetCephCommandsTimeout sets the timeout value of Ceph commands which are executed from Rook
|
|
func SetCephCommandsTimeout() {
|
|
strTimeoutSeconds := k8sutil.GetOperatorSetting("ROOK_CEPH_COMMANDS_TIMEOUT_SECONDS", "15")
|
|
timeoutSeconds, err := strconv.Atoi(strTimeoutSeconds)
|
|
if err != nil || timeoutSeconds < 1 {
|
|
logger.Warningf("ROOK_CEPH_COMMANDS_TIMEOUT is %q but it should be >= 1, set the default value 15", strTimeoutSeconds)
|
|
timeoutSeconds = 15
|
|
}
|
|
exec.CephCommandsTimeout = time.Duration(timeoutSeconds) * time.Second
|
|
}
|
|
|
|
func SetAllowLoopDevices() {
|
|
strLoopDevicesAllowed := k8sutil.GetOperatorSetting("ROOK_CEPH_ALLOW_LOOP_DEVICES", "false")
|
|
var err error
|
|
loopDevicesAllowed, err = strconv.ParseBool(strLoopDevicesAllowed)
|
|
if err != nil {
|
|
logger.Warningf("ROOK_CEPH_ALLOW_LOOP_DEVICES is set to an invalid value %v, set the default value false", strLoopDevicesAllowed)
|
|
loopDevicesAllowed = false
|
|
}
|
|
}
|
|
|
|
func LoopDevicesAllowed() bool {
|
|
return loopDevicesAllowed
|
|
}
|
|
|
|
func SetEnforceHostNetwork() {
|
|
strval := k8sutil.GetOperatorSetting(enforceHostNetworkSettingName, enforceHostNetworkDefaultValue)
|
|
val, err := strconv.ParseBool(strval)
|
|
if err != nil {
|
|
logger.Warningf("failed to parse value %q for %q. assuming false value", strval, enforceHostNetworkSettingName)
|
|
cephv1.SetEnforceHostNetwork(false)
|
|
return
|
|
}
|
|
cephv1.SetEnforceHostNetwork(val)
|
|
}
|
|
|
|
func EnforceHostNetwork() bool {
|
|
return cephv1.EnforceHostNetwork()
|
|
}
|
|
|
|
func SetRevisionHistoryLimit() {
|
|
strval := k8sutil.GetOperatorSetting(revisionHistoryLimitSettingName, "")
|
|
var limit int32
|
|
if strval == "" {
|
|
logger.Debugf("not parsing empty string to int for %q. assuming default value.", revisionHistoryLimitSettingName)
|
|
revisionHistoryLimit = nil
|
|
return
|
|
}
|
|
numval, err := strconv.ParseInt(strval, 10, 32)
|
|
if err != nil {
|
|
logger.Warningf("failed to parse value %q for %q. assuming default value. %v", strval, revisionHistoryLimitSettingName, err)
|
|
revisionHistoryLimit = nil
|
|
return
|
|
}
|
|
limit = int32(numval)
|
|
revisionHistoryLimit = &limit
|
|
}
|
|
|
|
func RevisionHistoryLimit() *int32 {
|
|
return revisionHistoryLimit
|
|
}
|
|
|
|
func SetObcAllowAdditionalConfigFields() {
|
|
strval := k8sutil.GetOperatorSetting(obcAllowAdditionalConfigFieldsSettingName, obcAllowAdditionalConfigFieldsDefaultValue)
|
|
obcAllowAdditionalConfigFields = strings.Split(strval, ",")
|
|
}
|
|
|
|
func ObcAdditionalConfigKeyIsAllowed(configField string) bool {
|
|
return slices.Contains(obcAllowAdditionalConfigFields, configField)
|
|
}
|
|
|
|
// canIgnoreHealthErrStatusInReconcile determines whether a status of HEALTH_ERR in the CephCluster can be ignored safely.
|
|
func canIgnoreHealthErrStatusInReconcile(cephCluster cephv1.CephCluster, controllerName string) bool {
|
|
// Get a list of all the keys causing the HEALTH_ERR status.
|
|
healthErrKeys := make([]string, 0)
|
|
for key, health := range cephCluster.Status.CephStatus.Details {
|
|
if health.Severity == "HEALTH_ERR" {
|
|
healthErrKeys = append(healthErrKeys, key)
|
|
}
|
|
}
|
|
|
|
// If there are no errors, the caller actually expects false to be returned so the absence
|
|
// of an error doesn't cause the health status to be ignored. In production, if there are no
|
|
// errors, we would anyway expect the health status to be ok or warning. False in this case
|
|
// will cover if the health status is blank.
|
|
if len(healthErrKeys) == 0 {
|
|
return false
|
|
}
|
|
|
|
allowedErrStatus := map[string]struct{}{
|
|
"MDS_ALL_DOWN": {},
|
|
"MGR_MODULE_ERROR": {},
|
|
}
|
|
allCanBeIgnored := true
|
|
for _, healthErrKey := range healthErrKeys {
|
|
if _, ok := allowedErrStatus[healthErrKey]; !ok {
|
|
allCanBeIgnored = false
|
|
break
|
|
}
|
|
}
|
|
if allCanBeIgnored {
|
|
logger.Debugf("%q: ignoring ceph error status (full status is %+v)", controllerName, cephCluster.Status.CephStatus)
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// IsReadyToReconcile determines if a controller is ready to reconcile or not
|
|
func IsReadyToReconcile(ctx context.Context, c client.Client, namespacedName types.NamespacedName, controllerName string) (cephv1.CephCluster, bool, bool, reconcile.Result) {
|
|
cephClusterExists := false
|
|
|
|
// Running ceph commands won't work and the controller will keep re-queuing so I believe it's fine not to check
|
|
// Make sure a CephCluster exists before doing anything
|
|
var cephCluster cephv1.CephCluster
|
|
clusterList := &cephv1.CephClusterList{}
|
|
err := c.List(ctx, clusterList, client.InNamespace(namespacedName.Namespace))
|
|
if err != nil {
|
|
logger.Errorf("%q: failed to fetch CephCluster %v", controllerName, err)
|
|
return cephCluster, false, cephClusterExists, ImmediateRetryResult
|
|
}
|
|
if len(clusterList.Items) == 0 {
|
|
logger.Debugf("%q: no CephCluster resource found in namespace %q", controllerName, namespacedName.Namespace)
|
|
return cephCluster, false, cephClusterExists, WaitForRequeueIfCephClusterNotReady
|
|
}
|
|
cephCluster = clusterList.Items[0]
|
|
// If the cluster has a cleanup policy to destroy the cluster and it has been marked for deletion, treat it as if it does not exist
|
|
if cephCluster.Spec.CleanupPolicy.HasDataDirCleanPolicy() && !cephCluster.DeletionTimestamp.IsZero() {
|
|
logger.Infof("%q: CephCluster has a destructive cleanup policy, allowing %q to be deleted", controllerName, namespacedName)
|
|
return cephCluster, false, cephClusterExists, WaitForRequeueIfCephClusterNotReady
|
|
}
|
|
|
|
cephClusterExists = true
|
|
logger.Debugf("%q: CephCluster resource %q found in namespace %q", controllerName, cephCluster.Name, namespacedName.Namespace)
|
|
|
|
// read the CR status of the cluster
|
|
if cephCluster.Status.CephStatus != nil {
|
|
operatorDeploymentOk := cephCluster.Status.CephStatus.Health == "HEALTH_OK" || cephCluster.Status.CephStatus.Health == "HEALTH_WARN"
|
|
|
|
if operatorDeploymentOk || canIgnoreHealthErrStatusInReconcile(cephCluster, controllerName) {
|
|
logger.Debugf("%q: ceph status is %q, operator is ready to run ceph command, reconciling", controllerName, cephCluster.Status.CephStatus.Health)
|
|
return cephCluster, true, cephClusterExists, WaitForRequeueIfCephClusterNotReady
|
|
}
|
|
|
|
details := cephCluster.Status.CephStatus.Details
|
|
message, ok := details["error"]
|
|
if ok && len(details) == 1 && strings.Contains(message.Message, "Error initializing cluster client") {
|
|
logger.Infof("%s: skipping reconcile since operator is still initializing", controllerName)
|
|
} else {
|
|
logger.Infof("%s: CephCluster %q found but skipping reconcile since ceph health is %+v", controllerName, cephCluster.Name, cephCluster.Status.CephStatus)
|
|
}
|
|
}
|
|
|
|
logger.Debugf("%q: CephCluster %q initial reconcile is not complete yet...", controllerName, namespacedName)
|
|
return cephCluster, false, cephClusterExists, WaitForRequeueIfCephClusterNotReady
|
|
}
|
|
|
|
// ClusterOwnerRef represents the owner reference of the CephCluster CR
|
|
func ClusterOwnerRef(clusterName, clusterID string) metav1.OwnerReference {
|
|
blockOwner := true
|
|
controller := true
|
|
return metav1.OwnerReference{
|
|
APIVersion: fmt.Sprintf("%s/%s", ClusterResource.Group, ClusterResource.Version),
|
|
Kind: ClusterResource.Kind,
|
|
Name: clusterName,
|
|
UID: types.UID(clusterID),
|
|
BlockOwnerDeletion: &blockOwner,
|
|
Controller: &controller,
|
|
}
|
|
}
|
|
|
|
// ClusterResource operator-kit Custom Resource Definition
|
|
var ClusterResource = k8sutil.CustomResource{
|
|
Name: "cephcluster",
|
|
Plural: "cephclusters",
|
|
Group: cephv1.CustomResourceGroup,
|
|
Version: cephv1.Version,
|
|
Kind: reflect.TypeOf(cephv1.CephCluster{}).Name(),
|
|
APIVersion: fmt.Sprintf("%s/%s", cephv1.CustomResourceGroup, cephv1.Version),
|
|
}
|