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

import (
	"bytes"
	"fmt"
	"math"
	"text/template"

	"k8s.io/api/core/v1"
	"k8s.io/api/rbac/v1beta1"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	tiapi "github.com/pingcap/tidb-cloud-manager/pkg/tidbinstance/api"
	tcapi "github.com/pingcap/private-tidb-operator/pkg/tidbcluster/api"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

const (
	// These values are filled in alert-rules config template to avoid undefined variables
	metaInstance = "{{ $labels.instance }}"
	metaType     = "{{ $labels.type }}"
	metaValue    = "{{ $value }}"
	leftDelim    = "<%"
	rightDelim   = "%>"
)

// MetricsAddr field is reserved for backward compatibility
type pdClusterConfig struct {
	MetricsAddr string
}

type tikvClusterConfig struct {
	MetricsAddr             string
	EndpointConcurrency     int32
	DefaultCFBlockCacheSize string
	WriteCFBlockCacheSize   string
}

type tidbClusterConfig struct {
	MetricsAddr    string
	TiDBLease      int
	SkipGrantTable bool
}

type tidbPumpConfig struct {
	KafkaAddr string
}

type prometheusConfig struct {
	AlertmanagerURL string
	Namespace       string
	ClusterName     string
}

type alertConfig struct {
	ClusterName  string
	MetaInstance string
	MetaType     string
	MetaValue    string
}

type GrafanaConfig struct {
	GrafanaUrl      string
	GrafanaUsername string
	GrafanaPassword string
}

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

func updatePushGatewayAnn(ann map[string]string, tikvPS *tiapi.PodSpec) map[string]string {
	if ann == nil {
		ann = map[string]string{}
	}
	ann["pushgateway-image"] = GetImageByTiPodSpec(PushgatewayContainerType, tikvPS)
	return ann
}

// GetImageByTiPodSpec get container image based on a specify container name and tidbinstance podspec
func GetImageByTiPodSpec(ctype ContainerType, podSpec *tiapi.PodSpec) string {
	var image string
	for _, c := range podSpec.Containers {
		if string(ctype) == c.Name {
			image = c.Image
			break
		}
	}
	return image
}

// SetImageByTiPodSpec set specify for a container based on a specify container name and tidbinstance podspec
func SetImageByTiPodSpec(ctype ContainerType, podSpec *tiapi.PodSpec, image string) {
	for i, c := range podSpec.Containers {
		if string(ctype) == c.Name {
			podSpec.Containers[i].Image = image
			break
		}
	}
}

func (tic *TiDBInstanceController) loadConfigTemplate(instance *tiapi.TidbInstance) (map[string]*template.Template, error) {
	tplName := instance.Spec.Config
	cm, err := tic.cli.CoreV1().ConfigMaps(tic.ns).Get(tplName, metav1.GetOptions{})
	if err != nil {
		return nil, errors.Trace(err)
	}
	configTmpl := make(map[string]*template.Template)
	tmpl, exist := cm.Data["pd-config"]
	if !exist {
		return nil, errors.Errorf("pd-config not found in configmap: %s/%s", tic.ns, tplName)
	}
	if configTmpl["pd-config"], err = template.New("pd-config").Delims(leftDelim, rightDelim).Parse(tmpl); err != nil {
		return nil, errors.Trace(err)
	}

	tmpl, exist = cm.Data["tikv-config"]
	if !exist {
		return nil, errors.Errorf("tikv-config not found in configmap: %s/%s", tic.ns, tplName)
	}
	if configTmpl["tikv-config"], err = template.New("tikv-config").Delims(leftDelim, rightDelim).Parse(tmpl); err != nil {
		return nil, errors.Trace(err)
	}

	tmpl, exist = cm.Data["tidb-config"]
	if !exist {
		return nil, errors.Errorf("tidb-config not found in configmap: %s/%s", tic.ns, tplName)
	}
	if configTmpl["tidb-config"], err = template.New("tidb-config").Delims(leftDelim, rightDelim).Parse(tmpl); err != nil {
		return nil, errors.Trace(err)
	}

	tmpl, exist = cm.Data["pump-config"]
	if !exist {
		return nil, errors.Errorf("pump-config not found in configmap: %s/%s", tic.ns, tplName)
	}
	if configTmpl["pump-config"], err = template.New("pump-config").Delims(leftDelim, rightDelim).Parse(tmpl); err != nil {
		return nil, errors.Trace(err)
	}

	tmpl, exist = cm.Data["prometheus-config"]
	if !exist {
		return nil, errors.Errorf("prometheus-config not found in configmap: %s/%s", tic.ns, tplName)
	}
	if configTmpl["prometheus-config"], err = template.New("prometheus-config").Delims(leftDelim, rightDelim).Parse(tmpl); err != nil {
		return nil, errors.Trace(err)
	}

	tmpl, exist = cm.Data["alert-rules-config"]
	if !exist {
		return nil, errors.Errorf("alert-rules-config not found in configmap: %s/%s", tic.ns, tplName)
	}
	if configTmpl["alert-rules-config"], err = template.New("alert-rules-config").Delims(leftDelim, rightDelim).Parse(tmpl); err != nil {
		return nil, errors.Trace(err)
	}

	//bakcward compatible
	tmpl, exist = cm.Data["grafana-config"]
	if exist {
		if configTmpl["grafana-config"], err = template.New("grafana-config").Delims(leftDelim, rightDelim).Parse(tmpl); err != nil {
			return nil, errors.Trace(err)
		}
	} else {

		log.Errorf(" grafana-config not found in configmap: %s/%s", tic.ns, tplName)
	}

	return configTmpl, nil
}

func (tic *TiDBInstanceController) genConfig(instance *tiapi.TidbInstance, configTmpl map[string]*template.Template) (map[string]string, error) {
	pdConfig := bytes.NewBuffer([]byte{})
	tikvConfig := bytes.NewBuffer([]byte{})
	tidbConfig := bytes.NewBuffer([]byte{})
	privilegedTidbConfig := bytes.NewBuffer([]byte{})
	promConfig := bytes.NewBuffer([]byte{})
	alertRulesConfig := bytes.NewBuffer([]byte{})
	pumpConfig := bytes.NewBuffer([]byte{})
	var pdConf pdClusterConfig
	if err := configTmpl["pd-config"].Execute(pdConfig, pdConf); err != nil {
		return nil, errors.Trace(err)
	}
	tikvConf := tikvClusterConfig{MetricsAddr: "http://127.0.0.1:9091"}
	tikvCS := getContainerSpec(TiKVContainerType, instance.Spec.TiKV)
	rr := getResourceRequirements(tikvCS)
	if cpuLimit, found := rr.Limits[v1.ResourceCPU]; found {
		if l, b := cpuLimit.AsInt64(); b {
			tikvConf.EndpointConcurrency = int32(float64(l) * 0.8)
		}
	}
	if memoryLimit, found := rr.Limits[v1.ResourceMemory]; found {
		if l, b := memoryLimit.AsInt64(); b {
			tikvConf.DefaultCFBlockCacheSize = fmt.Sprintf("%dGB", int(float64(l)*0.3/math.Pow(2, 30)))
			tikvConf.WriteCFBlockCacheSize = fmt.Sprintf("%dGB", int(float64(l)*0.2/math.Pow(2, 30)))
		}
	}
	if err := configTmpl["tikv-config"].Execute(tikvConfig, tikvConf); err != nil {
		return nil, errors.Trace(err)
	}
	tidbConf := tidbClusterConfig{
		TiDBLease:      instance.Spec.TiDBLease,
		SkipGrantTable: false,
	}
	if err := configTmpl["tidb-config"].Execute(tidbConfig, tidbConf); err != nil {
		return nil, errors.Trace(err)
	}
	// generate privileged tidb config
	tidbConf.SkipGrantTable = true
	if err := configTmpl["tidb-config"].Execute(privilegedTidbConfig, tidbConf); err != nil {
		return nil, errors.Trace(err)
	}

	promConf := prometheusConfig{
		AlertmanagerURL: AlertmanagerURL,
		Namespace:       instance.Namespace,
		ClusterName:     clusterName,
	}
	if err := configTmpl["prometheus-config"].Execute(promConfig, promConf); err != nil {
		return nil, errors.Trace(err)
	}

	alertConf := alertConfig{ClusterName: instance.Name, MetaInstance: metaInstance, MetaType: metaType, MetaValue: metaValue}
	if err := configTmpl["alert-rules-config"].Execute(alertRulesConfig, alertConf); err != nil {
		return nil, errors.Trace(err)
	}

	pumpConf := tidbPumpConfig{KafkaAddr: instance.Spec.KafkaAddr}
	if err := configTmpl["pump-config"].Execute(pumpConfig, pumpConf); err != nil {
		return nil, errors.Trace(err)
	}

	config := map[string]string{
		"pd-config":              pdConfig.String(),
		"tikv-config":            tikvConfig.String(),
		"tidb-config":            tidbConfig.String(),
		"privileged-tidb-config": privilegedTidbConfig.String(),
		"prometheus-config":      promConfig.String(),
		"alert-rules-config":     alertRulesConfig.String(),
		"pump-config":            pumpConfig.String(),
	}

	//backward compatible
	if _, ok := configTmpl["grafana-config"]; ok {
		grafanaConfig := bytes.NewBuffer([]byte{})
		grafanaUrl := instance.Spec.GrafanaUrl
		if grafanaUrl == "" {
			grafanaUrl = "http://localhost:3000"
		}
		grafanaUsername, ok := instance.Spec.Secrets["grafana-username"]
		if !ok {
			grafanaUsername = "admin"
		}
		grafanaPassword, ok := instance.Spec.Secrets["grafana-password"]
		if !ok {
			grafanaPassword = "admin"
		}
		grafanaConf := GrafanaConfig{
			GrafanaUrl:      grafanaUrl,
			GrafanaUsername: grafanaUsername,
			GrafanaPassword: grafanaPassword,
		}
		if err := configTmpl["grafana-config"].Execute(grafanaConfig, grafanaConf); err != nil {
			return nil, errors.Trace(err)
		}

		config["grafana-config"] = grafanaConfig.String()
	}

	return config, nil
}

func (tic *TiDBInstanceController) genRenderedConfig(instance *tiapi.TidbInstance) (map[string]string, error) {
	renderedConfig := map[string]string{}
	configTmpl, err := tic.loadConfigTemplate(instance)
	if err != nil {
		log.Errorf("failed to load TidbInstance: [%s/%s] config template: %v", instance.Namespace, instance.Name, err)
		return renderedConfig, errors.Trace(err)
	}
	config, err := tic.genConfig(instance, configTmpl)
	if err != nil {
		log.Errorf("failed to generate ConfigMap %s/%s from template: %v", instance.GetNamespace(), getName("config"), err)
		return config, errors.Trace(err)
	}
	return config, err
}

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 getResourceRequirements(cs *tiapi.ContainerSpec) *v1.ResourceRequirements {
	rr := v1.ResourceRequirements{}
	if cs.Requests != nil {
		rr.Requests = make(map[v1.ResourceName]resource.Quantity)
		if q, err := resource.ParseQuantity(cs.Requests.CPU); err != nil {
			log.Errorf("failed to parse CPU resource %s to quantity, error: %v", cs.Requests.CPU, err)
		} else {
			rr.Requests[v1.ResourceCPU] = q
		}
		if q, err := resource.ParseQuantity(cs.Requests.Memory); err != nil {
			log.Errorf("failed to parse memory resource %s to quantity, error: %v", cs.Requests.Memory, err)
		} else {
			rr.Requests[v1.ResourceMemory] = q
		}
	}
	if cs.Limits != nil {
		rr.Limits = make(map[v1.ResourceName]resource.Quantity)
		if q, err := resource.ParseQuantity(cs.Limits.CPU); err != nil {
			log.Errorf("failed to parse CPU resource %s to quantity, error: %v", cs.Limits.CPU, err)
		} else {
			rr.Limits[v1.ResourceCPU] = q
		}
		if q, err := resource.ParseQuantity(cs.Limits.Memory); err != nil {
			log.Errorf("failed to parse memory resource %s to quantity, error: %v", cs.Limits.Memory, err)
		} else {
			rr.Limits[v1.ResourceMemory] = q
		}
	}
	return &rr
}

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

func assembleMemberSpec(ctype ContainerType, podSpec *tiapi.PodSpec) tcapi.MemberSpec {
	memSpec := tcapi.MemberSpec{
		Size:                 podSpec.Size,
		NodeSelector:         podSpec.NodeSelector,
		NodeSelectorRequired: podSpec.NodeSelectorRequired,
	}
	for _, c := range podSpec.Containers {
		if string(ctype) == c.Name {
			memSpec.Image = c.Image
			memSpec.Limits = c.Limits
			memSpec.Requests = c.Requests
			break
		}
	}
	return memSpec
}

func HasBinlogEnabled(instance *tiapi.TidbInstance) bool {
	binlogEnabled := false
	for _, c := range instance.Spec.TiDB.Containers {
		if string(BinlogContainerType) == c.Name && len(instance.Spec.KafkaAddr) != 0 {
			binlogEnabled = true
			break
		}
	}
	return binlogEnabled
}

// Shouldresize returns whether should resize cluster
func ShouldResize(instance *tiapi.TidbInstance, tc *tcapi.TidbCluster) bool {
	return instance.Spec.PD.Size != tc.Spec.PD.Size ||
		instance.Spec.TiKV.Size != tc.Spec.TiKV.Size ||
		instance.Spec.TiDB.Size != tc.Spec.TiDB.Size
}

func generateOwnerRef(instance *tiapi.TidbInstance) metav1.OwnerReference {
	controller := true
	ownerRef := metav1.OwnerReference{
		APIVersion: controllerKind.GroupVersion().String(),
		Kind:       controllerKind.Kind,
		Name:       instance.GetName(),
		UID:        instance.GetUID(),
		Controller: &controller,
	}
	return ownerRef
}

func (tic *TiDBInstanceController) ensureRBAC(ns, name string) error {
	if err := tic.ensureSAExists(ns, name); err != nil {
		return err
	}
	if err := tic.ensureRoleExists(ns, name); err != nil {
		return err
	}
	return tic.ensureRBExists(ns, name)
}

// ensure Role exists
func (tic *TiDBInstanceController) ensureRoleExists(ns, name string) error {
	_, err := tic.cli.RbacV1beta1().Roles(ns).Get(name, metav1.GetOptions{})
	if err != nil && !apierrors.IsNotFound(err) {
		return errors.Trace(err)
	}
	if err == nil {
		return nil
	}
	role := &v1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: ns,
		},
		Rules: []v1beta1.PolicyRule{
			{
				APIGroups: []string{""},
				Resources: []string{"pods"},
				Verbs:     []string{"get", "watch", "list"},
			},
		},
	}
	_, err = tic.cli.RbacV1beta1().Roles(ns).Create(role)
	if err != nil {
		log.Errorf("failed to create Role %s/%s: %v", ns, name, err)
		return errors.Trace(err)
	}
	log.Infof("successfully created Role %s/%s", ns, name)
	return nil
}

// ensure RoleBinding exists
func (tic *TiDBInstanceController) ensureRBExists(ns, name string) error {
	_, err := tic.cli.RbacV1beta1().RoleBindings(ns).Get(name, metav1.GetOptions{})
	if err != nil && !apierrors.IsNotFound(err) {
		return errors.Trace(err)
	}
	if err == nil {
		return nil
	}
	rb := &v1beta1.RoleBinding{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: ns,
		},
		Subjects: []v1beta1.Subject{
			{
				Kind: v1beta1.ServiceAccountKind,
				Name: name,
			},
		},
		RoleRef: v1beta1.RoleRef{
			Kind:     "Role",
			Name:     name,
			APIGroup: v1beta1.GroupName,
		},
	}
	_, err = tic.cli.RbacV1beta1().RoleBindings(ns).Create(rb)
	if err != nil {
		log.Errorf("failed to create RoleBinding %s/%s: %v", ns, name, err)
		return errors.Trace(err)
	}
	log.Infof("successfully created RoleBinding %s/%s", ns, name)
	return nil
}

// ensure ServiceAccount exists
func (tic *TiDBInstanceController) ensureSAExists(ns, name string) error {
	_, err := tic.cli.CoreV1().ServiceAccounts(ns).Get(name, metav1.GetOptions{})
	if err != nil && !apierrors.IsNotFound(err) {
		return errors.Trace(err)
	}
	if err == nil {
		return nil
	}
	sa := &v1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: ns,
			Name:      name,
		},
	}
	_, err = tic.cli.CoreV1().ServiceAccounts(ns).Create(sa)
	if err != nil {
		log.Errorf("failed to create ServiceAccount %s/%s: %v", ns, name, err)
		return errors.Trace(err)
	}
	log.Infof("successfully created ServiceAccount %s/%s", ns, name)
	return nil
}

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