/*
Copyright The Kubernetes Authors.

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 controllers

import (
	"context"

	"github.com/awslabs/operatorpkg/controller"
	"github.com/awslabs/operatorpkg/object"
	"github.com/awslabs/operatorpkg/option"
	"github.com/awslabs/operatorpkg/status"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/samber/lo"
	"k8s.io/utils/clock"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/manager"

	"sigs.k8s.io/karpenter/pkg/state/virtualpods"

	corev1 "k8s.io/api/core/v1"

	autoscalingv1beta1 "sigs.k8s.io/karpenter/pkg/apis/autoscaling/v1beta1"
	v1 "sigs.k8s.io/karpenter/pkg/apis/v1"
	"sigs.k8s.io/karpenter/pkg/cloudprovider"
	"sigs.k8s.io/karpenter/pkg/controllers/capacitybuffer"
	"sigs.k8s.io/karpenter/pkg/controllers/disruption"
	"sigs.k8s.io/karpenter/pkg/controllers/dynamicresources/deviceallocation"
	metricsnode "sigs.k8s.io/karpenter/pkg/controllers/metrics/node"
	metricsnodepool "sigs.k8s.io/karpenter/pkg/controllers/metrics/nodepool"
	metricspod "sigs.k8s.io/karpenter/pkg/controllers/metrics/pod"
	nodehydration "sigs.k8s.io/karpenter/pkg/controllers/node/hydration"
	"sigs.k8s.io/karpenter/pkg/controllers/node/legacyrepair"
	"sigs.k8s.io/karpenter/pkg/controllers/node/termination"
	"sigs.k8s.io/karpenter/pkg/controllers/node/termination/terminator"
	nodeclaimconsistency "sigs.k8s.io/karpenter/pkg/controllers/nodeclaim/consistency"
	nodeclaimdisruption "sigs.k8s.io/karpenter/pkg/controllers/nodeclaim/disruption"
	"sigs.k8s.io/karpenter/pkg/controllers/nodeclaim/expiration"
	nodeclaimgarbagecollection "sigs.k8s.io/karpenter/pkg/controllers/nodeclaim/garbagecollection"
	nodeclaimhydration "sigs.k8s.io/karpenter/pkg/controllers/nodeclaim/hydration"
	nodeclaimlifecycle "sigs.k8s.io/karpenter/pkg/controllers/nodeclaim/lifecycle"
	"sigs.k8s.io/karpenter/pkg/controllers/nodeclaim/podevents"
	nodeclaimreboot "sigs.k8s.io/karpenter/pkg/controllers/nodeclaim/reboot"
	"sigs.k8s.io/karpenter/pkg/controllers/nodeoverlay"
	nodepoolcounter "sigs.k8s.io/karpenter/pkg/controllers/nodepool/counter"
	nodepoolhash "sigs.k8s.io/karpenter/pkg/controllers/nodepool/hash"
	nodepoolreadiness "sigs.k8s.io/karpenter/pkg/controllers/nodepool/readiness"
	nodepoolregistrationhealth "sigs.k8s.io/karpenter/pkg/controllers/nodepool/registrationhealth"
	nodepoolvalidation "sigs.k8s.io/karpenter/pkg/controllers/nodepool/validation"
	"sigs.k8s.io/karpenter/pkg/controllers/pod/deletioncost"
	"sigs.k8s.io/karpenter/pkg/controllers/provisioning"
	"sigs.k8s.io/karpenter/pkg/controllers/state"
	"sigs.k8s.io/karpenter/pkg/controllers/state/informer"
	statenodeclaimgc "sigs.k8s.io/karpenter/pkg/controllers/state/nodeclaimgc"
	staticdeprovisioning "sigs.k8s.io/karpenter/pkg/controllers/static/deprovisioning"
	staticprovisioning "sigs.k8s.io/karpenter/pkg/controllers/static/provisioning"
	"sigs.k8s.io/karpenter/pkg/events"
	"sigs.k8s.io/karpenter/pkg/operator/options"
	"sigs.k8s.io/karpenter/pkg/state/cost"
	"sigs.k8s.io/karpenter/pkg/state/nodepoolhealth"
	"sigs.k8s.io/karpenter/pkg/state/prediction"
)

type ControllerOptions struct {
	registrationHooks    []cloudprovider.NodeLifecycleHook
	disableVPAPrediction bool
}

// WithoutVPAPrediction disables the VPA prediction controller. Use this when
// a different prediction source is registered separately.
func WithoutVPAPrediction() option.Function[ControllerOptions] {
	return func(o *ControllerOptions) {
		o.disableVPAPrediction = true
	}
}

// WithRegistrationHook registers a hook that blocks Karpenter from marking a node as registered
// until the hook's preconditions are satisfied. This is useful when a cloud provider needs to
// apply well-known labels asynchronously after instance launch (e.g., capacity reservation labels
// used by topology spread constraints).
func WithRegistrationHook(hook cloudprovider.NodeLifecycleHook) option.Function[ControllerOptions] {
	return func(o *ControllerOptions) {
		o.registrationHooks = append(o.registrationHooks, hook)
	}
}

func NewControllers(
	ctx context.Context,
	mgr manager.Manager,
	clock clock.Clock,
	kubeClient client.Client,
	recorder events.Recorder,
	cloudProvider cloudprovider.CloudProvider,
	overlayUndecoratedCloudProvider cloudprovider.CloudProvider,
	cluster *state.Cluster,
	instanceTypeStore *nodeoverlay.InstanceTypeStore,
	predictionStore *prediction.Store,
	opts ...option.Function[ControllerOptions],
) []controller.Controller {
	o := option.Resolve(opts...)
	deviceAllocationController := deviceallocation.NewController(kubeClient)
	virtualPodCache := virtualpods.NewVirtualPodCache(kubeClient)
	p := provisioning.NewProvisioner(kubeClient, recorder, cloudProvider, cluster, clock, deviceAllocationController, virtualPodCache)
	evictionQueue := terminator.NewQueue(clock, kubeClient, recorder)
	disruptionQueue := disruption.NewQueue(kubeClient, recorder, cluster, clock, p)
	npState := nodepoolhealth.NewState()
	clusterCost := cost.NewClusterCost(ctx, cloudProvider, kubeClient)
	controllers := []controller.Controller{
		p, evictionQueue, disruptionQueue,
		disruption.NewController(ctx, clock, kubeClient, p, cloudProvider, recorder, cluster, disruptionQueue, clusterCost),
		provisioning.NewPodController(kubeClient, p, cluster),
		provisioning.NewNodeController(kubeClient, p),
		nodepoolhash.NewController(kubeClient, cloudProvider),
		expiration.NewController(clock, kubeClient, cloudProvider),
		informer.NewDaemonSetController(kubeClient, cluster),
		informer.NewNodeController(kubeClient, cluster),
		informer.NewPodController(kubeClient, cluster),
		informer.NewNodePoolController(kubeClient, cloudProvider, cluster, clusterCost),
		informer.NewNodeClaimController(kubeClient, cloudProvider, cluster, clusterCost),
		informer.NewPricingController(kubeClient, cloudProvider, clusterCost),
		statenodeclaimgc.NewController(kubeClient, cluster),
		termination.NewController(clock, kubeClient, cloudProvider, terminator.NewTerminator(clock, kubeClient, evictionQueue, recorder), recorder),
		nodepoolreadiness.NewController(clock, kubeClient, cloudProvider),
		nodepoolregistrationhealth.NewController(clock, kubeClient, cloudProvider, npState),
		nodepoolcounter.NewController(kubeClient, cloudProvider, cluster),
		nodepoolvalidation.NewController(clock, kubeClient, cloudProvider),
		podevents.NewController(clock, kubeClient, cloudProvider),
		nodeclaimconsistency.NewController(clock, kubeClient, cloudProvider, recorder),
		nodeclaimlifecycle.NewController(clock, kubeClient, cloudProvider, recorder, cluster, npState, o.registrationHooks),
		nodeclaimgarbagecollection.NewController(clock, kubeClient, cloudProvider),
		nodeclaimdisruption.NewController(clock, kubeClient, cloudProvider),
		nodeclaimhydration.NewController(kubeClient, cloudProvider),
		nodehydration.NewController(kubeClient, cloudProvider),
	}

	if !options.FromContext(ctx).IgnoreDRARequests {
		controllers = append(controllers, deviceAllocationController)
	}

	if !options.FromContext(ctx).DisableClusterStateObservability {
		controllers = append(controllers,
			metricspod.NewController(kubeClient, cluster),
			metricsnodepool.NewController(kubeClient, cloudProvider, clusterCost),
			metricsnode.NewController(cluster),
			status.NewController[*v1.NodeClaim](
				kubeClient,
				mgr.GetEventRecorderFor("karpenter"), //nolint:staticcheck // SA1019: will be replaced by mgr.GetEventRecorder once operatorpkg is updated
				status.EmitDeprecatedMetrics,
				status.WithHistogramBuckets(prometheus.ExponentialBuckets(0.5, 2, 15)), // 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192
				status.WithLabels(append(lo.Map(cloudProvider.GetSupportedNodeClasses(), func(obj status.Object, _ int) string { return v1.NodeClassLabelKey(object.GVK(obj).GroupKind()) }), v1.NodePoolLabelKey)...),
			),
			status.NewController[*v1.NodePool](
				kubeClient,
				mgr.GetEventRecorderFor("karpenter"), //nolint:staticcheck // SA1019: will be replaced by mgr.GetEventRecorder once operatorpkg is updated
				status.EmitDeprecatedMetrics,
				status.WithHistogramBuckets(prometheus.ExponentialBuckets(0.5, 2, 15)), // 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192
			),
			status.NewGenericObjectController[*corev1.Node](
				kubeClient,
				mgr.GetEventRecorderFor("karpenter"), //nolint:staticcheck // SA1019: will be replaced by mgr.GetEventRecorder once operatorpkg is updated
				status.WithHistogramBuckets(prometheus.ExponentialBuckets(0.5, 2, 15)), // 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192
				status.WithLabels(append(lo.Map(cloudProvider.GetSupportedNodeClasses(), func(obj status.Object, _ int) string { return v1.NodeClassLabelKey(object.GVK(obj).GroupKind()) }), v1.NodePoolLabelKey, v1.NodeInitializedLabelKey)...)),
		)
	}

	// The reboot controller only acts on NodeClaims with a committed reboot, so it always runs: a reboot already in flight
	// finishes even if node repair is disabled or the provider stops selecting reboot.
	controllers = append(controllers, nodeclaimreboot.NewController(clock, kubeClient, cloudProvider, terminator.NewTerminator(clock, kubeClient, evictionQueue, recorder), recorder))
	// The repair disruption method is registered in the disruption controller. --legacy-node-repair instead runs the
	// legacy node.health controller.
	if len(cloudProvider.RepairPolicies()) != 0 && options.FromContext(ctx).FeatureGates.NodeRepair && options.FromContext(ctx).LegacyNodeRepair {
		controllers = append(controllers, legacyrepair.NewController(kubeClient, cloudProvider, clock, recorder))
	}
	if options.FromContext(ctx).FeatureGates.StaticCapacity {
		controllers = append(controllers, staticprovisioning.NewController(kubeClient, cluster, recorder, cloudProvider, p, clock, deviceAllocationController, virtualPodCache))
		controllers = append(controllers, staticdeprovisioning.NewController(kubeClient, cluster, cloudProvider, clock, recorder))
	}

	if options.FromContext(ctx).FeatureGates.NodeOverlay {
		controllers = append(controllers, nodeoverlay.NewController(clock, kubeClient, overlayUndecoratedCloudProvider, instanceTypeStore, cluster))
	}

	if options.FromContext(ctx).FeatureGates.CapacityBuffer {
		controllers = append(controllers, capacitybuffer.NewController(kubeClient, p, virtualPodCache))
		if !options.FromContext(ctx).DisableClusterStateObservability {
			// Emit the standard operator_status_condition_* metrics for CapacityBuffer.
			// A GenericObjectController reads status.conditions reflectively, so the
			// upstream CapacityBuffer type is used as-is without implementing operatorpkg's
			// status.Object interface or persisting a Karpenter-specific Ready condition.
			controllers = append(controllers,
				status.NewGenericObjectController[*autoscalingv1beta1.CapacityBuffer](
					kubeClient,
					mgr.GetEventRecorderFor("karpenter"), //nolint:staticcheck // SA1019: will be replaced by mgr.GetEventRecorder once operatorpkg is updated
					status.EmitDeprecatedMetrics,
					status.WithHistogramBuckets(prometheus.ExponentialBuckets(0.5, 2, 15)), // 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192
				),
			)
		}
	}

	if options.FromContext(ctx).FeatureGates.PodDeletionCostManagement {
		deletionCostQueue := deletioncost.NewQueue(kubeClient)
		controllers = append(controllers,
			deletionCostQueue,
			deletioncost.NewController(clock, kubeClient, cloudProvider, cluster, deletionCostQueue),
		)
	}

	return controllers
}