// 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 (
	"fmt"
	"os/exec"
	"strings"
	"time"

	"k8s.io/api/core/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/private-tidb-operator/pkg/util/label"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
)

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

// NewController creates a new Controller
func NewController(cli kubernetes.Interface, nodeName string) *Controller {
	ctl := &Controller{cli: cli, queue: workqueue.New(), 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.cli.CoreV1().PersistentVolumes().List(opts)
			}),
			WatchFunc: cache.WatchFunc(func(_ metav1.ListOptions) (watch.Interface, error) {
				return ctl.cli.CoreV1().PersistentVolumes().Watch(opts)
			}),
		},
		&v1.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 {
				log.Error(err)
			}
		}()
	}
}

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.Now().Sub(startTime))
	}()

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

	if !exists {
		return nil
	}

	pv, ok := obj.(*v1.PersistentVolume)
	if !ok {
		return errors.Trace(fmt.Errorf("object %v is not a *v1.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.cli.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 nil
}