package cluster

import (
	"context"
	"fmt"
	"regexp"
	"strconv"
	"strings"

	"github.com/rancher/norman/types/convert"
	v3 "github.com/rancher/rancher/pkg/apis/management.cattle.io/v3"
	v1 "github.com/rancher/rancher/pkg/apis/provisioning.cattle.io/v1"
	rkev1 "github.com/rancher/rancher/pkg/apis/rke.cattle.io/v1"
	"github.com/rancher/rancher/pkg/capr"
	"github.com/rancher/rancher/pkg/features"
	fleetconst "github.com/rancher/rancher/pkg/fleet"
	capicontrollers "github.com/rancher/rancher/pkg/generated/controllers/cluster.x-k8s.io/v1beta1"
	mgmtcontrollers "github.com/rancher/rancher/pkg/generated/controllers/management.cattle.io/v3"
	rocontrollers "github.com/rancher/rancher/pkg/generated/controllers/provisioning.cattle.io/v1"
	rkecontrollers "github.com/rancher/rancher/pkg/generated/controllers/rke.cattle.io/v1"
	"github.com/rancher/rancher/pkg/provisioningv2/image"
	"github.com/rancher/rancher/pkg/provisioningv2/kubeconfig"
	"github.com/rancher/rancher/pkg/settings"
	"github.com/rancher/rancher/pkg/wrangler"
	"github.com/rancher/wrangler/v3/pkg/apply"
	"github.com/rancher/wrangler/v3/pkg/condition"
	corecontrollers "github.com/rancher/wrangler/v3/pkg/generated/controllers/core/v1"
	"github.com/rancher/wrangler/v3/pkg/generic"
	"github.com/rancher/wrangler/v3/pkg/genericcondition"
	"github.com/rancher/wrangler/v3/pkg/kstatus"
	"github.com/rancher/wrangler/v3/pkg/name"
	"github.com/rancher/wrangler/v3/pkg/randomtoken"
	"github.com/rancher/wrangler/v3/pkg/relatedresource"
	"github.com/rancher/wrangler/v3/pkg/yaml"
	corev1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	apierror "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
)

const (
	ByCluster          = "by-cluster"
	ByCloudCred        = "by-cloud-cred"
	ByETCDSnapshotName = "by-etcd-snapshot-name"

	creatorIDAnn              = "field.cattle.io/creatorId"
	administratedAnn          = "provisioning.cattle.io/administrated"
	mgmtClusterNameAnn        = "provisioning.cattle.io/management-cluster-name"
	mgmtClusterDisplayNameAnn = "provisioning.cattle.io/management-cluster-display-name"
	fleetWorkspaceNameAnn     = "provisioning.cattle.io/fleet-workspace-name"
	externallyManagedAnn      = "provisioning.cattle.io/externally-managed"

	manageSchedulingDefaultsAnn = "provisioning.cattle.io/enable-scheduling-customization"
)

var (
	mgmtNameRegexp = regexp.MustCompile("^(c-[a-z0-9]{5}|local)$")
)

type handler struct {
	mgmtClusterCache      mgmtcontrollers.ClusterCache
	mgmtClusters          mgmtcontrollers.ClusterController
	clusterTokenCache     mgmtcontrollers.ClusterRegistrationTokenCache
	clusterTokens         mgmtcontrollers.ClusterRegistrationTokenClient
	featureCache          mgmtcontrollers.FeatureCache
	featureClient         mgmtcontrollers.FeatureClient
	clusters              rocontrollers.ClusterController
	clusterCache          rocontrollers.ClusterCache
	rkeControlPlanes      rkecontrollers.RKEControlPlaneClient
	rkeControlPlanesCache rkecontrollers.RKEControlPlaneCache
	secretCache           corecontrollers.SecretCache
	kubeconfigManager     *kubeconfig.Manager
	apply                 apply.Apply

	capiClustersCache capicontrollers.ClusterCache
	capiClusters      capicontrollers.ClusterClient
	capiMachinesCache capicontrollers.MachineCache
}

func handlerWithoutCAPI(clients *wrangler.Context, kubeconfigManager *kubeconfig.Manager) *handler {
	return &handler{
		mgmtClusterCache:      clients.Mgmt.Cluster().Cache(),
		mgmtClusters:          clients.Mgmt.Cluster(),
		clusterTokenCache:     clients.Mgmt.ClusterRegistrationToken().Cache(),
		clusterTokens:         clients.Mgmt.ClusterRegistrationToken(),
		featureCache:          clients.Mgmt.Feature().Cache(),
		featureClient:         clients.Mgmt.Feature(),
		clusters:              clients.Provisioning.Cluster(),
		clusterCache:          clients.Provisioning.Cluster().Cache(),
		rkeControlPlanes:      clients.RKE.RKEControlPlane(),
		rkeControlPlanesCache: clients.RKE.RKEControlPlane().Cache(),
		secretCache:           clients.Core.Secret().Cache(),
		kubeconfigManager:     kubeconfigManager,
		apply: clients.Apply.WithCacheTypes(
			clients.Provisioning.Cluster(),
			clients.Mgmt.Cluster()),
	}
}

func EarlyRegister(ctx context.Context, clients *wrangler.Context, kubeconfigManager *kubeconfig.Manager) {
	h := handlerWithoutCAPI(clients, kubeconfigManager)
	// Register a generating handler in order to generate clusters.provisioning.cattle.io/v1 objects based on
	// clusters.management.cattle.io/v3 (legacy) cluster objects. Do this as early as possible to ensure
	// the UI shows the local cluster, even if CAPI is not fully ready.
	mgmtcontrollers.RegisterClusterGeneratingHandler(ctx,
		clients.Mgmt.Cluster(),
		clients.Apply.WithCacheTypes(clients.Provisioning.Cluster()),
		"",
		"provisioning-cluster-create",
		h.generateProvisioningClusterFromLegacyCluster,
		nil)

	clusterCreateApply := clients.Apply.WithCacheTypes(clients.Mgmt.Cluster(),
		clients.Mgmt.ClusterRegistrationToken(),
		clients.RBAC.ClusterRoleBinding(),
		clients.Core.Namespace(),
		clients.Core.Secret())

	if features.MCM.Enabled() {
		clusterCreateApply = clusterCreateApply.WithCacheTypes(clients.Mgmt.ClusterRoleTemplateBinding())
	}

	// Register a generating handler in order to generate clusters.management.cattle.io/v3 objects based on
	// clusters.provisioning.cattle.io/v1 objects.
	rocontrollers.RegisterClusterGeneratingHandler(ctx,
		clients.Provisioning.Cluster(),
		clusterCreateApply,
		"Created",
		"cluster-create",
		h.generateLegacyClusterFromProvisioningCluster,
		&generic.GeneratingHandlerOptions{
			AllowClusterScoped: true,
		},
	)

	clients.Provisioning.Cluster().OnChange(ctx, "provisioning-cluster-update", h.OnChange)

	clients.Mgmt.Cluster().OnChange(ctx, "cluster-watch", h.createToken)

	clients.Provisioning.Cluster().OnChange(ctx, "v1-scheduling-customization-backfill", h.updateV1SchedulingCustomization)
	clients.Mgmt.Cluster().OnChange(ctx, "v3-scheduling-customization-backfill", h.updateV3SchedulingCustomization)

	relatedresource.Watch(ctx, "cluster-watch", h.clusterWatch,
		clients.Provisioning.Cluster(), clients.Mgmt.Cluster())
}

func Register(
	ctx context.Context,
	clients *wrangler.CAPIContext, kubeconfigManager *kubeconfig.Manager) {
	h := handler{
		mgmtClusterCache:      clients.Mgmt.Cluster().Cache(),
		mgmtClusters:          clients.Mgmt.Cluster(),
		clusterTokenCache:     clients.Mgmt.ClusterRegistrationToken().Cache(),
		clusterTokens:         clients.Mgmt.ClusterRegistrationToken(),
		featureCache:          clients.Mgmt.Feature().Cache(),
		featureClient:         clients.Mgmt.Feature(),
		clusters:              clients.Provisioning.Cluster(),
		clusterCache:          clients.Provisioning.Cluster().Cache(),
		rkeControlPlanes:      clients.RKE.RKEControlPlane(),
		rkeControlPlanesCache: clients.RKE.RKEControlPlane().Cache(),
		secretCache:           clients.Core.Secret().Cache(),

		capiClustersCache: clients.CAPI.Cluster().Cache(),
		capiClusters:      clients.CAPI.Cluster(),
		capiMachinesCache: clients.CAPI.Machine().Cache(),

		kubeconfigManager: kubeconfigManager,
		apply: clients.Apply.WithCacheTypes(
			clients.Provisioning.Cluster(),
			clients.Mgmt.Cluster()),
	}

	clients.Mgmt.Cluster().OnRemove(ctx, "mgmt-cluster-remove", h.OnMgmtClusterRemove)
	clients.Provisioning.Cluster().OnRemove(ctx, "provisioning-cluster-remove", h.OnClusterRemove)
}

func RegisterIndexers(config *wrangler.Context) {
	config.Provisioning.Cluster().Cache().AddIndexer(ByCluster, byClusterIndex)
	config.Provisioning.Cluster().Cache().AddIndexer(ByCloudCred, byCloudCredentialIndex)
	if features.Provisioningv2ETCDSnapshotBackPopulation.Enabled() {
		config.RKE.ETCDSnapshot().Cache().AddIndexer(ByETCDSnapshotName, byETCDSnapshotName)
	}
}

func ETCDSnapshotKey(obj *rkev1.ETCDSnapshot) string {
	return fmt.Sprintf("%s/%s/%s", obj.Namespace, obj.Labels[capr.ClusterNameLabel], obj.Annotations[capr.SnapshotNameAnnotation])
}

func byETCDSnapshotName(obj *rkev1.ETCDSnapshot) ([]string, error) {
	if obj.Annotations == nil || obj.Annotations[capr.SnapshotNameAnnotation] == "" {
		return nil, nil
	}
	if obj.Labels == nil || obj.Labels[capr.ClusterNameLabel] == "" {
		return nil, nil
	}
	return []string{ETCDSnapshotKey(obj)}, nil
}

func byClusterIndex(obj *v1.Cluster) ([]string, error) {
	if obj.Status.ClusterName == "" {
		return nil, nil
	}
	return []string{obj.Status.ClusterName}, nil
}

func byCloudCredentialIndex(obj *v1.Cluster) ([]string, error) {
	if obj.Spec.CloudCredentialSecretName == "" {
		return nil, nil
	}
	return []string{obj.Spec.CloudCredentialSecretName}, nil
}

func (h *handler) clusterWatch(namespace, name string, obj runtime.Object) ([]relatedresource.Key, error) {
	cluster, ok := obj.(*v3.Cluster)
	if !ok {
		return nil, nil
	}
	operatorClusters, err := h.clusterCache.GetByIndex(ByCluster, cluster.Name)
	if err != nil || len(operatorClusters) == 0 {
		// ignore
		return nil, nil
	}
	return []relatedresource.Key{
		{
			Namespace: operatorClusters[0].Namespace,
			Name:      operatorClusters[0].Name,
		},
	}, nil
}

// isLegacyCluster returns true if the cluster name for a clusters.provisioning.cattle.io/v1 or
// clusters.management.cattle.io/v3 cluster name matches the regex for a legacy cluster (c-XXXXX|local) where XXXXX is a
// random string of characters.
func (h *handler) isLegacyCluster(cluster interface{}) bool {
	if c, ok := cluster.(*v3.Cluster); ok {
		return mgmtNameRegexp.MatchString(c.Name)
	} else if c, ok := cluster.(*v1.Cluster); ok {
		return mgmtNameRegexp.MatchString(c.Name)
	}
	return false
}

// generateProvisioningClusterFromLegacyCluster will generate a clusters.provisioning.cattle.io/v1 object with a passed
// in clusters.management.cattle.io/v3 object. It will not generate a clusters.provisioning.cattle.io/v1 cluster if the
// cluster FleetWorkspaceName is empty or if the cluster name does not match (c-XXXXX|local) where XXXXX is a random
// string of characters.
func (h *handler) generateProvisioningClusterFromLegacyCluster(cluster *v3.Cluster, status v3.ClusterStatus) ([]runtime.Object, v3.ClusterStatus, error) {
	// Clusters annotated with "provisioning.cattle.io/externally-managed": "true" should get cluster created
	// when the fleet workspace defaulting is disabled
	var clusterExternallyManaged bool
	// For legacy purposes of the way our API generally works, make sure the value is not set to "false"
	if ann := cluster.Annotations[externallyManagedAnn]; ann != "" && (strings.ToLower(ann) != "false") {
		clusterExternallyManaged = true
	}

	if !h.isLegacyCluster(cluster) || (cluster.Spec.FleetWorkspaceName == "" && !clusterExternallyManaged) {
		return nil, status, nil
	}
	provCluster := &v1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:        cluster.Name,
			Namespace:   cluster.Spec.FleetWorkspaceName,
			Labels:      yaml.CleanAnnotationsForExport(cluster.Labels),
			Annotations: yaml.CleanAnnotationsForExport(cluster.Annotations),
		},
	}

	provCluster.Annotations[mgmtClusterDisplayNameAnn] = cluster.Spec.DisplayName

	if cluster.Spec.ClusterAgentDeploymentCustomization != nil {
		clusterAgentCustomizationCopy := cluster.Spec.ClusterAgentDeploymentCustomization.DeepCopy()
		provCluster.Spec.ClusterAgentDeploymentCustomization = &v1.AgentDeploymentCustomization{
			AppendTolerations:            clusterAgentCustomizationCopy.AppendTolerations,
			OverrideAffinity:             clusterAgentCustomizationCopy.OverrideAffinity,
			OverrideResourceRequirements: clusterAgentCustomizationCopy.OverrideResourceRequirements,
		}
	}
	if cluster.Spec.FleetAgentDeploymentCustomization != nil {
		fleetAgentCustomizationCopy := cluster.Spec.FleetAgentDeploymentCustomization.DeepCopy()
		provCluster.Spec.FleetAgentDeploymentCustomization = &v1.AgentDeploymentCustomization{
			AppendTolerations:            fleetAgentCustomizationCopy.AppendTolerations,
			OverrideAffinity:             fleetAgentCustomizationCopy.OverrideAffinity,
			OverrideResourceRequirements: fleetAgentCustomizationCopy.OverrideResourceRequirements,
		}
	}

	if cluster.Spec.ClusterAgentDeploymentCustomization != nil && cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization != nil {
		provCluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization = &v1.AgentSchedulingCustomization{}

		if cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PodDisruptionBudget != nil {
			provCluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PodDisruptionBudget = &v1.PodDisruptionBudgetSpec{
				MinAvailable:   cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PodDisruptionBudget.MinAvailable,
				MaxUnavailable: cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PodDisruptionBudget.MaxUnavailable,
			}
		}

		if cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PriorityClass != nil {
			provCluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PriorityClass = &v1.PriorityClassSpec{
				Value:            cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PriorityClass.Value,
				PreemptionPolicy: cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PriorityClass.PreemptionPolicy,
			}
		}
	}

	return []runtime.Object{
		provCluster,
	}, status, nil
}

// generateLegacyClusterFromProvisioningCluster will generate and return a clusters.management.cattle.io/v3 object based
// on the clusters.provisioning.cattle.io/v1 object passed in.
func (h *handler) generateLegacyClusterFromProvisioningCluster(cluster *v1.Cluster, status v1.ClusterStatus) ([]runtime.Object, v1.ClusterStatus, error) {
	switch {
	case cluster.Spec.ClusterAPIConfig != nil:
		return h.createClusterAndDeployAgent(cluster, status)
	default:
		return h.createCluster(cluster, status, v3.ClusterSpec{
			ImportedConfig: &v3.ImportedConfig{},
		})
	}
}

func NormalizeCluster(cluster *v3.Cluster, isImportedCluster bool) (runtime.Object, error) {
	// We do this so that we don't clobber status because the rancher object is pretty dirty and doesn't have a status subresource
	data, err := convert.EncodeToMap(cluster)
	if err != nil {
		return nil, err
	}
	spec, _ := data["spec"].(map[string]interface{})
	if _, ok := spec["localClusterAuthEndpoint"]; ok && isImportedCluster {
		// For imported clusters, we need to delete the localClusterAuthEndpoint so that it doesn't get overwritten here.
		// In general, imported clusters don't support localClusterAuthEndpoint.
		// However, imported RKE2/K3S clusters do and this is driven by the management cluster.
		delete(spec, "localClusterAuthEndpoint")
	}
	data = map[string]interface{}{
		"metadata": data["metadata"],
		"spec":     spec,
	}
	data["kind"] = "Cluster"
	data["apiVersion"] = "management.cattle.io/v3"
	return &unstructured.Unstructured{Object: data}, nil
}

func (h *handler) createToken(_ string, cluster *v3.Cluster) (*v3.Cluster, error) {
	if cluster == nil {
		return cluster, nil
	}
	_, err := h.clusterTokenCache.Get(cluster.Name, "default-token")
	if apierror.IsNotFound(err) {
		_, err = h.clusterTokens.Create(&v3.ClusterRegistrationToken{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "default-token",
				Namespace: cluster.Name,
			},
			Spec: v3.ClusterRegistrationTokenSpec{
				ClusterName: cluster.Name,
			},
		})
		return cluster, err
	}
	return cluster, err
}

// OnChange updates annotations on clusters.provisioning.cattle.io/v1 for server-side pagination.
func (h *handler) OnChange(_ string, cluster *v1.Cluster) (*v1.Cluster, error) {
	if cluster == nil || cluster.DeletionTimestamp != nil || h.isLegacyCluster(cluster) {
		// generateProvisioningClusterFromLegacyCluster manages clusters.provisioning.cattle.io/v1
		// for legacy clusters.management.cattle.io/v3
		return cluster, nil
	}
	if cluster.Annotations == nil {
		cluster.Annotations = make(map[string]string)
	}
	// createNewCluster sets the display name of a clusters.management.cattle.io/v3 cluster
	// to match the user-provided name from clusters.provisioning.cattle.io/v1 object
	if cluster.Annotations[mgmtClusterDisplayNameAnn] == cluster.Name {
		return cluster, nil
	}
	clusterCopy := cluster.DeepCopy()
	clusterCopy.Annotations[mgmtClusterDisplayNameAnn] = cluster.Name
	return h.clusters.Update(clusterCopy)
}

func (h *handler) createCluster(cluster *v1.Cluster, status v1.ClusterStatus, spec v3.ClusterSpec) ([]runtime.Object, v1.ClusterStatus, error) {
	if h.isLegacyCluster(cluster) {
		mgmtCluster, err := h.mgmtClusterCache.Get(cluster.Name)
		if err != nil {
			return nil, status, err
		}
		return h.updateStatus(nil, cluster, status, mgmtCluster)
	}
	return h.createNewCluster(cluster, status, spec)
}

func mgmtClusterName() (string, error) {
	rand, err := randomtoken.Generate()
	if err != nil {
		return "", err
	}
	return name.SafeConcatName("c", "m", rand[:8]), nil
}

func (h *handler) updateImportedCluster(cluster *v1.Cluster, status v1.ClusterStatus, mgmtCluster *v3.Cluster) ([]runtime.Object, v1.ClusterStatus, error) {
	newCluster := &v3.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:        mgmtCluster.Name,
			Labels:      mgmtCluster.Labels,
			Annotations: mgmtCluster.Annotations,
		},
		Spec: mgmtCluster.Spec,
	}

	if features.ProvisioningV2FleetWorkspaceBackPopulation.Enabled() {
		if forcedFleetWorkspaceName, ok := cluster.Annotations[fleetWorkspaceNameAnn]; ok && forcedFleetWorkspaceName != "" { // force set the fleet workspace name
			newCluster.Spec.FleetWorkspaceName = forcedFleetWorkspaceName
		}
		// backpopulateMgmtClusterFleetWorkspaceName() is not required, newCluster already has FleetWorkspaceName from mgmt cluster
	}

	delete(cluster.Annotations, creatorIDAnn)
	status.FleetWorkspaceName = newCluster.Spec.FleetWorkspaceName

	normalizedCluster, err := NormalizeCluster(newCluster, cluster.Spec.RKEConfig == nil)
	if err != nil {
		return nil, status, err
	}

	return h.updateStatus([]runtime.Object{
		normalizedCluster,
	}, cluster, status, newCluster)
}

// createNewCluster creates a homologated clusters.management.cattle.io/v3 object based on a
// clusters.provisioning.cattle.io/v1 object and the spec of the clusters.management.cattle.io/v3 object passed in.
// It ensures the management cluster remains the source of truth for imported clusters after cluster creation.
func (h *handler) createNewCluster(cluster *v1.Cluster, status v1.ClusterStatus, spec v3.ClusterSpec) ([]runtime.Object, v1.ClusterStatus, error) {
	if cluster.Spec.RKEConfig == nil {
		mgmtCluster, err := h.mgmtClusterCache.Get(cluster.Status.ClusterName)
		if err == nil {
			return h.updateImportedCluster(cluster, status, mgmtCluster)
		} else if !apierror.IsNotFound(err) {
			return nil, status, err
		}
	}

	spec.DisplayName = cluster.Name
	spec.Description = cluster.Annotations["field.cattle.io/description"]
	spec.DefaultPodSecurityAdmissionConfigurationTemplateName = cluster.Spec.DefaultPodSecurityAdmissionConfigurationTemplateName
	spec.DefaultClusterRoleForProjectMembers = cluster.Spec.DefaultClusterRoleForProjectMembers
	spec.EnableNetworkPolicy = cluster.Spec.EnableNetworkPolicy
	spec.DesiredAgentImage = image.ResolveWithCluster(settings.AgentImage.Get(), cluster)
	spec.DesiredAuthImage = image.ResolveWithCluster(settings.AuthImage.Get(), cluster)

	spec.ClusterSecrets.PrivateRegistrySecret = image.GetPrivateRepoSecretFromCluster(cluster)
	spec.ClusterSecrets.PrivateRegistryURL = image.GetPrivateRepoURLFromCluster(cluster)

	spec.AgentEnvVars = nil
	for _, env := range cluster.Spec.AgentEnvVars {
		spec.AgentEnvVars = append(spec.AgentEnvVars, corev1.EnvVar{
			Name:  env.Name,
			Value: env.Value,
		})
	}

	if cluster.Spec.ClusterAgentDeploymentCustomization != nil {
		clusterAgentCustomizationCopy := cluster.Spec.ClusterAgentDeploymentCustomization.DeepCopy()
		spec.ClusterAgentDeploymentCustomization = &v3.AgentDeploymentCustomization{
			AppendTolerations:            clusterAgentCustomizationCopy.AppendTolerations,
			OverrideAffinity:             clusterAgentCustomizationCopy.OverrideAffinity,
			OverrideResourceRequirements: clusterAgentCustomizationCopy.OverrideResourceRequirements,
		}
	}
	if cluster.Spec.FleetAgentDeploymentCustomization != nil {
		fleetAgentCustomizationCopy := cluster.Spec.FleetAgentDeploymentCustomization.DeepCopy()
		spec.FleetAgentDeploymentCustomization = &v3.AgentDeploymentCustomization{
			AppendTolerations:            fleetAgentCustomizationCopy.AppendTolerations,
			OverrideAffinity:             fleetAgentCustomizationCopy.OverrideAffinity,
			OverrideResourceRequirements: fleetAgentCustomizationCopy.OverrideResourceRequirements,
		}
	}

	if cluster.Spec.ClusterAgentDeploymentCustomization != nil && cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization != nil {
		spec.ClusterAgentDeploymentCustomization.SchedulingCustomization = &v3.AgentSchedulingCustomization{}

		if cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PodDisruptionBudget != nil {
			spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PodDisruptionBudget = &v3.PodDisruptionBudgetSpec{
				MaxUnavailable: cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PodDisruptionBudget.MaxUnavailable,
				MinAvailable:   cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PodDisruptionBudget.MinAvailable,
			}
		}

		if cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PriorityClass != nil {
			spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PriorityClass = &v3.PriorityClassSpec{
				Value:            cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PriorityClass.Value,
				PreemptionPolicy: cluster.Spec.ClusterAgentDeploymentCustomization.SchedulingCustomization.PriorityClass.PreemptionPolicy,
			}
		}
	}

	if cluster.Spec.RKEConfig != nil {
		if cluster.Spec.RKEConfig.DataDirectories.SystemAgent != "" {
			spec.AgentEnvVars = append(spec.AgentEnvVars, corev1.EnvVar{
				Name:  capr.SystemAgentDataDirEnvVar,
				Value: cluster.Spec.RKEConfig.DataDirectories.SystemAgent,
			})
		}
		if err := h.updateFeatureLockedValue(true); err != nil {
			return nil, status, err
		}
	}

	spec.LocalClusterAuthEndpoint = v3.LocalClusterAuthEndpoint{
		FQDN:    cluster.Spec.LocalClusterAuthEndpoint.FQDN,
		CACerts: cluster.Spec.LocalClusterAuthEndpoint.CACerts,
		Enabled: cluster.Spec.LocalClusterAuthEndpoint.Enabled,
	}

	newCluster := &v3.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:        cluster.Status.ClusterName,
			Labels:      cluster.Labels,
			Annotations: map[string]string{administratedAnn: strconv.FormatBool(cluster.Spec.RKEConfig != nil)},
		},
		Spec: spec,
	}

	if mgmtClusterNameAnnVal, ok := cluster.Annotations[mgmtClusterNameAnn]; ok && mgmtClusterNameAnnVal != "" && newCluster.Name == "" {
		// If the management cluster name annotation is set to a non-empty value, and the mgmt cluster name has not been set yet, set the cluster name to the mgmt cluster name.
		newCluster.Name = mgmtClusterNameAnnVal
	} else if newCluster.Name == "" {
		// If the management cluster name annotation is not set and the cluster name has not yet been generated, generate and set a new mgmt cluster name.
		mgmtName, err := mgmtClusterName()
		if err != nil {
			return nil, status, err
		}
		newCluster.Name = mgmtName
	}

	for k, v := range cluster.Annotations {
		newCluster.Annotations[k] = v
	}

	delete(cluster.Annotations, creatorIDAnn)

	if features.ProvisioningV2FleetWorkspaceBackPopulation.Enabled() {
		if forcedFleetWorkspaceName, ok := cluster.Annotations[fleetWorkspaceNameAnn]; ok && forcedFleetWorkspaceName != "" { // force set the fleet workspace name
			newCluster.Spec.FleetWorkspaceName = forcedFleetWorkspaceName
		} else {
			if err := h.backpopulateMgmtClusterFleetWorkspaceName(newCluster); err != nil {
				return nil, status, err
			}
			if newCluster.Spec.FleetWorkspaceName == "" { // fall back to using the provisioning cluster namespace as the fleet workspace name
				newCluster.Spec.FleetWorkspaceName = cluster.Namespace
			}
		}
	} else {
		newCluster.Spec.FleetWorkspaceName = cluster.Namespace
	}

	status.FleetWorkspaceName = newCluster.Spec.FleetWorkspaceName

	normalizedCluster, err := NormalizeCluster(newCluster, cluster.Spec.RKEConfig == nil)
	if err != nil {
		return nil, status, err
	}

	return h.updateStatus([]runtime.Object{
		normalizedCluster,
	}, cluster, status, newCluster)
}

// backpopulateMgmtClusterFleetWorkspaceName backpopulates the fleet workspace name field from the v3 management cluster object onto the new desired object
func (h *handler) backpopulateMgmtClusterFleetWorkspaceName(rCluster *v3.Cluster) error {
	if rCluster == nil {
		return nil
	}
	existing, err := h.mgmtClusterCache.Get(rCluster.Name)
	if err != nil && !apierror.IsNotFound(err) {
		return err
	}
	if existing != nil {
		rCluster.Spec.FleetWorkspaceName = existing.Spec.FleetWorkspaceName
	}
	return nil
}

// updateStatus will update the status on the clusters.provisioning.cattle.io/v1 object.
func (h *handler) updateStatus(objs []runtime.Object, cluster *v1.Cluster, status v1.ClusterStatus, rCluster *v3.Cluster) ([]runtime.Object, v1.ClusterStatus, error) {
	ready := false
	existing, err := h.mgmtClusterCache.Get(rCluster.Name)
	if err != nil && !apierror.IsNotFound(err) {
		return nil, status, err
	} else if err == nil {
		if condition.Cond("Ready").IsTrue(existing) {
			ready = true
		}
		for _, messageCond := range existing.Status.Conditions {
			if messageCond.Type == "Updated" || messageCond.Type == "Provisioned" || messageCond.Type == "Removed" || (cluster.Spec.RKEConfig != nil && messageCond.Type == "Ready") {
				// Don't copy these conditions from v3 management object to the v1 provisioning object. This is because we copy these specific conditions from other places and we need to prevent clobbering.
				// Updated - This is copied from the rkecontrolplane Ready condition to the v1 object by the rke2/provisioningcluster/controller.go via the reconcile function
				// Provisioned - This is copied from the rkecontrolplane Ready condition to the v1 object by the rke2/provisioningcluster/controller.go via the reconcile function
				// Removed - This is copied from the rkecontrolplane Removed condition on rkecontrolplane deletion.
				// Ready - Conditionally, if RKEConfig != nil. This is copied from the rkecontrolplane Ready condition (if cluster is not stable) or v3/management cluster Ready condition (if cluster is stable) by the rke2/provisioningcluster/controller.go via the reconcile function
				continue
			}

			found := false
			newCond := genericcondition.GenericCondition{
				Type:               string(messageCond.Type),
				Status:             messageCond.Status,
				LastUpdateTime:     messageCond.LastUpdateTime,
				LastTransitionTime: messageCond.LastTransitionTime,
				Reason:             messageCond.Reason,
				Message:            messageCond.Message,
			}
			for i, provCond := range status.Conditions {
				if provCond.Type != string(messageCond.Type) {
					continue
				}
				found = true
				status.Conditions[i] = newCond
			}
			if !found {
				status.Conditions = append(status.Conditions, newCond)
			}
		}
		status.AgentDeployed = v3.ClusterConditionAgentDeployed.IsTrue(existing)
	}

	// Never set ready back to false because we will end up deleting the secret
	status.Ready = status.Ready || ready
	status.ObservedGeneration = cluster.Generation
	status.ClusterName = rCluster.Name
	if status.Ready {
		kstatus.SetActive(&status)
	} else {
		kstatus.SetTransitioning(&status, "")
	}

	if status.Ready {
		secret, err := h.kubeconfigManager.GetKubeConfig(cluster, status)
		if err != nil {
			return nil, status, err
		}
		if secret != nil {
			status.ClientSecretName = secret.Name

			if features.MCM.Enabled() {
				crtb, err := h.kubeconfigManager.GetCRTBForClusterOwner(cluster, status)
				if err != nil {
					return nil, status, err
				}
				objs = append(objs, crtb)
			} else if cluster.Namespace == fleetconst.ClustersLocalNamespace && cluster.Name == "local" {
				user, err := h.kubeconfigManager.EnsureUser(cluster.Namespace, cluster.Name)
				if err != nil {
					return objs, status, err
				}
				objs = append(objs, &rbacv1.ClusterRoleBinding{
					ObjectMeta: metav1.ObjectMeta{
						Name: "fleet-local-cluster-admin",
					},
					Subjects: []rbacv1.Subject{{
						Kind:     "User",
						APIGroup: rbacv1.GroupName,
						Name:     user,
					}},
					RoleRef: rbacv1.RoleRef{
						Kind: "ClusterRole",
						Name: "cluster-admin",
					},
				})
			}
		}
	}

	return objs, status, nil
}

func (h *handler) updateFeatureLockedValue(lockValueToTrue bool) error {
	feature, err := h.featureCache.Get(features.RKE2.Name())
	if err != nil {
		return err
	}

	if feature.Status.LockedValue == nil && !lockValueToTrue || feature.Status.LockedValue != nil && *feature.Status.LockedValue == lockValueToTrue {
		return nil
	}

	feature = feature.DeepCopy()
	if lockValueToTrue {
		feature.Status.LockedValue = &lockValueToTrue
	} else {
		clusters, err := h.clusters.Cache().List("", labels.Everything())
		if err != nil {
			return err
		}

		for _, cluster := range clusters {
			if cluster.DeletionTimestamp.IsZero() && !h.isLegacyCluster(cluster) && cluster.Spec.RKEConfig != nil {
				return nil
			}
		}
		feature.Status.LockedValue = nil
	}

	_, err = h.featureClient.Update(feature)
	return err
}
