my-rook-config/pkg/operator/ceph/controller/controller_utils.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"
	"runtime/debug"
	"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"
	"github.com/rook/rook/pkg/util/log"
	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}

	// 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 {
		log.NamedError(namespacedName, logger, "%q: failed to fetch CephCluster %v", controllerName, err)
		return cephCluster, false, cephClusterExists, ImmediateRetryResult
	}
	if len(clusterList.Items) == 0 {
		log.NamedDebug(namespacedName, logger, "%q: no CephCluster resource found in namespace", controllerName)
		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() {
		log.NamedInfo(namespacedName, logger, "%q: CephCluster has a destructive cleanup policy, allowing it to be deleted", controllerName)
		return cephCluster, false, cephClusterExists, WaitForRequeueIfCephClusterNotReady
	}

	cephClusterExists = true
	log.NamedDebug(namespacedName, logger, "%q: CephCluster resource found", controllerName)

	// 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) {
			log.NamedDebug(namespacedName, logger, "%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") {
			log.NamedInfo(namespacedName, logger, "%q: skipping reconcile since operator is still initializing", controllerName)
		} else {
			log.NamedInfo(namespacedName, logger, "%q: CephCluster %q found but skipping reconcile since ceph health is %+v", controllerName, cephCluster.Name, cephCluster.Status.CephStatus)
		}
	}

	log.NamedDebug(namespacedName, logger, "%q: CephCluster initial reconcile is not complete yet...", controllerName)
	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.TypeFor[cephv1.CephCluster]().Name(),
	APIVersion: fmt.Sprintf("%s/%s", cephv1.CustomResourceGroup, cephv1.Version),
}

// RecoverAndLogException handles and logs panics from a controller Reconcile loop.
func RecoverAndLogException() {
	if r := recover(); r != nil {
		logger.Errorf("Panic: %v", r)
		logger.Errorf("Stack trace:\n%s", string(debug.Stack()))
	}
}

func NsName(namespace, name string) types.NamespacedName {
	return types.NamespacedName{Namespace: namespace, Name: name}
}