// Copyright 2017 PingCAP, Inc.
//
// 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,
// See the License for the specific language governing permissions and
// limitations under the License.package spec

package manager

import (
	"fmt"
	"strings"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pborman/uuid"
	"github.com/pingcap/kvproto/pkg/metapb"
	tiapi "github.com/pingcap/tidb-cloud-manager/pkg/tidbinstance/api"
	tictl "github.com/pingcap/tidb-cloud-manager/pkg/tidbinstance/controller"
	opapi "github.com/pingcap/private-tidb-operator/pkg/api"
	tsapi "github.com/pingcap/private-tidb-operator/pkg/tidbset/api"
	outil "github.com/pingcap/private-tidb-operator/pkg/util"
	"github.com/pingcap/private-tidb-operator/pkg/util/label"
	"github.com/pingcap/private-tidb-operator/pkg/util/pdutil"
	"github.com/pingcap/private-tidb-operator/pkg/volumeprovisioner"
	"k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/conversion"
	"k8s.io/apimachinery/pkg/util/wait"
)

var (
	ControllerKind = opapi.SchemeGroupVersion.WithKind("TidbCluster")

	NodeUnreachablePodReason = "NodeLost"
)

func getVersion(name string) string {
	s := strings.Split(name, ":")
	if len(s) < 2 {
		return defaultVersion
	}
	return s[len(s)-1]
}

func getName(app string) string {
	return fmt.Sprintf("%s-%s", clusterName, app)
}

func affinityForNodeSelector(selector map[string]string) *v1.Affinity {
	affinity := &v1.Affinity{}
	if len(selector) == 0 {
		return affinity
	}
	affinity.NodeAffinity = &v1.NodeAffinity{
		// pod should be scheduled on node with exactly the kind label
		RequiredDuringSchedulingIgnoredDuringExecution: &v1.NodeSelector{
			NodeSelectorTerms: []v1.NodeSelectorTerm{
				{
					MatchExpressions: []v1.NodeSelectorRequirement{
						{
							Key:      "kind",
							Operator: v1.NodeSelectorOpIn,
							Values:   []string{selector["kind"]},
						},
					},
				},
			},
		},
	}
	return affinity
}

func makeRespPodSpec(podType tictl.PodType, instance *tiapi.TidbInstance) *tiapi.PodSpec {
	respPodSpec := &tiapi.PodSpec{}
	switch podType {
	case tictl.PDPodType:
		respPodSpec.Size = instance.Spec.PD.Size
		pdImage := tictl.GetImageByTiPodSpec(tictl.PDContainerType, instance.Spec.PD)
		respPodSpec.Containers = []tiapi.ContainerSpec{
			{
				Name:    string(tictl.PDContainerType),
				Image:   pdImage,
				Version: getVersion(pdImage),
			},
		}
	case tictl.TiKVPodType:
		respPodSpec.Size = instance.Spec.TiKV.Size
		tikvImage := tictl.GetImageByTiPodSpec(tictl.TiKVContainerType, instance.Spec.TiKV)
		respPodSpec.Containers = []tiapi.ContainerSpec{
			{
				Name:    string(tictl.TiKVContainerType),
				Image:   tikvImage,
				Version: getVersion(tikvImage),
			},
		}
	case tictl.TiDBPodType:
		respPodSpec.Size = instance.Spec.TiDB.Size
		tidbImage := tictl.GetImageByTiPodSpec(tictl.TiDBContainerType, instance.Spec.TiDB)
		respPodSpec.Containers = []tiapi.ContainerSpec{
			{
				Name:    string(tictl.TiDBContainerType),
				Image:   tidbImage,
				Version: getVersion(tidbImage),
			},
		}
		if tictl.HasBinlogEnabled(instance) {
			binlogImage := tictl.GetImageByTiPodSpec(tictl.BinlogContainerType, instance.Spec.TiDB)
			bcs := tiapi.ContainerSpec{
				Name:    string(tictl.BinlogContainerType),
				Image:   binlogImage,
				Version: getVersion(binlogImage),
			}
			respPodSpec.Containers = append(respPodSpec.Containers, bcs)
		}
	}
	return respPodSpec
}

func makePodStatus(pod *v1.Pod, containerType tictl.ContainerType, ts *tsapi.TidbSet) PodStatus {
	return PodStatus{
		Name:   pod.ObjectMeta.GetName(),
		PodIP:  pod.Status.PodIP,
		NodeIP: pod.Status.HostIP,
		Status: getRealStatus(pod, containerType, ts),
		Image:  GetImageByK8sPodSpec(containerType, &pod.Spec),
	}
}

func getRealStatus(pod *v1.Pod, containerType tictl.ContainerType, ts *tsapi.TidbSet) string {
	podState := parsePodStatus(pod)
	if podState != string(v1.PodRunning) {
		return podState
	}
	switch {
	case containerType == tictl.PDContainerType:
		if _, ok := ts.Status.PDStatus[pod.GetName()]; !ok {
			podState = string(v1.PodPending)
		}
	case containerType == tictl.TiKVContainerType:
		if tikvStatus, ok := ts.Status.TiKVStatus[pod.GetName()]; ok {
			if tikvStatus.State != "Up" {
				podState = string(v1.PodFailed)
			}
		}
	}
	return podState
}

// parse pod's status
func parsePodStatus(pod *v1.Pod) string {
	reason := string(pod.Status.Phase)
	if pod.Status.Reason != "" {
		reason = pod.Status.Reason
	}

	initializing := false
	for i := range pod.Status.InitContainerStatuses {
		container := pod.Status.InitContainerStatuses[i]
		switch {
		case container.State.Terminated != nil && container.State.Terminated.ExitCode == 0:
			continue
		case container.State.Terminated != nil:
			// initialization is failed
			if len(container.State.Terminated.Reason) == 0 {
				if container.State.Terminated.Signal != 0 {
					reason = fmt.Sprintf("Init:Signal:%d", container.State.Terminated.Signal)
				} else {
					reason = fmt.Sprintf("Init:ExitCode:%d", container.State.Terminated.ExitCode)
				}
			} else {
				reason = "Init:" + container.State.Terminated.Reason
			}
			initializing = true
		case container.State.Waiting != nil && len(container.State.Waiting.Reason) > 0 && container.State.Waiting.Reason != "PodInitializing":
			reason = "Init:" + container.State.Waiting.Reason
			initializing = true
		default:
			reason = fmt.Sprintf("Init:%d/%d", i, len(pod.Spec.InitContainers))
			initializing = true
		}
		break
	}
	if !initializing {
		for i := len(pod.Status.ContainerStatuses) - 1; i >= 0; i-- {
			container := pod.Status.ContainerStatuses[i]

			if container.State.Waiting != nil && container.State.Waiting.Reason != "" {
				reason = container.State.Waiting.Reason
			} else if container.State.Terminated != nil && container.State.Terminated.Reason != "" {
				reason = container.State.Terminated.Reason
			} else if container.State.Terminated != nil && container.State.Terminated.Reason == "" {
				if container.State.Terminated.Signal != 0 {
					reason = fmt.Sprintf("Signal:%d", container.State.Terminated.Signal)
				} else {
					reason = fmt.Sprintf("ExitCode:%d", container.State.Terminated.ExitCode)
				}
			}
		}
	}

	if pod.DeletionTimestamp != nil && pod.Status.Reason == NodeUnreachablePodReason {
		reason = "Unknown"
	} else if pod.DeletionTimestamp != nil {
		reason = "Terminating"
	}

	return reason
}

func (m Manager) revisePodSpec(podType tictl.PodType, podSpec *tiapi.PodSpec) *tiapi.PodSpec {
	if podType == tictl.MonitorPodType && (podSpec == nil || len(podSpec.Containers) != 3) {
		// Use the default monitor podSpec only when monitor pod's config is empty or
		// the number of containers specified is less than 3
		podSpec = makeDefaultMonitorPodSpec(podSpec)
	}

	if podType == tictl.TiKVPodType && !hasContainerExisted(tictl.PushgatewayContainerType, podSpec) {
		// the podSpec can't be nil when the podType is one of TiKV/TiDB/PD.
		pushContainer := tiapi.ContainerSpec{
			Name:  string(tictl.PushgatewayContainerType),
			Image: tiapi.PushgatewayImage,
		}
		podSpec.Containers = append(podSpec.Containers, pushContainer)
	}
	for i, c := range podSpec.Containers {
		switch tictl.ContainerType(c.Name) {
		case tictl.PDContainerType:
			podSpec.Containers[i].Image = generateImage(tictl.PDContainerType, c.Image, c.Version)
		case tictl.TiKVContainerType:
			podSpec.Containers[i].Image = generateImage(tictl.TiKVContainerType, c.Image, c.Version)
		case tictl.PushgatewayContainerType:
			podSpec.Containers[i].Image = generateImage(tictl.PushgatewayContainerType, c.Image, c.Version)
		case tictl.TiDBContainerType:
			podSpec.Containers[i].Image = generateImage(tictl.TiDBContainerType, c.Image, c.Version)
		case tictl.BinlogContainerType:
			podSpec.Containers[i].Image = generateImage(tictl.BinlogContainerType, c.Image, c.Version)
		case tictl.GrafanaContainerType:
			podSpec.Containers[i].Image = generateImage(tictl.GrafanaContainerType, c.Image, c.Version)
		case tictl.PrometheusContainerType:
			podSpec.Containers[i].Image = generateImage(tictl.PrometheusContainerType, c.Image, c.Version)
		case tictl.InstallerContainerType:
			podSpec.Containers[i].Image = generateImage(tictl.InstallerContainerType, c.Image, c.Version)
		}
	}
	return podSpec
}

func makeDefaultMonitorPodSpec(podSpec *tiapi.PodSpec) *tiapi.PodSpec {
	if podSpec == nil {
		return &tiapi.PodSpec{
			Size: 1,
			Containers: []tiapi.ContainerSpec{
				{
					Name:  string(tictl.GrafanaContainerType),
					Image: tiapi.GrafanaImage,
				},
				{
					Name:  string(tictl.PrometheusContainerType),
					Image: tiapi.PrometheusImage,
				},
				{
					Name:  string(tictl.InstallerContainerType),
					Image: tiapi.DashboardInstallerImage,
				},
			},
		}
	}
	if !hasContainerExisted(tictl.PrometheusContainerType, podSpec) {
		promContainer := tiapi.ContainerSpec{
			Name:  string(tictl.PrometheusContainerType),
			Image: tiapi.PrometheusImage,
		}
		podSpec.Containers = append(podSpec.Containers, promContainer)
	}
	if !hasContainerExisted(tictl.GrafanaContainerType, podSpec) {
		grafanaContainer := tiapi.ContainerSpec{
			Name:  string(tictl.GrafanaContainerType),
			Image: tiapi.GrafanaImage,
		}
		podSpec.Containers = append(podSpec.Containers, grafanaContainer)
	}
	if !hasContainerExisted(tictl.InstallerContainerType, podSpec) {
		installerContainer := tiapi.ContainerSpec{
			Name:  string(tictl.InstallerContainerType),
			Image: tiapi.DashboardInstallerImage,
		}
		podSpec.Containers = append(podSpec.Containers, installerContainer)
	}
	return podSpec
}

func getContainerSpec(ctype tictl.ContainerType, podSpec *tiapi.PodSpec) tiapi.ContainerSpec {
	cspec := tiapi.ContainerSpec{}
	for _, c := range podSpec.Containers {
		if string(ctype) == c.Name {
			cspec = c
			break
		}
	}
	return cspec
}

func GetImageByK8sPodSpec(ctype tictl.ContainerType, podSpec *v1.PodSpec) string {
	var image string
	for _, c := range podSpec.Containers {
		if string(ctype) == c.Name {
			image = c.Image
			break
		}
	}
	return image
}

func checkContainerSpec(ctype tictl.ContainerType, podSpec *tiapi.PodSpec) bool {
	var valid bool
	for _, c := range podSpec.Containers {
		if string(ctype) != c.Name {
			continue
		}
		if c.Version != "" || c.Image != "" {
			valid = true
		}
		break
	}
	return valid
}

func hasContainerExisted(ctype tictl.ContainerType, podSpec *tiapi.PodSpec) bool {
	var hasExisted bool
	if podSpec == nil {
		return hasExisted
	}
	for _, c := range podSpec.Containers {
		if string(ctype) == c.Name {
			hasExisted = true
			break
		}
	}
	return hasExisted
}

//Check whether the configurations for PD/TiDB/TiKV are correct
func CheckClusterConfig(cluster *Cluster) error {
	if !hasContainerExisted(tictl.PDContainerType, cluster.Pd) ||
		!hasContainerExisted(tictl.TiKVContainerType, cluster.Tikv) ||
		!hasContainerExisted(tictl.TiDBContainerType, cluster.Tidb) {
		return fmt.Errorf("PD, TiDB or TiKV are misconfigured, namespace: %s", cluster.Name)
	}
	return nil
}

func generateImage(ctype tictl.ContainerType, image, version string) string {
	if len(image) == 0 {
		// generate image from image version only when user not specify the container's image
		if version == "" {
			return fmt.Sprintf("%s/%s:%s", RepoPrefix, ctype, defaultVersion)
		}
		return fmt.Sprintf("%s/%s:%s", RepoPrefix, ctype, version)
	}
	return image
}

type NodesInfo struct {
	Nodes     []string
	UpdatedAt time.Time
}

var nodesInfo NodesInfo

func (m Manager) getLatestNodes() []string {
	if int(time.Now().Sub(nodesInfo.UpdatedAt).Seconds()) < 30 {
		return nodesInfo.Nodes
	}
	nodes, err := m.Cli.CoreV1().Nodes().List(metav1.ListOptions{})
	if err != nil {
		log.Errorf("failed to retrieve nodes: %v, error: %v", nodes, err)
		return nodesInfo.Nodes
	}
	nodesInfo.Nodes = []string{}
	for _, node := range nodes.Items {
		if node.Spec.Unschedulable {
			continue
		}
		for _, a := range node.Status.Addresses {
			if a.Type == v1.NodeInternalIP {
				nodesInfo.Nodes = append(nodesInfo.Nodes, a.Address)
			}
		}
	}
	nodesInfo.UpdatedAt = time.Now()
	return nodesInfo.Nodes
}

func (m Manager) getService(svc *v1.Service) Service {
	service := Service{
		ClusterIP:   svc.Spec.ClusterIP,
		ClusterPort: int(svc.Spec.Ports[0].Port),
	}
	if svc.Spec.Type == v1.ServiceTypeNodePort || svc.Spec.Type == v1.ServiceTypeLoadBalancer {
		service.NodePort = int(svc.Spec.Ports[0].NodePort)
		service.NodeIP = m.getLatestNodes()
	}
	if svc.Spec.Type == v1.ServiceTypeLoadBalancer && len(svc.Status.LoadBalancer.Ingress) > 0 {
		service.ExternalIP = svc.Status.LoadBalancer.Ingress[0].IP
	} else if svc.Spec.Type == v1.ServiceTypeLoadBalancer {
		log.Warnf("have not allocated ingress ip for loadBalancer, namespace: %s", svc.ObjectMeta.Namespace)
	}
	return service
}

func toFraction(val float64) string {
	return fmt.Sprintf("%d%%", int64(val))
}

func getPodsTotalRequestsAndLimits(podList *v1.PodList) (reqs map[v1.ResourceName]resource.Quantity, limits map[v1.ResourceName]resource.Quantity) {
	reqs, limits = map[v1.ResourceName]resource.Quantity{}, map[v1.ResourceName]resource.Quantity{}
	for _, pod := range podList.Items {
		podReqs, podLimits := podRequestsAndLimits(&pod)

		for podReqName, podReqValue := range podReqs {
			if value, ok := reqs[podReqName]; !ok {
				reqs[podReqName] = *podReqValue.Copy()
			} else {
				value.Add(podReqValue)
				reqs[podReqName] = value
			}
		}
		for podLimitName, podLimitValue := range podLimits {
			if value, ok := limits[podLimitName]; !ok {
				limits[podLimitName] = *podLimitValue.Copy()
			} else {
				value.Add(podLimitValue)
				limits[podLimitName] = value
			}
		}
	}
	return
}

func podRequestsAndLimits(pod *v1.Pod) (reqs map[v1.ResourceName]resource.Quantity, limits map[v1.ResourceName]resource.Quantity) {
	reqs, limits = map[v1.ResourceName]resource.Quantity{}, map[v1.ResourceName]resource.Quantity{}
	for _, container := range pod.Spec.Containers {
		for name, quantity := range container.Resources.Requests {
			if value, ok := reqs[name]; !ok {
				reqs[name] = *quantity.Copy()
			} else {
				value.Add(quantity)
				reqs[name] = value
			}
		}
		for name, quantity := range container.Resources.Limits {
			if value, ok := limits[name]; !ok {
				limits[name] = *quantity.Copy()
			} else {
				value.Add(quantity)
				limits[name] = value
			}
		}
	}
	// init containers define the minimum of any resource
	for _, container := range pod.Spec.InitContainers {
		for name, quantity := range container.Resources.Requests {
			value, ok := reqs[name]
			if !ok {
				reqs[name] = *quantity.Copy()
				continue
			}
			if quantity.Cmp(value) > 0 {
				reqs[name] = *quantity.Copy()
			}
		}
		for name, quantity := range container.Resources.Limits {
			value, ok := limits[name]
			if !ok {
				limits[name] = *quantity.Copy()
				continue
			}
			if quantity.Cmp(value) > 0 {
				limits[name] = *quantity.Copy()
			}
		}
	}
	return
}

// This function logic is copied from tikv controller
func (m Manager) addTiKVPod(tikvset *tsapi.TidbSet) error {
	ns := tikvset.Namespace
	name := tikvset.Name
	pvcName := fmt.Sprintf("pvc-%s-%s", name, uuid.New())
	storage := defaultTiKVRequestSize
	if tikvset.Annotations != nil {
		if storageFromAnn, ok := tikvset.Annotations[tsapi.AnnotationStorageSize]; ok {
			q, err := resource.ParseQuantity(storageFromAnn)
			if err != nil {
				return errors.Errorf("can't parse storage from [%s] when creating PVC %s for TiKVSet %s/%s: %v", storageFromAnn, pvcName, ns, name, err)
			}
			storage = q
		}
	}
	storageClassName := volumeprovisioner.StorageClassName
	pvcObj := &v1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Name:      pvcName,
			Namespace: ns,
			Labels:    label.New().Cluster(clusterName).Labels(),
		},
		Spec: v1.PersistentVolumeClaimSpec{
			AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
			Resources: v1.ResourceRequirements{
				Requests: v1.ResourceList{v1.ResourceStorage: storage},
			},
			StorageClassName: &storageClassName,
		},
	}
	pvc, err := m.Cli.CoreV1().PersistentVolumeClaims(ns).Create(pvcObj)
	if err != nil {
		log.Errorf("failed to create PVC %s/%s: %v", ns, pvcName, err)
		return err
	}
	uid := tikvset.UID
	controller := true
	ownerRef := metav1.OwnerReference{
		APIVersion: ControllerKind.GroupVersion().String(),
		Kind:       ControllerKind.Kind,
		Name:       name,
		UID:        uid,
		Controller: &controller,
	}
	obj, err := conversion.NewCloner().DeepCopy(tikvset.Spec.Template.Spec)
	if err != nil {
		log.Errorf("failed to deepcopy tikv pod spec from %s/%s: %v", tikvset.Namespace, tikvset.Name, err)
		return err
	}
	podSpec := obj.(v1.PodSpec)
	for _, vol := range podSpec.Volumes {
		if vol.Name == tsapi.TiDBVolumeName && vol.VolumeSource.PersistentVolumeClaim != nil {
			vol.VolumeSource.PersistentVolumeClaim.ClaimName = pvc.Name
			break
		}
	}
	podSpec.SchedulerName = outil.TiDBScheduler
	podObj := &v1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			GenerateName:    fmt.Sprintf("%s-tikv-", clusterName),
			Namespace:       ns,
			Labels:          tikvset.Spec.Selector.MatchLabels,
			OwnerReferences: []metav1.OwnerReference{ownerRef},
		},
		Spec: podSpec,
	}
	pod, err := m.Cli.CoreV1().Pods(ns).Create(podObj)
	if err != nil {
		log.Errorf("failed to create tikv pod %s/%s: %v", ns, podObj.Name, err)
		return err
	}
	// associate pvc with pod, used for gc
	defer func() {
		wait.Poll(1*time.Second, 5*time.Second, func() (bool, error) {
			p, err := m.Cli.CoreV1().PersistentVolumeClaims(ns).Get(pvcName, metav1.GetOptions{})
			if err != nil {
				log.Errorf("failed to get PVC %s/%s: %v", ns, pvcName, err)
				return false, nil
			}
			p.Annotations[label.AnnPodNameKey] = pod.Name
			if _, err = m.Cli.CoreV1().PersistentVolumeClaims(ns).Update(p); err != nil {
				log.Errorf("failed to update PVC %s/%s: %v", ns, pvcName, err)
				return false, nil
			}
			return true, nil
		})
	}()
	return nil
}

func getTikvStoresSummary(namespace string) (*TikvStoresSummary, error) {
	pdURL := fmt.Sprintf("http://%s.%s:2379", getName("pd"), namespace)
	pdClient := pdutil.New(pdURL, 2*time.Second)
	storesInfo, err := pdClient.GetStores()
	if err != nil {
		log.Errorf("failed to get stores info from pd, namespace: %s, error: %v", namespace, err)
		return nil, err
	}

	tss := &TikvStoresSummary{
		MinAvailableStorage: uint64(1<<64 - 1),
	}
	needMigrateStorage := uint64(0)
	for _, store := range storesInfo.Stores {
		usedStorage := uint64(store.Status.Capacity - store.Status.Available)
		if store.Store.GetState() == metapb.StoreState_Up {
			if tss.MaxUsedStorage < usedStorage {
				tss.MaxUsedStorage = usedStorage
			}
			if tss.MinAvailableStorage > uint64(store.Status.Available) {
				tss.MinAvailableStorage = uint64(store.Status.Available)
			}
			tss.Count++
			continue
		}
		if store.Store.GetState() == metapb.StoreState_Offline {
			//the store with offline status need migrate the remaining data
			needMigrateStorage += usedStorage
		}
	}
	if needMigrateStorage != 0 {
		averageMigrateStorage := needMigrateStorage / uint64(tss.Count)
		tss.MinAvailableStorage -= averageMigrateStorage
		tss.MaxUsedStorage += averageMigrateStorage
	}
	log.Debugf("namespace: %s, tikv stores summary %+v", namespace, tss)
	return tss, nil
}