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

import (
	"bytes"
	"fmt"
	"io/ioutil"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	"github.com/pingcap/tidb-operator/pkg/util"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	"github.com/pingcap/tidb-operator/pkg/util/marshal"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/serializer"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
	clientsetscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
)

var codecs serializer.CodecFactory

func init() {
	err := v1.AddToScheme(clientsetscheme.Scheme)
	if err != nil {
		panic(err)
	}
	codecs = clientsetscheme.Codecs
}

// Controller is the controller of volume provisioner controller
type Controller struct {
	kubeCli    kubernetes.Interface
	cli        versioned.Interface
	controller cache.Controller
	store      cache.Store
	queue      workqueue.RateLimitingInterface
	nodeName   string
}

// NewController creates a new Controller
func NewController(kubeCli kubernetes.Interface, cli versioned.Interface, nodeName string) *Controller {
	ctl := &Controller{
		kubeCli:  kubeCli,
		cli:      cli,
		queue:    workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "volumeprovisioner"),
		nodeName: nodeName}
	opts := metav1.ListOptions{LabelSelector: labels.SelectorFromSet(label.New().Labels()).String()}
	ctl.store, ctl.controller = cache.NewInformer(
		&cache.ListWatch{
			ListFunc: cache.ListFunc(func(_ metav1.ListOptions) (runtime.Object, error) {
				return ctl.kubeCli.CoreV1().PersistentVolumes().List(opts)
			}),
			WatchFunc: cache.WatchFunc(func(_ metav1.ListOptions) (watch.Interface, error) {
				return ctl.kubeCli.CoreV1().PersistentVolumes().Watch(opts)
			}),
		},
		&corev1.PersistentVolume{},
		30*time.Second,
		cache.ResourceEventHandlerFuncs{
			AddFunc: ctl.enqueuePV,
			UpdateFunc: func(old, cur interface{}) {
				ctl.enqueuePV(cur)
			},
			DeleteFunc: ctl.enqueuePV,
		},
	)
	return ctl
}

// Run method runs volume provisioner controller
func (ctl *Controller) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer ctl.queue.ShutDown()
	log.Infof("Starting PV manager controller")
	go ctl.controller.Run(wait.NeverStop)

	for i := 0; i < workers; i++ {
		go wait.Until(ctl.worker, time.Second, stopCh)
	}

	<-stopCh
	log.Infof("Shutting down PV manager controller")
}

func (ctl *Controller) worker() {
	for {
		func() {
			key, quit := ctl.queue.Get()
			if quit {
				return
			}
			defer ctl.queue.Done(key)
			if err := ctl.syncPV(key.(string)); err != nil {
				utilruntime.HandleError(fmt.Errorf("Error syncing PV %v, requeuing: %v", key.(string), err))
				ctl.queue.AddRateLimited(key)
			} else {
				ctl.queue.Forget(key)
			}
		}()
	}
}

func (ctl *Controller) enqueuePV(obj interface{}) {
	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	if err != nil {
		log.Errorf("can't get key for obj: %v, error: %v", obj, err)
		return
	}
	ctl.queue.Add(key)
}

func (ctl *Controller) syncPV(key string) error {
	startTime := time.Now()
	defer func() {
		log.Infof("Finished syncing PV[%s] (%v)", key, time.Since(startTime))
	}()

	obj, exists, err := ctl.store.GetByKey(key)
	if err != nil {
		return errors.Trace(err)
	}

	if !exists {
		return nil
	}

	pv, ok := obj.(*corev1.PersistentVolume)
	if !ok {
		return errors.Trace(fmt.Errorf("object %v is not a *corev1.PersistentVolume", obj))
	}
	if pv.Annotations == nil {
		return errors.Errorf("PV[%s] annotations is empty", pv.Name)
	}
	nodeName, ok := pv.Annotations[AnnProvisionerNode]
	if !ok || nodeName != ctl.nodeName {
		log.Debugf("PV[%s] is not on this node[%s]", pv.Name, ctl.nodeName)
		return nil
	}
	path, ok := pv.Annotations[AnnProvisionerHostPath]
	if !ok || path == "" {
		errMsg := fmt.Sprintf("PV[%s] hostpath annotation is empty", pv.Name)
		log.Error(errMsg)
		return errors.New(errMsg)
	}
	// This command is copied from kubernetes, it requires du command from coreutils
	// Uses the same niceness level as cadvisor.fs does when running du
	// Uses -B 1 to always scale to a blocksize of 1 byte
	output, err := exec.Command("nice", "-n", "19", "du", "-s", "-B", "1", path).CombinedOutput()
	if err != nil {
		log.Errorf("failed to get volume %s size: %v, %s", path, err, output)
		return errors.Trace(err)
	}
	used, err := resource.ParseQuantity(strings.Fields(string(output))[0])
	if err != nil {
		log.Errorf("failed to parse quantity %s: %v", output, err)
		return errors.Trace(err)
	}
	used.Format = resource.BinarySI
	pv.Annotations[AnnProvisionerSpaceUsed] = used.String()
	_, err = ctl.kubeCli.CoreV1().PersistentVolumes().Update(pv)
	if err != nil {
		log.Errorf("failed to update PV[%s] usage statics: %v", pv.Name, err)
		return errors.Trace(err)
	}
	log.Infof("PV[%s] used space updated successfully", pv.Name)

	return ctl.syncClusterInfo(pv)
}

func (ctl *Controller) syncClusterInfo(pv *corev1.PersistentVolume) error {
	ns := pv.Labels[label.NamespaceLabelKey]
	clusterName := pv.Labels[label.ClusterLabelKey]

	subDir := util.TiKVStoreDir
	annHostPath := pv.Annotations[AnnProvisionerHostPath]
	if kind, ok := pv.Labels[label.AppLabelKey]; ok && kind != "" {
		subDir = kind
	}
	dir := filepath.Join(annHostPath, subDir, "k8s-metadata")
	err := util.MkdirP(dir, 0644)
	if err != nil {
		return errors.Trace(err)
	}

	tc, err := ctl.cli.PingcapV1().TidbClusters(ns).Get(clusterName, metav1.GetOptions{})
	if err != nil {
		return errors.Trace(err)
	}
	tc.ResourceVersion = ""
	tc.Status = v1.ClusterStatus{}

	opts := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Namespace(ns).Cluster(clusterName).Labels(),
		).String(),
	}
	pvList, err := ctl.kubeCli.CoreV1().PersistentVolumes().List(opts)
	if err != nil {
		return errors.Trace(err)
	}

	pvList.ResourceVersion = ""
	pvNewList := corev1.PersistentVolumeList{
		TypeMeta: pvList.TypeMeta,
		ListMeta: pvList.ListMeta,
	}
	pvNewList.ResourceVersion = ""
	for _, pv := range pvList.Items {
		pv.ResourceVersion = ""
		pv.Status = corev1.PersistentVolumeStatus{}
		pvNewList.Items = append(pvNewList.Items, pv)
	}

	cmName := tc.Spec.ConfigMap
	cm, err := ctl.kubeCli.CoreV1().ConfigMaps(ns).Get(cmName, metav1.GetOptions{})
	if err != nil {
		return errors.Trace(err)
	}
	cm.ResourceVersion = ""

	err = syncTCInfo(tc, dir)
	if err != nil {
		return errors.Trace(err)
	}
	err = syncPVInfo(&pvNewList, dir)
	if err != nil {
		return errors.Trace(err)
	}
	return syncCMInfo(cm, dir)
}

func syncTCInfo(tc *v1.TidbCluster, dir string) error {
	fileName := filepath.Join(dir, "tidbcluster.yaml")
	newData, err := marshal.ToYamlForCodecs(tc, v1.SchemeGroupVersion, codecs)
	if err != nil {
		return errors.Trace(err)
	}

	return syncData(fileName, newData)
}

func syncPVInfo(pvList *corev1.PersistentVolumeList, dir string) error {
	fileName := filepath.Join(dir, "pv.yaml")
	newData, err := marshal.ToYamlForCodecs(pvList, corev1.SchemeGroupVersion, codecs)
	if err != nil {
		return errors.Trace(err)
	}

	return syncData(fileName, newData)
}

func syncCMInfo(cm *corev1.ConfigMap, dir string) error {
	fileName := filepath.Join(dir, "configmap.yaml")
	newData, err := marshal.ToYamlForCodecs(cm, corev1.SchemeGroupVersion, codecs)
	if err != nil {
		return errors.Trace(err)
	}

	return syncData(fileName, newData)
}
func syncData(fileName string, newData []byte) error {
	var oldData []byte
	var err error

	_, err = os.Stat(fileName)
	if err != nil && !os.IsNotExist(err) {
		return errors.Trace(err)
	}
	if err == nil {
		oldData, err = ioutil.ReadFile(fileName)
		if err != nil {
			log.Errorf("failed to readFile from %s", fileName)
			return errors.Trace(err)
		}
	}

	if bytes.Equal(newData, oldData) {
		return nil
	}

	err = ioutil.WriteFile(fileName, newData, 0644)
	if err != nil {
		log.Errorf("failed to writeFile %s", fileName)
		return errors.Trace(err)
	}

	return nil
}