// 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"
	"path/filepath"
	"strings"

	"k8s.io/api/core/v1"

	"github.com/juju/errors"
	"github.com/kubernetes-incubator/external-storage/lib/controller"
	"github.com/ngaut/log"
	"github.com/pingcap/private-tidb-operator/pkg/client"
	"github.com/pingcap/private-tidb-operator/pkg/util"
	"github.com/pingcap/private-tidb-operator/pkg/util/label"
	apierrs "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
	// StorageClassName is the name of tidb-volume-provisioner StorageClass
	StorageClassName = "pingcap-volume-provisioner"

	// ProvisionerName is the value of tidb-volume-provisioner StorageClass
	ProvisionerName = "pingcap.com/volume-provisioner"

	// AnnProvisionerNode is the annotation key of PVC stands for node name
	AnnProvisionerNode = "volume-provisioner.pingcap.com/node"

	// AnnProvisionerHostPath is the annotation key of PVC stands for hostPath
	AnnProvisionerHostPath = "volume-provisioner.pingcap.com/hostpath"

	// AnnProvisionerSpaceUsed is the annotation key of space used of the PV
	AnnProvisionerSpaceUsed = "volume-provisioner.pingcap.com/used"

	// AnnShouldDeletedPVName is the annotation key of PV should be deleted before restoring a new PV
	AnnShouldDeletedPVName = "volume-provisioner.pingcap.com/should-deleted-pvname"

	// AnnPVTombstoned is pvTomstoned annotations key
	AnnPVTombstoned string = "volume.pingcap.com/pvTombstoned"

	// AnnPVManualDeleted is pvManualDeleted annotations key
	AnnPVManualDeleted string = "volume.pingcap.com/pvManualDeleted"

	// AnnPVClusterDeleted is pvClusterDeleted annotations key
	AnnPVClusterDeleted string = "volume.pingcap.com/pvClusterDeleted"
)

type volumeProvisioner struct {
	// The directory to create PV-backing directories in
	pvDir    string
	nodeName string
	cli      client.Interface
}

var _ controller.Provisioner = &volumeProvisioner{}

// NewVolumeProvisioner creates a new volume provisioner
func NewVolumeProvisioner(pvDir, nodeName string, cli client.Interface) controller.Provisioner {
	return &volumeProvisioner{
		pvDir:    pvDir,
		nodeName: nodeName,
		cli:      cli,
	}
}

// Provision creates a storage asset and returns a PV object representing it.
func (p *volumeProvisioner) Provision(options controller.VolumeOptions) (*v1.PersistentVolume, error) {
	pvc := options.PVC
	ns := pvc.GetNamespace()
	name := pvc.GetName()

	// TODO drive upstream to support IgnoredError
	annNode, ok := pvc.Annotations[AnnProvisionerNode]
	if !ok || annNode != p.nodeName {
		return nil, &controller.IgnoredError{fmt.Sprintf("PVC[%s/%s] is not managed by this provisioner, skipping", pvc.Namespace, pvc.Name)}
	}

	annHostPath, ok := pvc.Annotations[AnnProvisionerHostPath]
	if !ok || len(annHostPath) == 0 {
		return nil, errors.Errorf("PVC[%s/%s] hostPath is empty, skipping", ns, name)
	}

	if !strings.HasPrefix(annHostPath, p.pvDir) {
		return nil, errors.Errorf("PVC[%s/%s] hostPath: %s doesn't begin with: %s", ns, name, annHostPath, p.pvDir)
	}

	podName, ok := pvc.Annotations[label.AnnPodNameKey]
	if !ok || len(podName) == 0 {
		return nil, errors.Errorf("PVC: [%s/%s] don't have annotations: [%s]", ns, name, label.AnnPodNameKey)
	}

	_, err := p.cli.CoreV1().Pods(ns).Get(podName, metav1.GetOptions{})
	if err != nil {
		return nil, errors.Trace(err)
	}

	pvLabels := pvc.Labels
	pvLabels[label.NamespaceLabelKey] = ns

	shouldDeletedPVName := pvc.Annotations[AnnShouldDeletedPVName]
	if len(shouldDeletedPVName) != 0 {
		err := p.cli.CoreV1().PersistentVolumes().Delete(shouldDeletedPVName, nil)
		if err != nil && !apierrs.IsNotFound(err) {
			log.Errorf("failed to delete PV: [%s], %v", shouldDeletedPVName, err)
			return nil, errors.Trace(err)
		}
		log.Infof("PV: [%s] deleted successfully", shouldDeletedPVName)
	}

	log.Infof("PV: [%s] bounding to PVC: [%s/%s] for Pod: [%s/%s]", options.PVName, ns, name, ns, podName)

	pv := &v1.PersistentVolume{
		ObjectMeta: metav1.ObjectMeta{
			Name: options.PVName,
			Annotations: map[string]string{
				AnnProvisionerNode:     annNode,
				AnnProvisionerHostPath: annHostPath,
				label.AnnPodNameKey:    podName,
			},
			Labels: pvLabels,
		},
		Spec: v1.PersistentVolumeSpec{
			PersistentVolumeReclaimPolicy: options.PersistentVolumeReclaimPolicy,
			AccessModes:                   options.PVC.Spec.AccessModes,
			Capacity: v1.ResourceList{
				v1.ResourceName(v1.ResourceStorage): options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)],
			},
			PersistentVolumeSource: v1.PersistentVolumeSource{
				HostPath: &v1.HostPathVolumeSource{
					Path: annHostPath,
				},
			},
		},
	}

	return pv, nil
}

// Delete removes the storage asset that was created by Provision represented
// by the given PV.
func (p *volumeProvisioner) Delete(pv *v1.PersistentVolume) error {
	pvName := pv.GetName()
	annNode, ok := pv.Annotations[AnnProvisionerNode]
	if !ok || annNode != p.nodeName {
		return &controller.IgnoredError{fmt.Sprintf("PV[%s] is not managed by this provisioner, managed by: %s,  skipping", pvName, annNode)}
	}

	annHostPath, ok := pv.Annotations[AnnProvisionerHostPath]
	if !ok || len(annHostPath) == 0 {
		return errors.Errorf("PV[%s] hostPath: %s doesn't exists, skipping", pvName, annHostPath)
	}

	if !strings.HasPrefix(annHostPath, p.pvDir) {
		return errors.Errorf("PV[%s] hostPath: %s doesn't begin with: %s, skipping", pvName, annHostPath, p.pvDir)
	}

	pvManualDeleted := pv.Annotations[AnnPVManualDeleted]
	pvTombstoned := pv.Annotations[AnnPVTombstoned]
	pvClusterDeleted := pv.Annotations[AnnPVClusterDeleted]

	if pvCanManualDeleted(pv) {
		log.Infof("PV: %s pvManualDeleted at: %s, pvTombstoned at: %s and pvClusterDeleted at: %s",
			pvName, pvManualDeleted, pvTombstoned, pvClusterDeleted)
		// This is for compatibility: old PVs are all tikv store and don't have app label
		subDir := util.TiKVStoreDir
		if kind, ok := pv.Labels[label.AppLabelKey]; ok && kind != "" {
			subDir = kind
		}

		dir := filepath.Join(annHostPath, subDir)
		if err := os.RemoveAll(dir); err != nil {
			return err
		}

		log.Infof("HostPath[%s:%s] cleaned successfully.", annNode, dir)
		return nil
	}

	return &controller.IgnoredError{fmt.Sprintf("PV: %s was not manual marked deleted, skipping", pvName)}
}

func pvCanManualDeleted(pv *v1.PersistentVolume) bool {
	return pv.Annotations[AnnPVManualDeleted] != "" &&
		(pv.Annotations[AnnPVTombstoned] != "" || pv.Annotations[AnnPVClusterDeleted] != "")
}