// 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 controller

import (
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	tsctl "github.com/pingcap/tidb-operator/pkg/tidbset/controller"
	"github.com/pingcap/tidb-operator/pkg/util"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	corev1 "k8s.io/api/core/v1"
	apierrs "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/tools/cache"
)

const (
	timeout = 2 * time.Second
)

func (syncc *SyncController) syncPod(key string) error {
	startTime := time.Now()
	defer func() {
		log.Infof("Finished syncing Pod[%s] (%v)", key, time.Since(startTime))
	}()

	obj, exists, keyErr := syncc.podStore.GetByKey(key)
	if keyErr != nil {
		return errors.Trace(keyErr)
	}

	ns, podName, nsKeyErr := cache.SplitMetaNamespaceKey(key)
	if nsKeyErr != nil {
		return errors.Trace(nsKeyErr)
	}

	if !exists {
		return nil
	}

	pod, ok := obj.(*corev1.Pod)
	if !ok {
		return errors.Trace(fmt.Errorf("object %v is not a corev1.Pod", obj))
	}
	if err := syncc.syncPodLabels(pod); err != nil {
		return errors.Trace(err)
	}

	var tsctlRef *metav1.OwnerReference

	for _, or := range pod.ObjectMeta.OwnerReferences {
		if or.Kind == tsctl.ControllerKind.Kind {
			tsctlRef = &or
			break
		}
	}

	if tsctlRef == nil {
		log.Debugf("Pod: [%s/%s] doesn't have an %s OwnerReference, skipping", pod.Namespace, pod.Name, tsctl.ControllerKind.Kind)
		return nil
	}

	tsName := tsctlRef.Name
	var claimName string

	for _, vol := range pod.Spec.Volumes {
		if vol.Name == v1.TiDBVolumeName && vol.PersistentVolumeClaim != nil {
			claimName = vol.PersistentVolumeClaim.ClaimName
			break
		}
	}

	if len(claimName) != 0 {
		v1 := syncc.kubeCli.CoreV1()
		pvc := v1.PersistentVolumeClaims(ns)
		if _, err := pvc.Get(claimName, metav1.GetOptions{}); err != nil {
			// If PVC doesn't exist, delete pod
			if apierrs.IsNotFound(err) {
				log.Infof("Pod [%s/%s]'s PVC: [%s/%s] doesn't exists, deleting", ns, podName, ns, claimName)
				podDelErr := v1.Pods(ns).Delete(podName, nil)
				if podDelErr != nil {
					log.Errorf("failed to delete Pod: [%s/%s], %v", ns, podName, podDelErr)
					return errors.Trace(err)
				}

				log.Infof("Pod: [%s/%s] deleted successfully.", ns, podName)
				return nil
			}
			return errors.Trace(err)
		}
	}

	// If TidbSet doesn't exist, delete PVC(if have) and pod
	_, err := syncc.cli.PingcapV1().TidbSets(ns).Get(tsName, metav1.GetOptions{})
	if err != nil && apierrs.IsNotFound(err) {
		log.Infof("TidbSet: [%s/%s] was deleted, deleting all its pvcs and pods", ns, tsName)

		if len(claimName) != 0 {
			log.Infof("Pod [%s/%s] have a related PVC: [%s/%s], deleting PVC first", ns, podName, ns, claimName)
			delErr := syncc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Delete(claimName, nil)
			if delErr != nil && !apierrs.IsNotFound(err) {
				log.Errorf("Pod: [%s/%s]'s PVC: [%s/%s] deleted failed: %v", ns, podName, ns, claimName, delErr)
				return errors.Trace(delErr)
			}

			log.Infof("Pod: [%s/%s]'s PVC[%s/%s] deleted successfully.", ns, podName, ns, claimName)
		}

		err = syncc.kubeCli.CoreV1().Pods(ns).Delete(podName, nil)
		if err != nil {
			log.Errorf("failed to delete Pod: [%s/%s], %v", ns, podName, err)
			return errors.Trace(err)
		}
		log.Infof("Pod: [%s/%s] deleted successfully.", ns, podName)
	}

	return err
}

func (syncc *SyncController) syncPodLabels(pod *corev1.Pod) error {
	ns := pod.GetNamespace()
	podName := pod.GetName()
	labels := pod.GetLabels()
	var clusterID, storeID, memberID string
	if labels == nil {
		return errors.Errorf("pod %s/%s has empty labels", ns, podName)
	}
	clusterName, ok := labels[label.ClusterLabelKey]
	if !ok {
		return errors.Errorf("pod %s/%s doesn't have %s label", ns, podName, label.ClusterLabelKey)
	}
	pdCli := util.NewPDClientForCluster(ns, clusterName, timeout)
	if labels[label.ClusterIDLabelKey] == "" {
		cluster, err := pdCli.GetCluster()
		if err != nil {
			return errors.Trace(err)
		}
		clusterID = strconv.FormatUint(cluster.Id, 10)
	}

	app := labels[label.AppLabelKey]
	switch app {
	case label.PDLabelVal:
		if labels[label.MemberIDLabelKey] == "" {
			// get member id
			members, err := pdCli.GetMembers()
			if err != nil {
				return errors.Trace(err)
			}
			for _, member := range members.Members {
				if member.GetName() == podName {
					memberID = strconv.FormatUint(member.GetMemberId(), 10)
					break
				}
			}
		}
	case label.TiKVLabelVal:
		if labels[label.StoreIDLabelKey] == "" {
			// get store id
			stores, err := pdCli.GetStores()
			if err != nil {
				return errors.Trace(err)
			}
			for _, store := range stores.Stores {
				addr := store.Store.GetAddress()
				if strings.Split(addr, ".")[0] == podName {
					storeID = strconv.FormatUint(store.Store.GetId(), 10)
					break
				}
			}
		}
	}
	if labels[label.ClusterIDLabelKey] == clusterID &&
		labels[label.MemberIDLabelKey] == memberID &&
		labels[label.StoreIDLabelKey] == storeID {
		log.Debugf("pod %s/%s already has cluster labels set, skipping", ns, podName)
		return nil
	}
	// labels is a pointer, modify labels will modify pod.Labels
	if clusterID != "" {
		labels[label.ClusterIDLabelKey] = clusterID
	}
	if memberID != "" {
		labels[label.MemberIDLabelKey] = memberID
	}
	if storeID != "" {
		labels[label.StoreIDLabelKey] = storeID
	}

	_, err := syncc.kubeCli.CoreV1().Pods(ns).Update(pod)
	if err != nil {
		log.Errorf("failed to update pod %s/%s with cluster labels %v: %v", ns, podName, labels, err)
		return errors.Trace(err)
	}
	log.Infof("update pod %s/%s with cluster labels %v successfully", ns, podName, labels)
	return nil
}