/*
Copyright 2024.

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 (
	"cmp"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"reflect"
	"regexp"
	"slices"
	"strconv"
	"strings"

	"github.com/go-logr/logr"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	storagev1 "k8s.io/api/storage/v1"
	k8serrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/utils/ptr"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	ctrlutil "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/event"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"

	csiv1 "github.com/ceph/ceph-csi-operator/api/v1"
	"github.com/ceph/ceph-csi-operator/internal/utils"
	sm "github.com/kubernetes-csi/external-snapshot-metadata/client/apis/snapshotmetadataservice/v1alpha1"
)

//+kubebuilder:rbac:groups=csi.ceph.io,resources=drivers,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=csi.ceph.io,resources=drivers/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=csi.ceph.io,resources=drivers/finalizers,verbs=update
//+kubebuilder:rbac:groups=csi.ceph.io,resources=operatorconfigs,verbs=get;list;watch
//+kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch;create;update
//+kubebuilder:rbac:groups=storage.k8s.io,resources=csidrivers,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=apps,resources=daemonsets,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups="cbt.storage.k8s.io",resources=snapshotmetadataservices,verbs=get;list;watch;

type DriverType string

const (
	RbdDriverType    = "rbd"
	CephFsDriverType = "cephfs"
	NfsDriverType    = "nfs"
)

const (
	// Annotation name for ownerref information
	ownerRefAnnotationKey = "csi.ceph.io/ownerref"
	// Annotation to enable CSI-Addons volume condition reporter
	driverCSIAddonsFeatureVolumeCondition = "addons.csi.ceph.io/volume-condition"

	logRotateCmd = `while true; do logrotate --verbose /logrotate-config/csi; sleep 15m; done`
)

// A regexp used to parse driver's prefix and type from the full name
var nameRegExp, _ = regexp.Compile(fmt.Sprintf(
	`^(?:.+\.)?(%s|%s|%s)\.csi\.ceph\.com$`,
	RbdDriverType,
	CephFsDriverType,
	NfsDriverType,
))

// DriverReconciler reconciles a Driver object
type DriverReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

// A local reconcile object tied to a single reconcile iteration
type driverReconcile struct {
	DriverReconciler

	ctx        context.Context
	log        logr.Logger
	driver     csiv1.Driver
	driverType DriverType
	images     map[string]string
}

// SetupWithManager sets up the controller with the Manager.
func (r *DriverReconciler) SetupWithManager(mgr ctrl.Manager) error {
	// Define conditions for an OperatorConfig change that the require queuing of reconciliation
	// Filter update events based on metadata.generation changes, will filter events
	// for non-spec changes on most resource types.
	genChangedPredicate := predicate.GenerationChangedPredicate{}

	// request for drivers
	driverDefaultsPredicate := predicate.Funcs{
		CreateFunc: func(e event.CreateEvent) bool {
			opConf, ok := e.Object.(*csiv1.OperatorConfig)
			return ok && opConf.Spec.DriverSpecDefaults != nil
		},
		UpdateFunc: func(e event.UpdateEvent) bool {
			oldConf, oldOk := e.ObjectOld.(*csiv1.OperatorConfig)
			newConf, newOk := e.ObjectNew.(*csiv1.OperatorConfig)
			return !oldOk || !newOk ||
				!reflect.DeepEqual(oldConf.Spec.DriverSpecDefaults, newConf.Spec.DriverSpecDefaults)
		},
		DeleteFunc: func(e event.DeleteEvent) bool {
			opConf, ok := e.Object.(*csiv1.OperatorConfig)
			return ok && opConf.Spec.DriverSpecDefaults != nil
		},
		GenericFunc: func(event.GenericEvent) bool {
			return false
		},
	}

	// Enqueue a reconcile request for all existing drivers, used to trigger a reconcile
	// for all drivers whenever the driver default configuration changes
	enqueueAllDrivers := handler.EnqueueRequestsFromMapFunc(
		func(ctx context.Context, obj client.Object) []reconcile.Request {
			driverList := csiv1.DriverList{}
			if err := r.List(ctx, &driverList); err != nil {
				return []reconcile.Request{}
			}

			requests := make([]reconcile.Request, len(driverList.Items))
			for i := range driverList.Items {
				requests[i].NamespacedName = client.ObjectKeyFromObject(&driverList.Items[i])
			}
			return requests
		},
	)

	// Enqueue a reconcile request based on an annotation marking a soft ownership
	enqueueFromOwnerRefAnnotation := handler.EnqueueRequestsFromMapFunc(
		func(_ context.Context, obj client.Object) []reconcile.Request {
			ownerRef := obj.GetAnnotations()[ownerRefAnnotationKey]
			if ownerRef == "" {
				return nil
			}

			ownerObjKey := client.ObjectKey{}
			if err := json.Unmarshal([]byte(ownerRef), &ownerObjKey); err != nil {
				return nil
			}

			return []reconcile.Request{{
				NamespacedName: ownerObjKey,
			}}
		},
	)

	return ctrl.NewControllerManagedBy(mgr).
		For(&csiv1.Driver{}).
		Owns(
			&appsv1.Deployment{},
			builder.WithPredicates(genChangedPredicate),
		).
		Owns(
			&appsv1.DaemonSet{},
			builder.WithPredicates(genChangedPredicate),
		).
		Owns(
			&corev1.Service{},
			builder.WithPredicates(genChangedPredicate),
		).
		Owns(
			&corev1.ConfigMap{},
			builder.MatchEveryOwner,
			builder.WithPredicates(
				utils.NamePredicate(utils.CsiConfigVolume.Name),
				utils.EventTypePredicate(false, false, true, false),
			),
		).
		Watches(
			&csiv1.OperatorConfig{},
			enqueueAllDrivers,
			builder.WithPredicates(driverDefaultsPredicate),
		).
		Watches(
			&storagev1.CSIDriver{},
			enqueueFromOwnerRefAnnotation,
		).
		Complete(r)
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.17.3/pkg/reconcile
func (r *DriverReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := ctrllog.FromContext(ctx)
	log.Info("Starting reconcile iteration for Ceph CSI driver", "req", req)

	reconcileHandler := driverReconcile{}
	reconcileHandler.DriverReconciler = *r
	reconcileHandler.ctx = ctx
	reconcileHandler.log = log
	reconcileHandler.driver.Name = req.Name
	reconcileHandler.driver.Namespace = req.Namespace

	err := reconcileHandler.reconcile()
	if err != nil {
		log.Error(err, "CSI Driver reconciliation failed")
	} else {
		log.Info("CSI Driver reconciliation completed successfully")
	}
	return ctrl.Result{}, err
}

func (r *driverReconcile) hasSnapshotMetadataServiceCr() (bool, error) {
	sms := &sm.SnapshotMetadataService{}
	sms.Name = r.driver.Name
	if err := r.Get(r.ctx, client.ObjectKeyFromObject(sms), sms); !meta.IsNoMatchError(err) && client.IgnoreNotFound(err) != nil {
		return false, fmt.Errorf("failed to get SnapshotMetadataService resource: %w", err)
	}

	return sms.UID != "", nil
}

func (r *driverReconcile) reconcile() error {
	// Load the driver desired state based on driver resource, operator config resource and default values.
	if err := r.LoadAndValidateDesiredState(); err != nil {
		return err
	}

	reconcilers := []func() error{
		r.reconcileCsiConfigMap,
		r.reconcileLogRotateConfigMap,
		r.reconcileK8sCsiDriver,
		r.reconcileControllerPluginDeployment,
		r.reconcileNodePluginDaemonSet,
		r.reconcileLivenessService,
		r.reconcileNodePluginDaemonSetForCsiAddons,
	}

	// Concurrently reconcile different aspects of the clusters actual state to meet
	// the desired state defined on the driver object
	errChan := utils.RunConcurrently(reconcilers...)

	// Check if any reconcilatin error where raised during the concurrent execution
	// of the reconciliation steps.
	errList := utils.ChannelToSlice(errChan)
	if err := errors.Join(errList...); err != nil {
		return err
	}

	return nil
}

func (r *driverReconcile) LoadAndValidateDesiredState() error {
	// Validate that the requested name for the CSI driver isn't already claimed by an existing CSI driver
	// (Can happen if a driver with an identical name was created in a different namespace)
	csiDriver := storagev1.CSIDriver{}
	csiDriver.Name = r.driver.Name
	if err := r.Get(r.ctx, client.ObjectKeyFromObject(&csiDriver), &csiDriver); client.IgnoreNotFound(err) != nil {
		r.log.Error(err, "Failed to query the existence of a CSI Driver")
		return err
	}
	if csiDriver.UID != "" {
		// If the k8s csi driver the we found does not have an owner ref annotation
		// we assume someone else is owning it
		ownerRef := csiDriver.GetAnnotations()[ownerRefAnnotationKey]
		if ownerRef == "" {
			err := fmt.Errorf("invalid driver name")
			r.log.Error(err, "Desired name already in use by a different CSI Driver")
			return err
		}

		ownerObjKey := client.ObjectKey{}
		if err := json.Unmarshal([]byte(ownerRef), &ownerObjKey); err != nil {
			r.log.Error(err, "Failed to parse owner ref annotation on CSI Driver")
			return err
		}

		// If the owner annotation does not correlate with the driver we are reconciling
		// we assume there is another driver CR with an identical name on some other namespace
		if r.driver.Namespace != ownerObjKey.Namespace || r.driver.Name != ownerObjKey.Name {
			err := fmt.Errorf("invalid driver name")
			r.log.Error(err, "Desired name already in use by a different CSI Driver", "current owner", ownerObjKey)
			return err
		}
	}

	// Extract the driver sort name and driver type
	matches := nameRegExp.FindStringSubmatch(r.driver.Name)
	if len(matches) != 2 {
		return fmt.Errorf("invalid driver name %s", r.driver.Name)
	}

	r.driverType = DriverType(strings.ToLower(matches[1]))

	// Load operator configuration resource
	opConfig := csiv1.OperatorConfig{}
	opConfig.Name = operatorConfigName
	opConfig.Namespace = operatorNamespace
	if err := r.Get(r.ctx, client.ObjectKeyFromObject(&opConfig), &opConfig); client.IgnoreNotFound(err) != nil {
		r.log.Error(err, "Unable to load operatorconfig.csi.ceph.io", "name", client.ObjectKeyFromObject(&opConfig))
		return err
	}

	// Load the current desired state in the form of a ceph csi driver resource
	if err := r.Get(r.ctx, client.ObjectKeyFromObject(&r.driver), &r.driver); err != nil {
		r.log.Error(err, "Unable to load driver.csi.ceph.io", "name", client.ObjectKeyFromObject(&r.driver))
		return err
	}

	// Cloning default images as the base images before
	// merging any user provided images
	r.images = maps.Clone(imageDefaults)

	if opConfig.Spec.DriverSpecDefaults != nil {
		mergeDriverSpecs(&r.driver.Spec, opConfig.Spec.DriverSpecDefaults)

		// If provided, load an imageset from configmap to overwrite default images
		imageSetSpec := opConfig.Spec.DriverSpecDefaults.ImageSet
		if imageSetSpec != nil && imageSetSpec.Name != "" {
			imageSetCM := corev1.ConfigMap{}
			imageSetCM.Name = imageSetSpec.Name
			imageSetCM.Namespace = operatorNamespace
			if err := r.Get(r.ctx, client.ObjectKeyFromObject(&imageSetCM), &imageSetCM); err != nil {
				r.log.Error(
					err,
					"Unable to load operator config specified image set config map",
					"name",
					client.ObjectKeyFromObject(&imageSetCM),
				)
				return err
			}
			maps.Copy(r.images, imageSetCM.Data)
		}
	}

	// If provided, load an imageset from driver spec overwrite default images
	imageSetSpec := r.driver.Spec.ImageSet
	if imageSetSpec != nil && imageSetSpec.Name != "" {
		imageSetCM := corev1.ConfigMap{}
		imageSetCM.Name = imageSetSpec.Name
		imageSetCM.Namespace = r.driver.Namespace
		if err := r.Get(r.ctx, client.ObjectKeyFromObject(&imageSetCM), &imageSetCM); err != nil {
			r.log.Error(
				err,
				"Unable to load driver specified image set config map",
				"name",
				client.ObjectKeyFromObject(&imageSetCM),
			)
			return err
		}
		maps.Copy(r.images, imageSetCM.Data)
	}

	return nil
}

func (r *driverReconcile) reconcileLogRotateConfigMap() error {
	logRotateConfigmap := &corev1.ConfigMap{}
	logRotateConfigmap.Name = utils.LogRotateConfigMapName(r.driver.Name)
	logRotateConfigmap.Namespace = r.driver.Namespace

	log := r.log.WithValues("logRotateConfigMap", logRotateConfigmap.Name)
	log.Info("Reconciling logRotate configmap")

	logRotationSpec := cmp.Or(r.driver.Spec.Log, &csiv1.LogSpec{}).Rotation
	if logRotationSpec != nil {
		opResult, err := ctrlutil.CreateOrUpdate(r.ctx, r.Client, logRotateConfigmap, func() error {
			if _, err := utils.ToggleOwnerReference(true, logRotateConfigmap, &r.driver, r.Scheme); err != nil {
				log.Error(err, "Failed adding an owner reference on the LogRotate config map")
				return err
			}
			if logRotationSpec.Periodicity != "" && logRotationSpec.MaxLogSize.IsZero() {
				err := fmt.Errorf("invalid Log.Rotation spec")
				log.Error(err, "Either \"maxLogSize\" or \"periodicity\" fields must be set")
				return err
			}
			maxFiles := cmp.Or(logRotationSpec.MaxFiles, defaultLogRotateMaxFiles)
			cronLogRotateSettings := []string{
				"\tmissingok",
				"\tcompress",
				"\tcopytruncate",
				"\tnotifempty",
				fmt.Sprintf("\trotate %d", maxFiles),
			}
			if logRotationSpec.Periodicity != "" {
				periodicity := "\t" + string(logRotationSpec.Periodicity)
				cronLogRotateSettings = append(cronLogRotateSettings, periodicity)
			}
			if logRotationSpec.MaxLogSize.String() != "0" {
				maxSize := fmt.Sprintf("\tmaxsize %s", logRotationSpec.MaxLogSize.String())
				cronLogRotateSettings = append(cronLogRotateSettings, maxSize)
			}
			logRotateConfigmap.Data = map[string]string{
				"csi": fmt.Sprintf(
					"/csi-logs/*.log {\n%s\n}\n",
					strings.Join(cronLogRotateSettings, "\n"),
				),
			}
			return nil
		})

		logCreateOrUpdateResult(log, "LogRotateConfigMap", logRotateConfigmap, opResult, err)
		return err
	} else {
		// Remove the logrotate configmap if logrotate setting is removed from the driver's spec
		if err := r.Delete(r.ctx, logRotateConfigmap); client.IgnoreNotFound(err) != nil {
			log.Error(err, "Unable to delete LogRotate configmap")
			return err
		}
		return nil
	}
}

func (r *driverReconcile) reconcileCsiConfigMap() error {
	csiConfigMap := &corev1.ConfigMap{}
	csiConfigMap.Name = utils.CsiConfigVolume.Name
	csiConfigMap.Namespace = r.driver.Namespace

	log := r.log.WithValues("csiConfigMap", csiConfigMap.Name)
	log.Info("Reconciling CSI Config Map")

	opResult, err := ctrlutil.CreateOrUpdate(r.ctx, r.Client, csiConfigMap, func() error {
		if _, err := utils.ToggleOwnerReference(true, csiConfigMap, &r.driver, r.Scheme); err != nil {
			log.Error(err, "Failed adding an owner referce on Ceph CSI config map")
			return err
		}
		return nil
	})

	logCreateOrUpdateResult(log, "CSI Config Map", csiConfigMap, opResult, err)
	return err
}

func (r *driverReconcile) reconcileK8sCsiDriver() error {
	csiDriver := &storagev1.CSIDriver{}
	csiDriver.Name = r.driver.Name

	log := r.log.WithValues("driverName", csiDriver.Name)
	log.Info("Reconciling CSI Driver")

	opResult, err := ctrlutil.CreateOrUpdate(r.ctx, r.Client, csiDriver, func() error {
		ownerObjKey := client.ObjectKeyFromObject(&r.driver)
		if bytes, err := json.Marshal(ownerObjKey); err == nil {
			if utils.AddAnnotation(csiDriver, ownerRefAnnotationKey, string(bytes)) {
				log.Info("ownerref annotation added to CSI driver resource")
			}
		} else {
			log.Error(
				err,
				"Failed to JSON marshal owner obj key for CSI driver resource",
				"ownerObjKey",
				ownerObjKey,
			)
			return err
		}

		csiDriver.Spec.PodInfoOnMount = ptr.To(false)
		csiDriver.Spec.AttachRequired = cmp.Or(
			r.driver.Spec.AttachRequired,
			ptr.To(true),
		)
		csiDriver.Spec.FSGroupPolicy = ptr.To(
			cmp.Or(
				r.driver.Spec.FsGroupPolicy,
				storagev1.FileFSGroupPolicy,
			),
		)
		csiDriver.Spec.SELinuxMount = ptr.To(true)

		return nil
	})

	// We are expecting an Invalid operation error, on an existing CSIDriver, in the rear case
	// where the new desired state require reconfiguration of an immutable field.
	// For CSIDriver, ".spec.attachRequired" is an invalid field
	if csiDriver.UID != "" && k8serrors.IsInvalid(err) {
		r.log.Info("CSIDriver exists but cannot be updated, trying recreate instead")

		if err = r.Delete(r.ctx, csiDriver); err != nil {
			r.log.Error(err, "Failed deleting existing CSIDriver")
			return err
		}

		csiDriver = &storagev1.CSIDriver{
			ObjectMeta: metav1.ObjectMeta{
				Name:        csiDriver.Name,
				Labels:      csiDriver.Labels,
				Annotations: csiDriver.Annotations,
			},
			Spec: csiDriver.Spec,
		}
		if err := r.Create(r.ctx, csiDriver); err != nil {
			r.log.Error(err, "Failed recreating CSIDriver")
			return err
		}

		log.Info("CSIDriver recreated successfully")
		return nil

	} else {
		logCreateOrUpdateResult(log, "CSIDriver", csiDriver, opResult, err)
		return err
	}
}

func (r *driverReconcile) reconcileControllerPluginDeployment() error {
	deploy := &appsv1.Deployment{}
	deploy.Name = r.generateName("ctrlplugin")
	deploy.Namespace = r.driver.Namespace

	log := r.log.WithValues("deploymentName", deploy.Name)
	log.Info("Reconciling controller plugin deployment")

	opResult, err := ctrlutil.CreateOrUpdate(r.ctx, r.Client, deploy, func() error {
		if err := ctrlutil.SetControllerReference(&r.driver, deploy, r.Scheme); err != nil {
			log.Error(err, "Failed setting an owner reference on deployment")
			return err
		}

		appName := deploy.Name
		appSelector := metav1.LabelSelector{
			MatchLabels: map[string]string{"app": appName},
		}

		leaderElectionSpec := cmp.Or(r.driver.Spec.LeaderElection, &defaultLeaderElection)
		pluginSpec := cmp.Or(r.driver.Spec.ControllerPlugin, &csiv1.ControllerPluginSpec{})
		serviceAccountName := cmp.Or(
			ptr.Deref(pluginSpec.ServiceAccountName, ""),
			fmt.Sprintf("%s%s-ctrlplugin-sa", serviceAccountPrefix, r.driverType),
		)
		imagePullPolicy := cmp.Or(pluginSpec.ImagePullPolicy, corev1.PullIfNotPresent)
		grpcTimeout := cmp.Or(r.driver.Spec.GRpcTimeout, defaultGRrpcTimeout)
		logVerbosity := ptr.Deref(r.driver.Spec.Log, csiv1.LogSpec{}).Verbosity
		forceKernelClient := r.isCephFsDriver() && r.driver.Spec.CephFsClientType == csiv1.KernelCephFsClient
		snPolicy := cmp.Or(r.driver.Spec.SnapshotPolicy, csiv1.VolumeSnapshotSnapshotPolicy)
		logRotationSpec := cmp.Or(r.driver.Spec.Log, &csiv1.LogSpec{}).Rotation
		logRotationEnabled := logRotationSpec != nil
		logRotateSecurityContext := utils.If(
			pluginSpec.Privileged != nil && logRotationEnabled,
			&corev1.SecurityContext{
				Privileged: pluginSpec.Privileged,
				Capabilities: &corev1.Capabilities{
					Drop: []corev1.Capability{"All"},
				},
			},
			nil,
		)

		leaderElectionSettingsArg := []string{
			utils.LeaderElectionNamespaceContainerArg(r.driver.Namespace),
			utils.LeaderElectionLeaseDurationContainerArg(leaderElectionSpec.LeaseDuration),
			utils.LeaderElectionRenewDeadlineContainerArg(leaderElectionSpec.RenewDeadline),
			utils.LeaderElectionRetryPeriodContainerArg(leaderElectionSpec.RetryPeriod),
		}

		// TODO: Move the Topology field from NodePlugin to Driver.Spec
		nodePluginSpec := cmp.Or(r.driver.Spec.NodePlugin, &csiv1.NodePluginSpec{})
		topology := r.isRbdDriver() && nodePluginSpec.Topology != nil

		deploy.Spec = appsv1.DeploymentSpec{
			Replicas: pluginSpec.Replicas,
			Selector: &appSelector,
			Strategy: ptr.Deref(pluginSpec.DeploymentStrategy, defaultDeploymentStrategy),
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: utils.Call(func() map[string]string {
						podLabels := map[string]string{}
						maps.Copy(podLabels, pluginSpec.Labels)
						podLabels["app"] = appName
						return podLabels
					}),
					Annotations: maps.Clone(pluginSpec.Annotations),
				},
				Spec: corev1.PodSpec{
					ServiceAccountName: serviceAccountName,
					PriorityClassName:  ptr.Deref(pluginSpec.PrioritylClassName, ""),
					HostNetwork:        ptr.Deref(pluginSpec.HostNetwork, false),
					Affinity:           getControllerPluginPodAffinity(pluginSpec, &appSelector),
					Tolerations:        pluginSpec.Tolerations,
					Containers: utils.Call(func() []corev1.Container {
						containers := []corev1.Container{
							// Plugin Container
							{
								Name:            fmt.Sprintf("csi-%splugin", r.driverType),
								Image:           r.images["plugin"],
								ImagePullPolicy: imagePullPolicy,
								SecurityContext: logRotateSecurityContext,
								Args: utils.DeleteZeroValues(
									[]string{
										utils.TypeContainerArg(string(r.driverType)),
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.EndpointContainerArg,
										utils.NodeIdContainerArg,
										utils.ControllerServerContainerArg,
										utils.DriverNameContainerArg(r.driver.Name),
										utils.PidlimitContainerArg,
										utils.SetMetadataContainerArg(ptr.Deref(r.driver.Spec.EnableMetadata, false)),
										utils.SetFencingContainerArg(ptr.Deref(r.driver.Spec.EnableFencing, false)),
										utils.ClusterNameContainerArg(ptr.Deref(r.driver.Spec.ClusterName, "")),
										utils.If(forceKernelClient, utils.ForceCephKernelClientContainerArg, ""),
										utils.If(
											ptr.Deref(r.driver.Spec.DeployCsiAddons, false),
											utils.CsiAddonsEndpointContainerArg,
											"",
										),
										utils.If(logRotationEnabled, utils.LogToStdErrContainerArg, ""),
										utils.If(logRotationEnabled, utils.AlsoLogToStdErrContainerArg, ""),
										utils.If(
											logRotationEnabled,
											utils.LogFileContainerArg(fmt.Sprintf("csi-%splugin", r.driverType)),
											"",
										),
									},
								),
								Env: []corev1.EnvVar{
									utils.PodIpEnvVar,
									utils.NodeIdEnvVar,
									utils.PodNamespaceEnvVar,
								},
								VolumeMounts: utils.Call(func() []corev1.VolumeMount {
									mounts := []corev1.VolumeMount{
										utils.SocketDirVolumeMount,
										utils.HostDevVolumeMount,
										utils.HostSysVolumeMount,
										utils.LibModulesVolumeMount,
										utils.KeysTmpDirVolumeMount,
										utils.CsiConfigVolumeMount,
									}
									if r.driver.Spec.Encryption != nil {
										mounts = append(mounts, utils.KmsConfigVolumeMount)
									}
									if r.isRbdDriver() {
										mounts = append(mounts, utils.OidcTokenVolumeMount)
									}
									if logRotationEnabled {
										mounts = append(mounts, utils.LogsDirVolumeMount)
									}
									// Add user defined volume mounts at the end to make sure they
									// can overwrite built in volumes mounts.
									mounts = utils.MapMergeByKey(
										mounts,
										pluginSpec.Volumes,
										func(v csiv1.VolumeSpec) (corev1.VolumeMount, bool) {
											return v.Mount, v.Mount.Name != ""
										},
										func(v corev1.VolumeMount) string {
											return v.Name
										},
									)

									return mounts
								}),
								Resources: ptr.Deref(
									pluginSpec.Resources.Plugin,
									corev1.ResourceRequirements{},
								),
							},
							// Provisioner Sidecar Container
							{
								Name:            "csi-provisioner",
								ImagePullPolicy: imagePullPolicy,
								Image:           r.images["provisioner"],
								Args: utils.DeleteZeroValues(
									append(
										slices.Clone(leaderElectionSettingsArg),
										utils.LeaderElectionContainerArg,
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.CsiAddressContainerArg,
										utils.TimeoutContainerArg(grpcTimeout),
										utils.RetryIntervalStartContainerArg,
										utils.DefaultFsTypeContainerArg,
										utils.PreventVolumeModeConversionContainerArg,
										utils.HonorPVReclaimPolicyContainerArg,
										utils.If(r.isRbdDriver(), utils.DefaultFsTypeContainerArg, ""),
										utils.TopologyContainerArg(topology),
										utils.If(!r.isNfsDriver(), utils.ExtraCreateMetadataContainerArg, ""),
									),
								),
								VolumeMounts: []corev1.VolumeMount{
									utils.SocketDirVolumeMount,
								},
								Resources: ptr.Deref(
									pluginSpec.Resources.Provisioner,
									corev1.ResourceRequirements{},
								),
							},
							// Resizer Sidecar Container
							{
								Name:            "csi-resizer",
								ImagePullPolicy: imagePullPolicy,
								Image:           r.images["resizer"],
								Args: utils.DeleteZeroValues(
									append(
										slices.Clone(leaderElectionSettingsArg),
										utils.LeaderElectionContainerArg,
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.CsiAddressContainerArg,
										utils.TimeoutContainerArg(grpcTimeout),
										utils.HandleVolumeInuseErrorContainerArg,
										utils.RecoverVolumeExpansionFailureContainerArg,
									),
								),
								VolumeMounts: []corev1.VolumeMount{
									utils.SocketDirVolumeMount,
								},
								Resources: ptr.Deref(
									pluginSpec.Resources.Resizer,
									corev1.ResourceRequirements{},
								),
							},
							// Attacher Sidecar Container
							{
								Name:            "csi-attacher",
								ImagePullPolicy: imagePullPolicy,
								Image:           r.images["attacher"],
								Args: utils.DeleteZeroValues(
									append(
										slices.Clone(leaderElectionSettingsArg),
										utils.LeaderElectionContainerArg,
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.CsiAddressContainerArg,
										utils.TimeoutContainerArg(grpcTimeout),
										utils.If(r.isRbdDriver(), utils.DefaultFsTypeContainerArg, ""),
									),
								),
								VolumeMounts: []corev1.VolumeMount{
									utils.SocketDirVolumeMount,
								},
								Resources: ptr.Deref(
									pluginSpec.Resources.Attacher,
									corev1.ResourceRequirements{},
								),
							},
						}
						// Snapshotter Sidecar Container
						if snPolicy != csiv1.NoneSnapshotPolicy {
							containers = append(containers, corev1.Container{
								Name:            "csi-snapshotter",
								ImagePullPolicy: imagePullPolicy,
								Image:           r.images["snapshotter"],
								Args: utils.DeleteZeroValues(
									append(
										slices.Clone(leaderElectionSettingsArg),
										utils.LeaderElectionContainerArg,
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.CsiAddressContainerArg,
										utils.TimeoutContainerArg(grpcTimeout),
										utils.If(!r.isNfsDriver(), utils.ExtraCreateMetadataContainerArg, ""),
										utils.If(
											r.driverType != NfsDriverType && snPolicy == csiv1.VolumeGroupSnapshotPolicy,
											utils.EnableVolumeGroupSnapshotsContainerArg,
											"",
										),
									),
								),
								VolumeMounts: []corev1.VolumeMount{
									utils.SocketDirVolumeMount,
								},
								Resources: ptr.Deref(
									pluginSpec.Resources.Snapshotter,
									corev1.ResourceRequirements{},
								),
							})
						}
						// Extended Snapshotter Sidecar Container
						if r.images["ex-snapshotter"] != "" && snPolicy != csiv1.NoneSnapshotPolicy && r.driverType == CephFsDriverType {
							containers = append(containers, corev1.Container{
								Name:            "ex-csi-snapshotter",
								ImagePullPolicy: imagePullPolicy,
								Image:           r.images["ex-snapshotter"],
								// This is necessary only for systems with SELinux, where
								// non-privileged sidecar containers cannot access unix domain socket
								// created by privileged CSI driver container.
								SecurityContext: &corev1.SecurityContext{
									Privileged: ptr.To(true),
									Capabilities: &corev1.Capabilities{
										Drop: []corev1.Capability{"All"},
									},
								},
								Args: utils.DeleteZeroValues(
									append(
										slices.Clone(leaderElectionSettingsArg),
										utils.LeaderElectionContainerArg,
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.CsiAddressContainerArg,
										utils.TimeoutContainerArg(grpcTimeout),
										utils.ExtraCreateMetadataContainerArg,
										utils.EnableVolumeGroupSnapshotsContainerArg,
									),
								),
								VolumeMounts: []corev1.VolumeMount{
									utils.SocketDirVolumeMount,
								},
								Resources: ptr.Deref(
									pluginSpec.Resources.Snapshotter,
									corev1.ResourceRequirements{},
								),
							})
						}
						// Addons Sidecar Container
						if !r.isNfsDriver() && ptr.Deref(r.driver.Spec.DeployCsiAddons, false) {
							port := r.controllerPluginCsiAddonsContainerPort()
							containers = append(containers, corev1.Container{
								Name:            "csi-addons",
								Image:           r.images["addons"],
								ImagePullPolicy: imagePullPolicy,
								SecurityContext: logRotateSecurityContext,
								Args: utils.DeleteZeroValues(
									append(
										slices.Clone(leaderElectionSettingsArg),
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.CsiAddonsNodeIdContainerArg,
										utils.PodContainerArg,
										utils.PodUidContainerArg,
										utils.CsiAddonsAddressContainerArg,
										utils.ContainerPortArg(port),
										utils.NamespaceContainerArg,
										utils.If(logRotationEnabled, utils.LogToStdErrContainerArg, ""),
										utils.If(logRotationEnabled, utils.AlsoLogToStdErrContainerArg, ""),
										utils.If(logRotationEnabled, utils.LogFileContainerArg("csi-addons"), ""),
									),
								),
								Ports: []corev1.ContainerPort{
									port,
								},
								Env: []corev1.EnvVar{
									utils.NodeIdEnvVar,
									utils.PodUidEnvVar,
									utils.PodNameEnvVar,
									utils.PodNamespaceEnvVar,
								},
								VolumeMounts: utils.Call(func() []corev1.VolumeMount {
									mounts := []corev1.VolumeMount{
										utils.SocketDirVolumeMount,
									}
									if logRotationEnabled {
										mounts = append(mounts, utils.LogsDirVolumeMount)
									}
									return mounts
								}),
								Resources: ptr.Deref(
									pluginSpec.Resources.Addons,
									corev1.ResourceRequirements{},
								),
							})
						}
						// OMap Generator Sidecar Container
						if r.isRbdDriver() && ptr.Deref(r.driver.Spec.GenerateOMapInfo, false) {
							containers = append(containers, corev1.Container{
								Name:            "csi-omap-generator",
								Image:           r.images["plugin"],
								ImagePullPolicy: imagePullPolicy,
								Args: utils.DeleteZeroValues(
									[]string{
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.TypeContainerArg("controller"),
										utils.DriverNamespaceContainerArg,
										utils.DriverNameContainerArg(r.driver.Name),
										utils.SetMetadataContainerArg(ptr.Deref(r.driver.Spec.EnableMetadata, false)),
										utils.ClusterNameContainerArg(ptr.Deref(r.driver.Spec.ClusterName, "")),
									},
								),
								Env: []corev1.EnvVar{
									utils.DriverNamespaceEnvVar,
								},
								VolumeMounts: []corev1.VolumeMount{
									utils.CsiConfigVolumeMount,
									utils.KeysTmpDirVolumeMount,
								},
								Resources: ptr.Deref(
									pluginSpec.Resources.OMapGenerator,
									corev1.ResourceRequirements{},
								),
							})
						}
						// Liveness Sidecar Container
						if r.driver.Spec.Liveness != nil {
							containers = append(containers, corev1.Container{
								Name:            "liveness-prometheus",
								Image:           r.images["plugin"],
								ImagePullPolicy: imagePullPolicy,
								Args: utils.DeleteZeroValues(
									[]string{
										utils.TypeContainerArg("liveness"),
										utils.EndpointContainerArg,
										utils.MetricsPortContainerArg(r.driver.Spec.Liveness.MetricsPort),
										utils.MetricsPathContainerArg,
										utils.PoolTimeContainerArg,
										utils.TimeoutContainerArg(3),
									},
								),
								Env: []corev1.EnvVar{
									utils.PodIpEnvVar,
								},
								VolumeMounts: []corev1.VolumeMount{
									utils.SocketDirVolumeMount,
								},
								Resources: ptr.Deref(
									pluginSpec.Resources.Liveness,
									corev1.ResourceRequirements{},
								),
							})
						}
						// CSI LogRotate Container
						if logRotationEnabled {
							resources := ptr.Deref(pluginSpec.Resources.LogRotator, corev1.ResourceRequirements{})
							containers = append(containers, corev1.Container{
								Name:            "log-rotator",
								Image:           r.images["plugin"],
								ImagePullPolicy: imagePullPolicy,
								Resources:       resources,
								SecurityContext: logRotateSecurityContext,
								Command:         []string{"/bin/bash", "-c", logRotateCmd},
								VolumeMounts: []corev1.VolumeMount{
									utils.LogsDirVolumeMount,
									utils.LogRotateDirVolumeMount,
								},
							})
						}
						// CSI snapshot-metadata Sidecar Container
						if r.isRbdDriver() {
							if deploy, err := r.hasSnapshotMetadataServiceCr(); err != nil {
								r.log.Error(err, "Failed to check for SnapshotMetadataService CR")
							} else if deploy {
								containers = append(containers, corev1.Container{
									Name:            "csi-snapshot-metadata",
									ImagePullPolicy: imagePullPolicy,
									Image:           r.images["snapshot-metadata"],
									Args: utils.DeleteZeroValues(
										[]string{
											utils.LogVerbosityContainerArg(logVerbosity),
											utils.TimeoutContainerArg(grpcTimeout),
											utils.SnapshotMetadataGrpcServicePortArg,
											utils.CsiAddressContainerArg,
											utils.SnapshotMetadataTlsCertArg,
											utils.SnapshotMetadataTlsKeyArg,
										},
									),
									Ports: []corev1.ContainerPort{
										utils.SnapshotMetadataGrpcPort,
									},
									VolumeMounts: utils.Call(func() []corev1.VolumeMount {
										mounts := []corev1.VolumeMount{}
										tlsMountIndex := slices.IndexFunc(pluginSpec.Volumes, func(vol csiv1.VolumeSpec) bool {
											return vol.Volume.Name == "tls-key"
										})
										if tlsMountIndex != -1 {
											mounts = append(mounts, pluginSpec.Volumes[tlsMountIndex].Mount)
										}
										mounts = append(mounts, utils.SocketDirVolumeMount)

										return mounts
									}),
								})
							}
						}

						return containers
					}),
					Volumes: utils.Call(func() []corev1.Volume {
						volumes := []corev1.Volume{
							utils.HostDevVolume,
							utils.HostSysVolume,
							utils.LibModulesVolume,
							utils.SocketDirVolume,
							utils.KeysTmpDirVolume,
							utils.OidcTokenVolume,
							utils.CsiConfigVolume,
						}
						if r.driver.Spec.Encryption != nil {
							volumes = append(
								volumes,
								utils.KmsConfigVolume(&r.driver.Spec.Encryption.ConfigMapRef))
						}
						if logRotationEnabled {
							logHostPath := cmp.Or(logRotationSpec.LogHostPath, defaultLogHostPath)
							volumes = append(
								volumes,
								utils.LogsDirVolume(logHostPath, deploy.Name),
								utils.LogRotateDirVolumeName(r.driver.Name),
							)
						}
						// Add user defined volumes at the end to make sure they
						// can overwrite built in volumes.
						volumes = utils.MapMergeByKey(volumes,
							pluginSpec.Volumes,
							func(v csiv1.VolumeSpec) (corev1.Volume, bool) {
								return v.Volume, v.Volume.Name != ""
							},
							func(v corev1.Volume) string {
								return v.Name
							},
						)

						return volumes
					}),
				},
			},
		}

		return nil
	})

	logCreateOrUpdateResult(log, "controller plugin deployment", deploy, opResult, err)
	return err
}

func (r *driverReconcile) controllerPluginCsiAddonsContainerPort() corev1.ContainerPort {
	// the cephFS and rbd drivers need to use different ports
	// to avoid port collisions with host network.
	port := utils.ControllerPluginCsiAddonsContainerRbdPort
	if r.isCephFsDriver() {
		port = utils.ControllerPluginCsiAddonsContainerCephFsPort
	}

	return port
}

func (r *driverReconcile) reconcileNodePluginDaemonSetForCsiAddons() error {
	daemonSet := &appsv1.DaemonSet{}
	daemonSet.Name = r.generateName("nodeplugin-csi-addons")
	daemonSet.Namespace = r.driver.Namespace

	log := r.log.WithValues("csiAddonsDaemonSetName", daemonSet.Name)

	withCsiAddonsDaemonSet := false
	withCsiAddonsVolumeCondition := false

	if r.isRbdDriver() {
		withCsiAddonsDaemonSet = ptr.Deref(r.driver.Spec.DeployCsiAddons, false)
	}

	// check if the driver wants CSI-Addons features
	if feature := r.driver.GetAnnotations()[driverCSIAddonsFeatureVolumeCondition]; feature != "" {
		enabled, err := strconv.ParseBool(feature)
		if err != nil {
			r.log.Error(
				err,
				"Unable to parse annotation on driver.csi.ceph.io",
				"name",
				client.ObjectKeyFromObject(&r.driver),
				driverCSIAddonsFeatureVolumeCondition,
				feature,
			)
			return err
		}

		withCsiAddonsVolumeCondition = enabled
		// if the feature is enabled, enable the daemonset too
		withCsiAddonsDaemonSet = withCsiAddonsDaemonSet || withCsiAddonsVolumeCondition
	}

	if r.isNfsDriver() {
		withCsiAddonsDaemonSet = false
	}

	if !withCsiAddonsDaemonSet {
		if err := r.Delete(r.ctx, daemonSet); client.IgnoreNotFound(err) != nil {
			log.Error(err, "failed to delete csi addons daemonset")
			return err
		}
		return nil
	}

	log.Info("Reconciling csi addons nodeplugin daemonset")

	opResult, err := ctrlutil.CreateOrUpdate(r.ctx, r.Client, daemonSet, func() error {
		if err := ctrlutil.SetControllerReference(&r.driver, daemonSet, r.Scheme); err != nil {
			log.Error(err, "Failed to set owner reference on csi addons nodeplugin daemonset")

			return err
		}

		appName := daemonSet.Name
		pluginSpec := cmp.Or(r.driver.Spec.NodePlugin, &csiv1.NodePluginSpec{})
		serviceAccountName := cmp.Or(
			ptr.Deref(pluginSpec.ServiceAccountName, ""),
			fmt.Sprintf("%s%s-nodeplugin-sa", serviceAccountPrefix, r.driverType),
		)
		imagePullPolicy := cmp.Or(pluginSpec.ImagePullPolicy, corev1.PullIfNotPresent)
		logVerbosity := ptr.Deref(r.driver.Spec.Log, csiv1.LogSpec{}).Verbosity
		kubeletDirPath := cmp.Or(pluginSpec.KubeletDirPath, defaultKubeletDirPath)
		port := utils.NodePluginCsiAddonsContainerPort

		logRotationSpec := cmp.Or(r.driver.Spec.Log, &csiv1.LogSpec{}).Rotation
		logRotationEnabled := logRotationSpec != nil

		daemonSet.Spec = appsv1.DaemonSetSpec{
			Selector: &metav1.LabelSelector{
				MatchLabels: map[string]string{"app": appName},
			},
			UpdateStrategy: ptr.Deref(pluginSpec.UpdateStrategy, defaultDaemonSetUpdateStrategy),
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: utils.Call(func() map[string]string {
						podLabels := map[string]string{}
						maps.Copy(podLabels, pluginSpec.Labels)
						podLabels["app"] = appName
						return podLabels
					}),
					Annotations: maps.Clone(pluginSpec.Annotations),
				},
				Spec: corev1.PodSpec{
					ServiceAccountName: serviceAccountName,
					PriorityClassName:  ptr.Deref(pluginSpec.PrioritylClassName, ""),
					// to use e.g. Rook orchestrated cluster, and mons' FQDN is
					// resolved through k8s service, set dns policy to cluster first
					DNSPolicy:   corev1.DNSClusterFirstWithHostNet,
					Tolerations: pluginSpec.Tolerations,
					Affinity:    pluginSpec.Affinity,
					Containers: utils.Call(func() []corev1.Container {
						containers := []corev1.Container{
							{
								Name:            "csi-addons",
								Image:           r.images["addons"],
								ImagePullPolicy: imagePullPolicy,
								// We need this in order for this container to be able to access
								// the sockets created by the privileged nodeplugin container
								// on systems with enforcing selinux.
								SecurityContext: &corev1.SecurityContext{
									Privileged: ptr.To(true),
									Capabilities: &corev1.Capabilities{
										Drop: []corev1.Capability{"All"},
									},
								},
								Args: utils.DeleteZeroValues(
									[]string{
										utils.CsiAddonsNodeIdContainerArg,
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.CsiAddonsAddressContainerArg,
										utils.ContainerPortArg(port),
										utils.PodContainerArg,
										utils.NamespaceContainerArg,
										utils.PodUidContainerArg,
										utils.StagingPathContainerArg(kubeletDirPath),
										utils.If(logRotationEnabled, utils.LogToStdErrContainerArg, ""),
										utils.If(logRotationEnabled, utils.AlsoLogToStdErrContainerArg, ""),
										utils.If(logRotationEnabled, utils.LogFileContainerArg("csi-addons"), ""),
										utils.If(withCsiAddonsVolumeCondition, utils.CsiAddonsVolumeConditionArg, ""),
									},
								),
								Ports: []corev1.ContainerPort{
									port,
								},
								Env: []corev1.EnvVar{
									utils.NodeIdEnvVar,
									utils.PodNameEnvVar,
									utils.PodNamespaceEnvVar,
									utils.PodUidEnvVar,
								},
								VolumeMounts: utils.Call(func() []corev1.VolumeMount {
									mounts := []corev1.VolumeMount{
										utils.PluginDirVolumeMount,
									}
									if logRotationEnabled {
										mounts = append(mounts, utils.LogsDirVolumeMount)
									}
									if withCsiAddonsVolumeCondition {
										mounts = append(mounts,
											utils.PluginMountDirVolumeMount(kubeletDirPath),
											utils.PodsMountDirVolumeMount(kubeletDirPath),
										)
									}
									return mounts
								}),
								Resources: ptr.Deref(
									pluginSpec.Resources.Addons,
									corev1.ResourceRequirements{},
								),
							},
						}
						// CSI LogRotate Container
						if logRotationEnabled {
							resources := ptr.Deref(pluginSpec.Resources.LogRotator, corev1.ResourceRequirements{})
							containers = append(containers, corev1.Container{
								Name:            "log-rotator",
								Image:           r.images["plugin"],
								ImagePullPolicy: imagePullPolicy,
								Resources:       resources,
								SecurityContext: &corev1.SecurityContext{
									Privileged: ptr.To(true),
									Capabilities: &corev1.Capabilities{
										Drop: []corev1.Capability{"All"},
									},
								},
								Command: []string{"/bin/bash", "-c", logRotateCmd},
								VolumeMounts: []corev1.VolumeMount{
									utils.LogsDirVolumeMount,
									utils.LogRotateDirVolumeMount,
								},
							})
						}
						return containers
					}),
					Volumes: utils.Call(func() []corev1.Volume {
						volumes := []corev1.Volume{
							utils.PluginDirVolume(kubeletDirPath, r.driver.Name),
						}

						if logRotationEnabled {
							logHostPath := cmp.Or(logRotationSpec.LogHostPath, defaultLogHostPath)
							volumes = append(
								volumes,
								utils.LogsDirVolume(logHostPath, daemonSet.Name),
								utils.LogRotateDirVolumeName(r.driver.Name),
							)
						}

						if withCsiAddonsVolumeCondition {
							volumes = append(
								volumes,
								utils.PluginMountDirVolume(kubeletDirPath),
								utils.PodsMountDirVolume(kubeletDirPath),
							)
						}
						return volumes
					}),
				},
			},
		}

		return nil
	})

	logCreateOrUpdateResult(log, "csi addons node plugin daemonset", daemonSet, opResult, err)
	return err
}

func (r *driverReconcile) reconcileNodePluginDaemonSet() error {
	daemonSet := &appsv1.DaemonSet{}
	daemonSet.Name = r.generateName("nodeplugin")
	daemonSet.Namespace = r.driver.Namespace

	log := r.log.WithValues("daemonSetName", daemonSet.Name)
	log.Info("Reconciling node plugin deployment")

	opResult, err := ctrlutil.CreateOrUpdate(r.ctx, r.Client, daemonSet, func() error {
		if err := ctrlutil.SetControllerReference(&r.driver, daemonSet, r.Scheme); err != nil {
			log.Error(err, "Failed setting an owner reference on deployment")
			return err
		}

		appName := daemonSet.Name
		pluginSpec := cmp.Or(r.driver.Spec.NodePlugin, &csiv1.NodePluginSpec{})
		serviceAccountName := cmp.Or(
			ptr.Deref(pluginSpec.ServiceAccountName, ""),
			fmt.Sprintf("%s%s-nodeplugin-sa", serviceAccountPrefix, r.driverType),
		)
		imagePullPolicy := cmp.Or(pluginSpec.ImagePullPolicy, corev1.PullIfNotPresent)
		logVerbosity := ptr.Deref(r.driver.Spec.Log, csiv1.LogSpec{}).Verbosity
		kubeletDirPath := cmp.Or(pluginSpec.KubeletDirPath, defaultKubeletDirPath)
		forceKernelClient := r.isCephFsDriver() && r.driver.Spec.CephFsClientType == csiv1.KernelCephFsClient

		topology := r.isRbdDriver() && pluginSpec.Topology != nil
		domainLabels := cmp.Or(pluginSpec.Topology, &csiv1.TopologySpec{}).DomainLabels
		logRotationSpec := cmp.Or(r.driver.Spec.Log, &csiv1.LogSpec{}).Rotation
		logRotationEnabled := logRotationSpec != nil

		daemonSet.Spec = appsv1.DaemonSetSpec{
			Selector: &metav1.LabelSelector{
				MatchLabels: map[string]string{"app": appName},
			},
			UpdateStrategy: ptr.Deref(pluginSpec.UpdateStrategy, defaultDaemonSetUpdateStrategy),
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: utils.Call(func() map[string]string {
						podLabels := map[string]string{}
						maps.Copy(podLabels, pluginSpec.Labels)
						podLabels["app"] = appName
						if r.driver.Spec.Liveness != nil {
							podLabels["contains"] = fmt.Sprintf("%s-metrics", appName)
						}
						return podLabels
					}),
					Annotations: maps.Clone(pluginSpec.Annotations),
				},
				Spec: corev1.PodSpec{
					ServiceAccountName: serviceAccountName,
					PriorityClassName:  ptr.Deref(pluginSpec.PrioritylClassName, ""),
					HostNetwork:        true,
					HostPID:            r.isRbdDriver(),
					// to use e.g. Rook orchestrated cluster, and mons' FQDN is
					// resolved through k8s service, set dns policy to cluster first
					DNSPolicy:   corev1.DNSClusterFirstWithHostNet,
					Tolerations: pluginSpec.Tolerations,
					Affinity:    pluginSpec.Affinity,
					Containers: utils.Call(func() []corev1.Container {
						containers := []corev1.Container{
							// Node Plugin Container
							{
								Name:            fmt.Sprintf("csi-%splugin", r.driverType),
								Image:           r.images["plugin"],
								ImagePullPolicy: imagePullPolicy,
								SecurityContext: &corev1.SecurityContext{
									Privileged: ptr.To(true),
									Capabilities: &corev1.Capabilities{
										Add:  []corev1.Capability{"SYS_ADMIN"},
										Drop: []corev1.Capability{"All"},
									},
									AllowPrivilegeEscalation: ptr.To(true),
								},
								Args: utils.DeleteZeroValues(
									[]string{
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.TypeContainerArg(string(r.driverType)),
										utils.NodeServerContainerArg,
										utils.NodeIdContainerArg,
										utils.DriverNameContainerArg(r.driver.Name),
										utils.SetFencingContainerArg(ptr.Deref(r.driver.Spec.EnableFencing, false)),
										utils.EndpointContainerArg,
										utils.PidlimitContainerArg,
										utils.If(forceKernelClient, utils.ForceCephKernelClientContainerArg, ""),
										utils.If(
											ptr.Deref(r.driver.Spec.DeployCsiAddons, false),
											utils.CsiAddonsEndpointContainerArg,
											"",
										),
										utils.If(
											r.isRbdDriver(),
											utils.StagingPathContainerArg(kubeletDirPath),
											"",
										),
										utils.If(
											r.isCephFsDriver(),
											utils.KernelMountOptionsContainerArg(r.driver.Spec.KernelMountOptions),
											"",
										),
										utils.If(
											r.isCephFsDriver(),
											utils.FuseMountOptionsContainerArg(r.driver.Spec.FuseMountOptions),
											"",
										),
										utils.If(
											topology,
											utils.DomainLabelsContainerArg(domainLabels),
											"",
										),
										utils.If(logRotationEnabled, utils.LogToStdErrContainerArg, ""),
										utils.If(logRotationEnabled, utils.AlsoLogToStdErrContainerArg, ""),
										utils.If(
											logRotationEnabled,
											utils.LogFileContainerArg(fmt.Sprintf("csi-%splugin", r.driverType)),
											"",
										),
									},
								),
								Env: []corev1.EnvVar{
									utils.PodIpEnvVar,
									utils.NodeIdEnvVar,
									utils.PodNamespaceEnvVar,
								},
								VolumeMounts: utils.Call(func() []corev1.VolumeMount {
									mounts := []corev1.VolumeMount{
										utils.HostDevVolumeMount,
										utils.HostSysVolumeMount,
										utils.HostRunMountVolumeMount,
										utils.LibModulesVolumeMount,
										utils.KeysTmpDirVolumeMount,
										utils.PluginDirVolumeMount,
										utils.CsiConfigVolumeMount,
										utils.PluginMountDirVolumeMount(kubeletDirPath),
										utils.PodsMountDirVolumeMount(kubeletDirPath),
									}
									if ptr.Deref(pluginSpec.EnableSeLinuxHostMount, false) {
										mounts = append(mounts, utils.EtcSelinuxVolumeMount)
									}
									if r.driver.Spec.Encryption != nil {
										mounts = append(mounts, utils.KmsConfigVolumeMount)
									}
									if r.isCephFsDriver() {
										mounts = append(mounts, utils.CsiMountInfoVolumeMount)
									}
									if r.isRbdDriver() {
										mounts = append(mounts, utils.OidcTokenVolumeMount)
									}
									if logRotationEnabled {
										mounts = append(mounts, utils.LogsDirVolumeMount)
									}
									// Add user defined volume mounts at the end to make sure they
									// can overwrite built in volumes mounts.
									mounts = utils.MapMergeByKey(
										mounts,
										pluginSpec.Volumes,
										func(v csiv1.VolumeSpec) (corev1.VolumeMount, bool) {
											return v.Mount, v.Mount.Name != ""
										},
										func(v corev1.VolumeMount) string {
											return v.Name
										},
									)

									return mounts
								}),
								Resources: ptr.Deref(
									pluginSpec.Resources.Plugin,
									corev1.ResourceRequirements{},
								),
							},
							// Registrar Sidecar Container
							{
								Name:            "driver-registrar",
								Image:           r.images["registrar"],
								ImagePullPolicy: imagePullPolicy,
								// This is necessary only for systems with SELinux, where
								// non-privileged sidecar containers cannot access unix domain socket
								// created by privileged CSI driver container.
								SecurityContext: &corev1.SecurityContext{
									Privileged: ptr.To(true),
									Capabilities: &corev1.Capabilities{
										Drop: []corev1.Capability{"All"},
									},
								},
								Args: utils.DeleteZeroValues(
									[]string{
										utils.LogVerbosityContainerArg(logVerbosity),
										utils.KubeletRegistrationPathContainerArg(kubeletDirPath, r.driver.Name),
										utils.CsiAddressContainerArg,
									},
								),
								VolumeMounts: []corev1.VolumeMount{
									utils.PluginDirVolumeMount,
									utils.RegistrationDirVolumeMount,
								},
								Resources: ptr.Deref(
									pluginSpec.Resources.Registrar,
									corev1.ResourceRequirements{},
								),
							},
						}
						// Liveness Sidecar Container
						if r.driver.Spec.Liveness != nil {
							containers = append(containers, corev1.Container{
								Name:            "liveness-prometheus",
								Image:           r.images["plugin"],
								ImagePullPolicy: imagePullPolicy,
								SecurityContext: &corev1.SecurityContext{
									Privileged: ptr.To(true),
									Capabilities: &corev1.Capabilities{
										Drop: []corev1.Capability{"All"},
									},
								},
								Args: utils.DeleteZeroValues(
									[]string{
										utils.TypeContainerArg("liveness"),
										utils.EndpointContainerArg,
										utils.MetricsPortContainerArg(r.driver.Spec.Liveness.MetricsPort),
										utils.MetricsPathContainerArg,
										utils.PoolTimeContainerArg,
										utils.TimeoutContainerArg(3),
									},
								),
								Env: []corev1.EnvVar{
									utils.PodIpEnvVar,
								},
								VolumeMounts: []corev1.VolumeMount{
									utils.PluginDirVolumeMount,
								},
								Resources: ptr.Deref(
									pluginSpec.Resources.Liveness,
									corev1.ResourceRequirements{},
								),
							})
						}
						// CSI LogRotate Container
						if logRotationEnabled {
							resources := ptr.Deref(pluginSpec.Resources.LogRotator, corev1.ResourceRequirements{})
							containers = append(containers, corev1.Container{
								Name:            "log-rotator",
								Image:           r.images["plugin"],
								ImagePullPolicy: imagePullPolicy,
								Resources:       resources,
								SecurityContext: &corev1.SecurityContext{
									Privileged: ptr.To(true),
									Capabilities: &corev1.Capabilities{
										Drop: []corev1.Capability{"All"},
									},
								},
								Command: []string{"/bin/bash", "-c", logRotateCmd},
								VolumeMounts: []corev1.VolumeMount{
									utils.LogsDirVolumeMount,
									utils.LogRotateDirVolumeMount,
								},
							})
						}
						return containers
					}),
					Volumes: utils.Call(func() []corev1.Volume {
						volumes := []corev1.Volume{
							utils.HostDevVolume,
							utils.HostSysVolume,
							utils.HostRunMountVolume,
							utils.LibModulesVolume,
							utils.KeysTmpDirVolume,
							utils.CsiConfigVolume,
							utils.PluginDirVolume(kubeletDirPath, r.driver.Name),
							utils.PluginMountDirVolume(kubeletDirPath),
							utils.PodsMountDirVolume(kubeletDirPath),
							utils.RegistrationDirVolume(kubeletDirPath),
						}
						if r.isCephFsDriver() {
							volumes = append(
								volumes,
								utils.CsiMountInfoVolume(kubeletDirPath, r.driver.Name),
							)
						}
						if ptr.Deref(pluginSpec.EnableSeLinuxHostMount, false) {
							volumes = append(
								volumes,
								utils.EtcSelinuxVolume,
							)
						}
						if r.driver.Spec.Encryption != nil {
							volumes = append(
								volumes,
								utils.KmsConfigVolume(&r.driver.Spec.Encryption.ConfigMapRef),
							)
						}
						if r.isRbdDriver() {
							volumes = append(
								volumes,
								utils.OidcTokenVolume,
							)
						}
						if logRotationEnabled {
							logHostPath := cmp.Or(logRotationSpec.LogHostPath, defaultLogHostPath)
							volumes = append(
								volumes,
								utils.LogsDirVolume(logHostPath, daemonSet.Name),
								utils.LogRotateDirVolumeName(r.driver.Name),
							)
						}
						// Add user defined volumes at the end to make sure they
						// can overwrite built in volumes.
						volumes = utils.MapMergeByKey(
							volumes,
							pluginSpec.Volumes,
							func(v csiv1.VolumeSpec) (corev1.Volume, bool) {
								return v.Volume, v.Volume.Name != ""
							},
							func(v corev1.Volume) string {
								return v.Name
							},
						)
						return volumes
					}),
				},
			},
		}

		return nil
	})

	logCreateOrUpdateResult(log, "node plugin daemonset", daemonSet, opResult, err)
	return err
}

func (r *driverReconcile) reconcileLivenessService() error {
	service := &corev1.Service{}
	service.Namespace = r.driver.Namespace
	service.Name = r.generateServiceName("liveness")

	log := r.log.WithValues("service", service.Name)
	log.Info("Reconciling liveness service")

	if r.driver.Spec.Liveness != nil {
		opResult, err := ctrlutil.CreateOrUpdate(r.ctx, r.Client, service, func() error {
			if err := ctrlutil.SetControllerReference(&r.driver, service, r.Scheme); err != nil {
				log.Error(err, "Failed setting an owner reference on service")
				return err
			}

			service.Spec = corev1.ServiceSpec{
				Ports: []corev1.ServicePort{
					{
						Name:       "csi-http-metrics",
						Port:       8080,
						Protocol:   corev1.ProtocolTCP,
						TargetPort: intstr.FromInt(r.driver.Spec.Liveness.MetricsPort),
					},
				},
			}

			return nil
		})

		logCreateOrUpdateResult(log, "livness service", service, opResult, err)
		return err

	} else {
		// Remove the liveness serive if livness setting removed from the driver's spec
		if err := r.Delete(r.ctx, service); client.IgnoreNotFound(err) != nil {
			log.Error(err, "Unable to delete liveness service")
			return err
		}
		return nil
	}
}

func (r *driverReconcile) isRbdDriver() bool {
	return r.driverType == RbdDriverType
}

func (r *driverReconcile) isCephFsDriver() bool {
	return r.driverType == CephFsDriverType
}

func (r *driverReconcile) isNfsDriver() bool {
	return r.driverType == NfsDriverType
}

func (r *driverReconcile) generateName(suffix string) string {
	return fmt.Sprintf("%s-%s", r.driver.Name, suffix)
}

// generateServiceName generates a service name by replacing all special characters with a hyphen
// as required by Kubernetes for service name, This assumes that the name
// from the generateName is RFC 1123 compliant.
func (r *driverReconcile) generateServiceName(suffix string) string {
	name := r.generateName(suffix)
	// Define a regex pattern to match all special characters except hyphen
	re := regexp.MustCompile(`[^a-z0-9-]`)
	// Replace all special characters with a hyphen
	return re.ReplaceAllString(name, "-")
}

func getControllerPluginPodAffinity(
	pluginSpec *csiv1.ControllerPluginSpec,
	selector *metav1.LabelSelector,
) *corev1.Affinity {
	if pluginSpec.Affinity != nil && pluginSpec.Affinity.PodAntiAffinity != nil {
		return pluginSpec.Affinity
	}

	affinity := &corev1.Affinity{}
	if pluginSpec.Affinity != nil {
		pluginSpec.Affinity.DeepCopyInto(affinity)
	}

	affinity.PodAntiAffinity = &corev1.PodAntiAffinity{
		RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{{
			LabelSelector: selector,
			TopologyKey:   corev1.LabelHostname,
		}},
	}

	return affinity
}

func logCreateOrUpdateResult(
	log logr.Logger,
	subject string,
	obj client.Object,
	opRes ctrlutil.OperationResult,
	err error,
) {
	if err != nil {
		verb := utils.If(obj.GetUID() == "", "creating", "updating")
		log.Error(err, fmt.Sprintf("Failed %s %s", verb, subject))
		return
	}

	switch opRes {
	case ctrlutil.OperationResultNone:
		log.Info(fmt.Sprintf("%s is already up to date", subject))
	default:
		log.Info(fmt.Sprintf("%s %s successfully", subject, opRes))
	}
}

// mergeDriverSpecs will fill in any unset fields in dest with a copy of the same field in src
func mergeDriverSpecs(dest, src *csiv1.DriverSpec) {
	// Create a copy of the src, making sure that any value copied into dest is a not shared
	// with the original src
	src = src.DeepCopy()

	if dest.Log == nil {
		dest.Log = src.Log
	}
	if dest.ClusterName == nil {
		dest.ClusterName = src.ClusterName
	}
	if dest.EnableMetadata == nil {
		dest.EnableMetadata = src.EnableMetadata
	}
	if dest.GRpcTimeout == 0 {
		dest.GRpcTimeout = src.GRpcTimeout
	}
	if dest.SnapshotPolicy == "" {
		dest.SnapshotPolicy = src.SnapshotPolicy
	}
	if dest.GenerateOMapInfo == nil {
		dest.GenerateOMapInfo = src.GenerateOMapInfo
	}
	if dest.EnableFencing == nil {
		dest.EnableFencing = src.EnableFencing
	}
	if dest.FsGroupPolicy == "" {
		dest.FsGroupPolicy = src.FsGroupPolicy
	}
	if dest.Encryption == nil {
		dest.Encryption = src.Encryption
	}
	if src.NodePlugin != nil {
		if dest.NodePlugin == nil {
			dest.NodePlugin = src.NodePlugin
		} else {
			dest, src := dest.NodePlugin, src.NodePlugin
			if dest.ServiceAccountName == nil {
				dest.ServiceAccountName = src.ServiceAccountName
			}
			if dest.PrioritylClassName == nil {
				dest.PrioritylClassName = src.PrioritylClassName
			}
			if dest.Labels == nil {
				dest.Labels = src.Labels
			}
			if dest.Annotations == nil {
				dest.Annotations = src.Annotations
			}
			if dest.Affinity == nil {
				dest.Affinity = src.Affinity
			}
			if dest.Tolerations == nil {
				dest.Tolerations = src.Tolerations
			}
			if dest.Volumes == nil {
				dest.Volumes = src.Volumes
			}
			if dest.ImagePullPolicy == "" {
				dest.ImagePullPolicy = src.ImagePullPolicy
			}
			if dest.UpdateStrategy == nil {
				dest.UpdateStrategy = src.UpdateStrategy
			}
			if dest.KubeletDirPath == "" {
				dest.KubeletDirPath = src.KubeletDirPath
			}
			if dest.EnableSeLinuxHostMount == nil {
				dest.EnableSeLinuxHostMount = src.EnableSeLinuxHostMount
			}
			if dest.Resources.Registrar == nil {
				dest.Resources.Registrar = src.Resources.Registrar
			}
			if dest.Resources.Liveness == nil {
				dest.Resources.Liveness = src.Resources.Liveness
			}
			if dest.Resources.Plugin == nil {
				dest.Resources.Plugin = src.Resources.Plugin
			}
			if dest.Resources.LogRotator == nil {
				dest.Resources.LogRotator = src.Resources.LogRotator
			}
		}
	}
	if src.ControllerPlugin != nil {
		if dest.ControllerPlugin == nil {
			dest.ControllerPlugin = src.ControllerPlugin
		} else {
			dest, src := dest.ControllerPlugin, src.ControllerPlugin
			if dest.ServiceAccountName == nil {
				dest.ServiceAccountName = src.ServiceAccountName
			}
			if dest.PrioritylClassName == nil {
				dest.PrioritylClassName = src.PrioritylClassName
			}
			if dest.Labels == nil {
				dest.Labels = src.Labels
			}
			if dest.Annotations == nil {
				dest.Annotations = src.Annotations
			}
			if dest.Affinity == nil {
				dest.Affinity = src.Affinity
			}
			if dest.Tolerations == nil {
				dest.Tolerations = src.Tolerations
			}
			if dest.Volumes == nil {
				dest.Volumes = src.Volumes
			}
			if dest.ImagePullPolicy == "" {
				dest.ImagePullPolicy = src.ImagePullPolicy
			}
			if dest.Replicas == nil {
				dest.Replicas = src.Replicas
			}
			if dest.Privileged == nil {
				dest.Privileged = src.Privileged
			}
			if dest.Resources.Attacher == nil {
				dest.Resources.Attacher = src.Resources.Attacher
			}
			if dest.Resources.Snapshotter == nil {
				dest.Resources.Snapshotter = src.Resources.Snapshotter
			}
			if dest.Resources.Resizer == nil {
				dest.Resources.Resizer = src.Resources.Resizer
			}
			if dest.Resources.Provisioner == nil {
				dest.Resources.Provisioner = src.Resources.Provisioner
			}
			if dest.Resources.OMapGenerator == nil {
				dest.Resources.OMapGenerator = src.Resources.OMapGenerator
			}
			if dest.Resources.Liveness == nil {
				dest.Resources.Liveness = src.Resources.Liveness
			}
			if dest.Resources.Plugin == nil {
				dest.Resources.Plugin = src.Resources.Plugin
			}
			if dest.Resources.LogRotator == nil {
				dest.Resources.LogRotator = src.Resources.LogRotator
			}
		}
	}
	if dest.AttachRequired == nil {
		dest.AttachRequired = src.AttachRequired
	}
	if dest.Liveness == nil {
		dest.Liveness = src.Liveness
	}
	if dest.LeaderElection == nil {
		dest.LeaderElection = src.LeaderElection
	}
	if dest.DeployCsiAddons == nil {
		dest.DeployCsiAddons = src.DeployCsiAddons
	}
	if dest.KernelMountOptions == nil {
		dest.KernelMountOptions = src.KernelMountOptions
	}
	if dest.FuseMountOptions == nil {
		dest.FuseMountOptions = src.FuseMountOptions
	}
	if dest.CephFsClientType == "" {
		dest.CephFsClientType = src.CephFsClientType
	}
}
