/*
Copyright 2018 The Volcano 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 validate

import (
	"fmt"
	"strings"

	admissionv1 "k8s.io/api/admission/v1"
	whv1 "k8s.io/api/admissionregistration/v1"
	v1 "k8s.io/api/core/v1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/validation"
	"k8s.io/apimachinery/pkg/util/validation/field"
	utilfeature "k8s.io/apiserver/pkg/util/feature"
	"k8s.io/klog/v2"
	k8score "k8s.io/kubernetes/pkg/apis/core"
	k8scorev1 "k8s.io/kubernetes/pkg/apis/core/v1"
	k8scorevalid "k8s.io/kubernetes/pkg/apis/core/validation"
	"k8s.io/kubernetes/pkg/capabilities"
	kubefeatures "k8s.io/kubernetes/pkg/features"

	"volcano.sh/apis/pkg/apis/batch/v1alpha1"
	schedulingv1beta1 "volcano.sh/apis/pkg/apis/scheduling/v1beta1"
	jobhelpers "volcano.sh/volcano/pkg/controllers/job/helpers"
	"volcano.sh/volcano/pkg/controllers/job/plugins"
	controllerMpi "volcano.sh/volcano/pkg/controllers/job/plugins/distributed-framework/mpi"
	"volcano.sh/volcano/pkg/webhooks/router"
	"volcano.sh/volcano/pkg/webhooks/schema"
	"volcano.sh/volcano/pkg/webhooks/util"
)

func init() {
	capabilities.Initialize(capabilities.Capabilities{
		AllowPrivileged: true,
		PrivilegedSources: capabilities.PrivilegedSources{
			HostNetworkSources: []string{},
			HostPIDSources:     []string{},
			HostIPCSources:     []string{},
		},
	})
	router.RegisterAdmission(service)
}

var service = &router.AdmissionService{
	Path: "/jobs/validate",
	Func: AdmitJobs,

	Config: config,

	ValidatingConfig: &whv1.ValidatingWebhookConfiguration{
		Webhooks: []whv1.ValidatingWebhook{{
			Name: "validatejob.volcano.sh",
			Rules: []whv1.RuleWithOperations{
				{
					Operations: []whv1.OperationType{whv1.Create, whv1.Update},
					Rule: whv1.Rule{
						APIGroups:   []string{"batch.volcano.sh"},
						APIVersions: []string{"v1alpha1"},
						Resources:   []string{"jobs"},
					},
				},
			},
		}},
	},
}

var config = &router.AdmissionServiceConfig{}

// AdmitJobs is to admit jobs and return response.
func AdmitJobs(ar admissionv1.AdmissionReview) *admissionv1.AdmissionResponse {
	klog.V(3).Infof("admitting jobs -- %s", ar.Request.Operation)

	job, err := schema.DecodeJob(ar.Request.Object, ar.Request.Resource)
	if err != nil {
		return util.ToAdmissionResponse(err)
	}
	var msg string
	reviewResponse := admissionv1.AdmissionResponse{}
	reviewResponse.Allowed = true

	switch ar.Request.Operation {
	case admissionv1.Create:
		msg = validateJobCreate(job, &reviewResponse)
	case admissionv1.Update:
		oldJob, err := schema.DecodeJob(ar.Request.OldObject, ar.Request.Resource)
		if err != nil {
			return util.ToAdmissionResponse(err)
		}
		err = validateJobUpdate(oldJob, job)
		if err != nil {
			return util.ToAdmissionResponse(err)
		}
	default:
		err := fmt.Errorf("expect operation to be 'CREATE' or 'UPDATE'")
		return util.ToAdmissionResponse(err)
	}

	if !reviewResponse.Allowed {
		reviewResponse.Result = &metav1.Status{Message: strings.TrimSpace(msg)}
	}
	return &reviewResponse
}

func validateJobCreate(job *v1alpha1.Job, reviewResponse *admissionv1.AdmissionResponse) string {
	var b strings.Builder
	taskNames := map[string]struct{}{}
	var totalReplicas int32

	// Note: Basic validations like minAvailable >= 0, maxRetry >= 0, TTLSecondsAfterFinished >= 0,
	// task.Replicas >= 0, task.MinAvailable >= 0, task.MinAvailable <= task.Replicas,
	// job.MinAvailable <= totalReplicas, duplicate task name, task name DNS1123 format,
	// These validations have been removed from webhook to avoid duplication.
	if len(job.Spec.Tasks) == 0 {
		reviewResponse.Allowed = false
		return "No task specified in job spec"
	}

	if _, ok := job.Spec.Plugins[controllerMpi.MPIPluginName]; ok {
		mp := controllerMpi.NewInstance(job.Spec.Plugins[controllerMpi.MPIPluginName])
		masterIndex := jobhelpers.GetTaskIndexUnderJob(mp.GetMasterName(), job)
		if masterIndex == -1 {
			reviewResponse.Allowed = false
			return "The specified mpi master task was not found"
		}
	}

	b.WriteString(validateNetworkTopology(job.Spec.NetworkTopology))
	hasDependenciesBetweenTasks := false
	for index, task := range job.Spec.Tasks {
		if task.DependsOn != nil {
			hasDependenciesBetweenTasks = true
		}

		// Validate minAvailable and partitionPolicy.minPartitions mutual exclusivity
		// This validation is not covered by CRD schema or VAP, so it remains in webhook.
		if task.MinAvailable != nil {
			if *task.MinAvailable > task.Replicas {
				fmt.Fprintf(&b, " 'minAvailable' is greater than 'replicas' in task: %s, job: %s;", task.Name, job.Name)
			}
		}
		// count replicas
		totalReplicas += task.Replicas

		// duplicate task name
		if _, found := taskNames[task.Name]; found {
			fmt.Fprintf(&b, " duplicated task name %s;", task.Name)
			break
		} else {
			taskNames[task.Name] = struct{}{}
		}
		if err := validatePolicies(task.Policies, field.NewPath("spec.tasks.policies")); err != nil {
			b.WriteString(err.Error())
			fmt.Fprintf(&b, " valid events are %v, valid actions are %v;",
				getValidEvents(), getValidActions())
		}
		podName := jobhelpers.MakePodName(job.Name, task.Name, index)
		b.WriteString(validateK8sPodNameLength(podName))
		b.WriteString(validateTaskTemplate(task, job, index))
		b.WriteString(validatePartitionPolicy(task, job))
	}

	b.WriteString(validateJobName(job))

	if totalReplicas < job.Spec.MinAvailable {
		b.WriteString(" job 'minAvailable' should not be greater than total replicas in tasks;")
	}
	if err := validatePolicies(job.Spec.Policies, field.NewPath("spec.policies")); err != nil {
		b.WriteString(err.Error())
		fmt.Fprintf(&b, " valid events are %v, valid actions are %v;",
			getValidEvents(), getValidActions())
	}

	// invalid job plugins
	if len(job.Spec.Plugins) != 0 {
		for name := range job.Spec.Plugins {
			if _, found := plugins.GetPluginBuilder(name); !found {
				fmt.Fprintf(&b, " unable to find job plugin: %s;", name)
			}
		}
	}

	if err := validateIO(job.Spec.Volumes); err != nil {
		b.WriteString(err.Error())
	}

	queue, err := config.QueueLister.Get(job.Spec.Queue)
	if err != nil {
		fmt.Fprintf(&b, " unable to find job queue: %v;", err)
	} else {
		if queue.Status.State != schedulingv1beta1.QueueStateOpen {
			fmt.Fprintf(&b, " can only submit job to queue with state `Open`, "+
				"queue `%s` status is `%s`;", queue.Name, queue.Status.State)
		}

		// validate hierarchical queue
		if queue.Name == "root" {
			b.WriteString(" can not submit job to root queue;")
		} else {
			childQueues, err := config.GetQueuesByParent(queue.Name)
			if err != nil {
				fmt.Fprintf(&b, "failed to get child queues for queue %s: %v;", queue.Name, err)
			}
			if len(childQueues) > 0 {
				fmt.Fprintf(&b, " can only submit job to leaf queue, "+"queue `%s` has %d child queues;", queue.Name, len(childQueues))
			}
		}
	}

	if hasDependenciesBetweenTasks {
		_, isDag := topoSort(job)
		if !isDag {
			b.WriteString(" job has dependencies between tasks, but doesn't form a directed acyclic graph(DAG);")
		}
	}

	if b.Len() > 0 {
		reviewResponse.Allowed = false
	}

	return b.String()
}

func validateJobUpdate(old, new *v1alpha1.Job) error {
	var totalReplicas int32
	for _, task := range new.Spec.Tasks {
		if task.Replicas < 0 {
			return fmt.Errorf("'replicas' must be >= 0 in task: %s", task.Name)
		}

		if task.MinAvailable != nil {
			if *task.MinAvailable < 0 {
				return fmt.Errorf("'minAvailable' must be >= 0 in task: %s", task.Name)
			} else if *task.MinAvailable > task.Replicas {
				return fmt.Errorf("'minAvailable' must be <= 'replicas' in task: %s", task.Name)
			}
		}
		msg := validatePartitionPolicy(task, new)
		if msg != "" {
			return fmt.Errorf("%s", msg)
		}

		// count replicas
		totalReplicas += task.Replicas
	}
	if new.Spec.MinAvailable > totalReplicas {
		return fmt.Errorf("job 'minAvailable' must not be greater than total replicas")
	}
	if new.Spec.MinAvailable < 0 {
		return fmt.Errorf("job 'minAvailable' must be >= 0")
	}
	networkTopology := new.Spec.NetworkTopology
	if networkTopology != nil && networkTopology.HighestTierAllowed != nil && networkTopology.HighestTierName != "" {
		return fmt.Errorf("must not specify 'highestTierAllowed' and 'highestTierName' in networkTopology simultaneously")
	}

	if len(old.Spec.Tasks) != len(new.Spec.Tasks) {
		return fmt.Errorf("job updates may not add or remove tasks")
	}
	// other fields under spec are not allowed to mutate
	new.Spec.MinAvailable = old.Spec.MinAvailable
	new.Spec.PriorityClassName = old.Spec.PriorityClassName

	// K8S also permit mutating spec.schedulingGates
	// We do not support this for vcjob  (More details in design doc pod-scheduling-readiness.md)

	for i := range new.Spec.Tasks {
		new.Spec.Tasks[i].Replicas = old.Spec.Tasks[i].Replicas
		new.Spec.Tasks[i].MinAvailable = old.Spec.Tasks[i].MinAvailable
	}

	// job controller will update the pvc name if not provided
	for i := range new.Spec.Volumes {
		if new.Spec.Volumes[i].VolumeClaim != nil {
			new.Spec.Volumes[i].VolumeClaimName = ""
		}
	}
	for i := range old.Spec.Volumes {
		if old.Spec.Volumes[i].VolumeClaim != nil {
			old.Spec.Volumes[i].VolumeClaimName = ""
		}
	}

	if !apiequality.Semantic.DeepEqual(new.Spec, old.Spec) {
		return fmt.Errorf("job updates may not change fields other than `minAvailable`, `tasks[*].replicas under spec` and `PriorityClassName`")
	}

	return nil
}

func validatePartitionPolicy(task v1alpha1.TaskSpec, job *v1alpha1.Job) string {
	var msg string
	if task.PartitionPolicy != nil {
		if task.PartitionPolicy.TotalPartitions <= 0 {
			msg += fmt.Sprintf("'TotalPartitions' must be greater than 0 in task: %s, job: %s", task.Name, job.Name)
		} else if task.PartitionPolicy.PartitionSize <= 0 {
			msg += fmt.Sprintf("'PartitionSize' must be greater than 0 in task: %s, job: %s", task.Name, job.Name)
		} else if task.Replicas != task.PartitionPolicy.TotalPartitions*task.PartitionPolicy.PartitionSize {
			msg += fmt.Sprintf("'Replicas' are not equal to TotalPartitions*PartitionSize in task: %s, job: %s", task.Name, job.Name)
		} else if task.MinAvailable != nil && task.PartitionPolicy.MinPartitions > 0 && task.PartitionPolicy.MinPartitions*task.PartitionPolicy.PartitionSize != *task.MinAvailable {
			// Only enforce the minAvailable == minPartitions*partitionSize relationship
			// when minPartitions is explicitly set (> 0). When minPartitions is omitted
			// (defaults to 0), minAvailable falls back to replicas, so skip this check.
			msg += fmt.Sprintf("'MinAvailable' is not equal to MinPartitions*PartitionSize in task: %s, job: %s", task.Name, job.Name)
		}
		msg += validateNetworkTopology(task.PartitionPolicy.NetworkTopology)
	}

	return msg
}

func validateNetworkTopology(networkTopology *v1alpha1.NetworkTopologySpec) string {
	if networkTopology != nil && networkTopology.HighestTierAllowed != nil && networkTopology.HighestTierName != "" {
		return "must not specify 'highestTierAllowed' and 'highestTierName' in networkTopology simultaneously"
	}
	return ""
}

func validateTaskTemplate(task v1alpha1.TaskSpec, job *v1alpha1.Job, index int) string {
	var v1PodTemplate v1.PodTemplate
	v1PodTemplate.Template = *task.Template.DeepCopy()
	k8scorev1.SetObjectDefaults_PodTemplate(&v1PodTemplate)

	var coreTemplateSpec k8score.PodTemplateSpec
	k8scorev1.Convert_v1_PodTemplateSpec_To_core_PodTemplateSpec(&v1PodTemplate.Template, &coreTemplateSpec, nil)

	corePodTemplate := k8score.PodTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      task.Name,
			Namespace: job.Namespace,
		},
		Template: coreTemplateSpec,
	}

	opts := k8scorevalid.PodValidationOptions{
		PodLevelResourcesEnabled: utilfeature.DefaultFeatureGate.Enabled(kubefeatures.PodLevelResources),
	}
	if allErrs := k8scorevalid.ValidatePodTemplate(&corePodTemplate, opts); len(allErrs) > 0 {
		msg := fmt.Sprintf("spec.task[%d].", index)
		for index := range allErrs {
			msg += allErrs[index].Error() + ". "
		}
		return msg
	}

	msg := validateTaskTopoPolicy(task, index)
	return msg
}

func validateK8sPodNameLength(podName string) string {
	if errMsgs := validation.IsQualifiedName(podName); len(errMsgs) > 0 {
		return fmt.Sprintf("create pod with name %s validate failed %v;", podName, errMsgs)
	}
	return ""
}

func validateJobName(job *v1alpha1.Job) string {
	if errMsgs := validation.IsQualifiedName(job.Name); len(errMsgs) > 0 {
		return fmt.Sprintf("create job with name %s validate failed %v", job.Name, errMsgs)
	}
	return ""
}

func validateTaskTopoPolicy(task v1alpha1.TaskSpec, index int) string {
	if task.TopologyPolicy == "" || task.TopologyPolicy == v1alpha1.None {
		return ""
	}

	template := task.Template.DeepCopy()

	for id, container := range template.Spec.Containers {
		if len(container.Resources.Requests) == 0 {
			template.Spec.Containers[id].Resources.Requests = container.Resources.Limits.DeepCopy()
		}
	}

	for id, container := range template.Spec.InitContainers {
		if len(container.Resources.Requests) == 0 {
			template.Spec.InitContainers[id].Resources.Requests = container.Resources.Limits.DeepCopy()
		}
	}

	for id, container := range append(template.Spec.Containers, template.Spec.InitContainers...) {
		requestNum := guaranteedCPUs(container)
		if requestNum == 0 {
			return fmt.Sprintf("the cpu request isn't an integer in spec.task[%d] container[%d].",
				index, id)
		}
	}

	return ""
}

func guaranteedCPUs(container v1.Container) int {
	cpuQuantity := container.Resources.Requests[v1.ResourceCPU]
	if cpuQuantity.Value()*1000 != cpuQuantity.MilliValue() {
		return 0
	}

	return int(cpuQuantity.Value())
}
