package k8s

import (
	"bytes"
	"context"
	_ "embed"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/hashicorp/go-multierror"
	"github.com/imdario/mergo"
	"github.com/pingcap/errors"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/yaml"

	apiresource "github.com/pingcap/test-infra/common/model/resource"

	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
)

var (
	_ resource.Client = TiDBClusterClient{}

	//go:embed assets/tidb-cluster.yaml
	tidbClusterSpec []byte
	//go:embed assets/tidb-monitor.yaml
	tidbMonitorSpec []byte

	//go:embed assets/tidb-ngmonitor.yaml
	tidbNGMonitorSpec []byte

	defaultTiDBClusterSpec   map[string]interface{}
	defaultTiDBMonitorSpec   map[string]interface{}
	defaultTiDBNGMonitorSpec map[string]interface{}

	// TODO(mahjonp): we should remove the resource deletion to a CRD reconcile, and let the resource be delete on the background
	waitForDeletionTimeout = 10 * time.Minute
)

type TiDBClusterClient struct {
	*KubeClientSet
}

func NewTiDBClusterClient(k *KubeClientSet) TiDBClusterClient {
	return TiDBClusterClient{k}
}

func (t TiDBClusterClient) Create(spec json.RawMessage) (core.Resource, error) {
	tc, tm, tngm, apiRes, err := t.buildTiDBClusterObjects(spec)
	if err != nil {
		return nil, errors.Annotate(err, "build tidb-cluster resource object")
	}
	if err := validatorSpecIfNeed(tc, tm); err != nil {
		return nil, errors.Annotatef(err, "validator tidbcluster spec")
	}

	if len(apiRes.Qos) != 0 {
		if err := t.applyQosClass(apiRes, tc); err != nil {
			return nil, err
		}
	}

	t.applyMaxReplicas(apiRes, tc)

	_, err = t.DynamicClient.Resource(tcGVR).Namespace(t.Namespace).Create(context.TODO(), tc, metav1.CreateOptions{})
	if err != nil {
		return nil, errors.Annotate(err, "create tidb-cluster resource")
	}
	_, err = t.DynamicClient.Resource(tmGVR).Namespace(t.Namespace).Create(context.TODO(), tm, metav1.CreateOptions{})
	if err != nil {
		return nil, errors.Annotate(err, "create tidb-monitor resource")
	}
	if tngm != nil {
		_, err = t.DynamicClient.Resource(tngmGVR).Namespace(t.Namespace).Create(context.TODO(), tngm, metav1.CreateOptions{})
		if err != nil {
			return nil, errors.Annotate(err, "create tidb-ngmonitoring resource")
		}
	}

	if err = t.deployPostStartJob(apiRes); err != nil {
		return nil, err
	}
	tidb := &TiDBCluster{
		KubeClientSet: *t.KubeClientSet,
		TiDBCluster: apiresource.TiDBCluster{
			ResourceHandle: apiRes.ResourceHandle,
		},
	}
	tidb.Provider = t.FQDN()
	if err := tidb.SyncTiDBCluster(); err != nil {
		return nil, err
	}
	return tidb, nil
}

func (t TiDBClusterClient) Get(handle apiresource.ResourceHandle) (core.Resource, error) {
	tidb := &TiDBCluster{
		KubeClientSet: *t.KubeClientSet,
		TiDBCluster: apiresource.TiDBCluster{
			ResourceHandle: handle,
		},
	}
	tidb.Provider = t.FQDN()
	if err := tidb.SyncTiDBCluster(); err != nil {
		return nil, err
	}
	return tidb, nil
}

func (t TiDBClusterClient) GetEndpoints(handle apiresource.ResourceHandle) (apiresource.ResourceEndpoints, error) {
	res := apiresource.ResourceEndpoints{}

	svcList, err := t.ClientSet.CoreV1().
		Services(handle.Testbed).
		List(context.TODO(), metav1.ListOptions{
			LabelSelector: labels.Set{"app.kubernetes.io/instance": handle.Name}.AsSelector().String(),
		})
	if err != nil {
		return res, err
	}

	for _, svc := range svcList.Items {
		if svc.Spec.Type != corev1.ServiceTypeNodePort {
			continue
		}

		pods, err := t.ClientSet.CoreV1().Pods(handle.Testbed).List(context.TODO(), metav1.ListOptions{
			LabelSelector: labels.FormatLabels(svc.Spec.Selector),
		})
		if err != nil {
			return res, err
		}
		if len(pods.Items) == 0 {
			return res, fmt.Errorf("service %s/%s has no target pod", svc.Namespace, svc.Name)
		}

		if svc.Name == handle.Name+"-tidb" {
			res = append(res, apiresource.Endpoint{
				Type:               apiresource.EndpointTypeDsn,
				Name:               svc.Name,
				ExternalSvcAddress: fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", "root", "", pods.Items[0].Status.HostIP, svc.Spec.Ports[0].NodePort, ""),
			})

			res = append(res, apiresource.Endpoint{
				Type: apiresource.EndpointTypeOther,
				Name: svc.Name + "-" + svc.Spec.Ports[1].Name,
				// Tidb Status port
				ExternalSvcAddress: fmt.Sprintf("http://%s:%d", pods.Items[0].Status.HostIP, svc.Spec.Ports[1].NodePort),
			})
		} else if svc.Name == handle.Name+"-grafana" {
			res = append(res, apiresource.Endpoint{
				Type:               apiresource.EndpointTypeGrafana,
				Name:               svc.Name,
				ExternalSvcAddress: fmt.Sprintf("http://%s:%d", pods.Items[0].Status.HostIP, svc.Spec.Ports[0].NodePort),
			})
		} else if svc.Name == handle.Name+"-prometheus" {
			res = append(res, apiresource.Endpoint{
				Type:               apiresource.EndpointTypePrometheus,
				Name:               svc.Name,
				ExternalSvcAddress: fmt.Sprintf("http://%s:%d", pods.Items[0].Status.HostIP, svc.Spec.Ports[0].NodePort),
			})
		}
	}
	return res, nil
}

func (t TiDBClusterClient) Delete(handle apiresource.ResourceHandle) error {
	watcher, err := t.DynamicClient.Resource(tcGVR).Namespace(t.Namespace).Watch(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{TestInfraName: handle.Name}.AsSelector().String(),
	})
	if err != nil {
		return err
	}
	ch := watcher.ResultChan()

	err = t.DynamicClient.Resource(tcGVR).Namespace(t.Namespace).Delete(context.TODO(), handle.Name, metav1.DeleteOptions{
		PropagationPolicy: &[]metav1.DeletionPropagation{metav1.DeletePropagationForeground}[0],
	})
	if err != nil {
		return err
	}

	// wait for tc deleted before delete pvcs
waitLoop:
	for {
		select {
		case event := <-ch:
			switch event.Type {
			case watch.Deleted:
				break waitLoop
			}
		case <-time.After(waitForDeletionTimeout):
			return fmt.Errorf("waiting for tidbcluster %s/%s deleted timeout", t.Namespace, handle.Name)
		}
	}

	err = t.ClientSet.CoreV1().PersistentVolumeClaims(t.Namespace).DeleteCollection(context.TODO(), metav1.DeleteOptions{}, metav1.ListOptions{
		LabelSelector: labels.Set{"app.kubernetes.io/instance": handle.Name}.AsSelector().String(),
	})
	if err != nil {
		return errors.Annotate(err, "delete persistent volume claims of "+handle.Name)
	}
	return nil
}

func (t TiDBClusterClient) ReserveEnv(handle apiresource.ResourceHandle, action apiresource.GcReserveEnvAction) error {
	return nil
}

func (t TiDBClusterClient) Artifact(handle apiresource.ResourceHandle, action apiresource.GcArtifactAction, archiver resource.ArchiverFunc) (map[string]*url.URL, error) {
	var errs error
	var urlMap = make(map[string]*url.URL)
	pods, err := t.ClientSet.CoreV1().Pods(t.Namespace).List(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{"app.kubernetes.io/instance": handle.Name}.AsSelector().String(),
	})
	if err != nil {
		return nil, err
	}
	for _, pod := range pods.Items {
		if pod.Status.Phase == corev1.PodRunning {
			component := pod.Labels["app.kubernetes.io/component"]
			switch component {
			case "tidb", "tikv", "pd", "ticdc", "tiflash":
				path := map[string]string{
					"tidb":    "/var/lib/tidb/log/",
					"tikv":    "/var/lib/tikv/log/",
					"pd":      "/var/lib/pd/log/",
					"tiflash": "/data0/logs/",
					"ticdc":   "/var/lib/ticdc/log/",
				}[component]
				// remove extraneous path shortcuts, for example:
				//  tar cf - /Users/manjunpeng/zero
				// the prefix is Users/manjunpeng/zero
				//  Users/manjunpeng/zero000644 000765 000024 00000000000 14113106412 016507 0ustar00manjunpengstaff000000 000000
				tarFilesPrefix := path[1:]
				cUrls, err := untarAll(t.TarStreamFrom(pod.Name, component, path), archiver, pod.Name, tarFilesPrefix)
				if err != nil {
					errs = multierror.Append(errs, err)
					continue
				}
				for objectName, url := range cUrls {
					urlMap[objectName] = url
				}
			case "monitor":
				tarStream := t.TarStreamFrom(pod.Name, "prometheus", "/data/prometheus/")
				// FIXME: need to add the tc name as the prefix since a testbed may contain more than one TIDB_CLUSTER
				objectName := "prometheus.tar"
				url, err := archiver("prometheus.tar", tarStream)
				if err != nil {
					errs = multierror.Append(errs, err)
					tarStream.Close()
					continue
				}
				tarStream.Close()
				urlMap[objectName] = url
			default:
				continue
			}
		}
	}
	return urlMap, errs
}

func (t TiDBClusterClient) FQDN() string {
	return resource.FQDN(t)
}

// buildTiDBClusterObjects ...
// params should be the apiresource.TiDBCluster with some extension fields
func (t TiDBClusterClient) buildTiDBClusterObjects(params json.RawMessage) (*unstructured.Unstructured, *unstructured.Unstructured, *unstructured.Unstructured, *apiresource.TiDBCluster, error) {
	var (
		rawClusterSpec   map[string]interface{}                                        // custom tidb cluster spec
		rawMonitorSpec   map[string]interface{}                                        // custom monitor spec
		rawNGMonitorSpec map[string]interface{}                                        // custom ngMonitoring spec
		tc               = new(unstructured.Unstructured)                              // the tidb-cluster CR object
		resHandle        = &apiresource.TiDBCluster{Spec: newDefaultTiDBClusterSpec()} // the api resource handle object
	)
	// unmarshal rawClusterSpec obj
	var rawObj map[string]interface{}
	if err := json.Unmarshal(params, &rawObj); err != nil {
		return nil, nil, nil, nil, errors.Trace(err)
	}
	if spec, ok := rawObj["spec"].(map[string]interface{}); ok {
		t.normalizeRawSpec(spec)
		rawClusterSpec = spec
	}
	if monitor, ok := rawClusterSpec["monitor"].(map[string]interface{}); ok {
		rawMonitorSpec = monitor
		delete(rawClusterSpec, "monitor")
	}
	if ngMonitoring, ok := rawClusterSpec["ngMonitoring"].(map[string]interface{}); ok {
		rawNGMonitorSpec = ngMonitoring
		delete(rawClusterSpec, "ngMonitoring")
	}

	// unmarshal resource handle
	if err := json.Unmarshal(params, resHandle); err != nil {
		return nil, nil, nil, nil, errors.Trace(err)
	}
	if err := mergo.Merge(&tc.Object, runtime.DeepCopyJSON(defaultTiDBClusterSpec)); err != nil {
		return nil, nil, nil, nil, errors.Trace(err)
	}
	// set the tidb-cluster CR object
	tc.SetName(resHandle.Name)
	tc.SetNamespace(t.Namespace)
	tcSpec := tc.Object["spec"].(map[string]interface{})
	if err := t.ApplyRawClusterSpec(&tcSpec, rawClusterSpec); err != nil {
		return nil, nil, nil, nil, errors.Trace(err)
	}
	// customObj labels
	tcLabels := tc.GetLabels()
	if err := mergo.Merge(&tcLabels, map[string]string{TestInfraName: resHandle.Name}, mergo.WithOverride); err != nil {
		return nil, nil, nil, nil, errors.Trace(err)
	}
	tc.SetLabels(tcLabels)

	// build the tidb-monitor CR object
	var tm = new(unstructured.Unstructured)
	tm.Object = runtime.DeepCopyJSON(defaultTiDBMonitorSpec)
	spec := tm.Object["spec"].(map[string]interface{})
	if err := t.ApplyRawMonitorSpec(&spec, rawMonitorSpec); err != nil {
		return nil, nil, nil, nil, errors.Trace(err)
	}
	tm.SetName(tc.GetName())
	tm.SetNamespace(tc.GetNamespace())
	clusters := []interface{}{map[string]interface{}{"name": tc.GetName()}}
	_ = unstructured.SetNestedField(tm.Object, clusters, "spec", "clusters")

	// build the tidb-ngmonitor CR object
	var tngm *unstructured.Unstructured
	if rawNGMonitorSpec != nil {
		create, ok := rawNGMonitorSpec["create"].(bool)
		if ok && create {
			tngm = new(unstructured.Unstructured)
			tngm.Object = runtime.DeepCopyJSON(defaultTiDBNGMonitorSpec)
			// set tc's spec.version as tngm's spec.ngMonitoring.version
			_ = unstructured.SetNestedField(tngm.Object, tcSpec["version"], "spec", "ngMonitoring", "version")

			spec = tngm.Object["spec"].(map[string]interface{})
			if err := t.ApplyRawNGMonitorSpec(spec, rawNGMonitorSpec); err != nil {
				return nil, nil, nil, nil, errors.Trace(err)
			}
			tngm.SetName(tc.GetName())
			tngm.SetNamespace(tc.GetNamespace())
			clusters = []interface{}{map[string]interface{}{"name": tc.GetName(), "namespace": tc.GetNamespace()}}
			_ = unstructured.SetNestedField(tngm.Object, clusters, "spec", "clusters")
		}
	}

	// add the testbed to tc's ownerReference
	owner, err := GetOwnerReferenceFromTestbed(t.KubeClientSet, resHandle.Testbed)
	if err != nil {
		return nil, nil, nil, nil, err
	}
	tc.SetOwnerReferences([]metav1.OwnerReference{*owner})
	tm.SetOwnerReferences([]metav1.OwnerReference{*owner})
	if tngm != nil {
		tngm.SetOwnerReferences([]metav1.OwnerReference{*owner})
	}
	return tc, tm, tngm, resHandle, nil
}

func (t TiDBClusterClient) applyQosClass(res *apiresource.TiDBCluster, tc *unstructured.Unstructured) error {
	for comp, qosClass := range res.Qos {
		pg, err := t.DynamicClient.Resource(podGroupGVR).Namespace(t.Namespace).Get(context.TODO(), qosClass.PodGroup, metav1.GetOptions{})
		if err != nil {
			return errors.Trace(err)
		}

		subGroupSpec := map[string]interface{}{
			"minMember": int64(1),
		}
		if qosClass.Exclusive != nil {
			subGroupSpec["exclusive"] = *qosClass.Exclusive
		}
		switch comp {
		case "tidb":
			subGroupSpec["minMember"] = int64(res.Spec.TiDB.Replicas)
		case "pd":
			subGroupSpec["minMember"] = int64(res.Spec.PD.Replicas)
		case "tikv":
			subGroupSpec["minMember"] = int64(res.Spec.TiKV.Replicas)
		case "tiflash":
			subGroupSpec["minMember"] = int64(res.Spec.TiFlash.Replicas)
		case "ticdc":
			subGroupSpec["minMember"] = int64(res.Spec.TiCDC.Replicas)
		}
		// append a component's annotation: $podgroup.$name_$comp format, for example: cluster-1.tc-1_tidb
		// a valid label must be an empty string or consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyValue',  or 'my_value',  or '12345', regex used for validation is '(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])?')
		_ = unstructured.SetNestedField(tc.Object, fmt.Sprintf("%s.%s_%s", qosClass.PodGroup, res.ResourceHandle.Name, comp), "spec", comp, "labels", podGroupLabelKey)

		// append a subgroup spec on the podgroup: $name_$comp format, for example: tc-1_tidb
		_ = unstructured.SetNestedField(pg.Object, subGroupSpec, "spec", "subGroups", fmt.Sprintf("%s_%s", res.ResourceHandle.Name, comp))
		if _, err := t.DynamicClient.Resource(podGroupGVR).Namespace(t.Namespace).Update(context.TODO(), pg, metav1.UpdateOptions{}); err != nil {
			return errors.Annotatef(err, "update podGroup %s", pg.GetName())
		}
	}
	return nil
}

// applyMaxReplicas set each component's max-replicas annotation
func (t TiDBClusterClient) applyMaxReplicas(res *apiresource.TiDBCluster, tc *unstructured.Unstructured) {
	componentSpecs := map[resource.TiDBCompType]*apiresource.TiDBClusterComponentSpec{
		resource.PD:      &res.Spec.PD,
		resource.TiDB:    &res.Spec.TiDB,
		resource.TiKV:    &res.Spec.TiKV,
		resource.TiFlash: &res.Spec.TiFlash,
		resource.TiCDC:   &res.Spec.TiCDC,
	}
	for comp, compSpec := range componentSpecs {
		if compSpec.MaxReplicas <= 0 {
			// do not bother testbeds doesn't explicitly set maxReplicas for now
			// setting this will slows down the scheduling process
			continue
		}

		_ = unstructured.SetNestedField(tc.Object, strconv.Itoa(compSpec.MaxReplicas), "spec", comp.String(), "annotations", maxReplicasAnnotationKey)
	}
}

// normalizeRawSpec normalizes the raw tidb spec
//
// for example: transform the below spec:
//  tiflash:
//    config: |
//      ## Log levels: trace, debug, info, warning, error, critical.
//      ## Note that `debug` and `trace` are only available in development builds.
//      log-level = "info"
// into:
//  tiflash:
//    config:
//      config: |
//        ## Log levels: trace, debug, info, warning, error, critical.
//        ## Note that `debug` and `trace` are only available in development builds.
//        log-level = "info"
func (t TiDBClusterClient) normalizeRawSpec(spec map[string]interface{}) {
	if tiflash, ok := spec["tiflash"].(map[string]interface{}); ok {
		if config, ok := tiflash["config"].(string); ok {
			tiflash["config"] = map[string]interface{}{"config": config}
		}
	}
}

func (t TiDBClusterClient) ApplyRawClusterSpec(defaultSpec *map[string]interface{}, rawSpec map[string]interface{}) error {
	for _, component := range []string{"tidb", "tikv", "pd", "ticdc"} {
		defaultConfig, err := getComponentConfig(*defaultSpec, component)
		if err != nil {
			return err
		}
		customConfig, err := getComponentConfig(rawSpec, component)
		if err != nil {
			return err
		}
		mp, _ := mergeMap(defaultConfig.MP, customConfig.MP)
		if err := setComponentTOMLConfig(rawSpec, component, mp); err != nil {
			return err
		}
	}
	if err := mergo.Merge(defaultSpec, rawSpec, mergo.WithOverride); err != nil {
		return err
	}
	return nil
}

func (t TiDBClusterClient) ApplyRawMonitorSpec(destSpec *map[string]interface{}, rawSpec map[string]interface{}) error {
	if rawSpec == nil {
		return nil
	}
	return mergo.Merge(destSpec, rawSpec, mergo.WithOverride)
}

func (t TiDBClusterClient) ApplyRawNGMonitorSpec(destSpec map[string]interface{}, rawSpec map[string]interface{}) error {
	if rawSpec == nil {
		return nil
	}
	ngm := destSpec["ngMonitoring"].(map[string]interface{})
	return mergo.Merge(&ngm, rawSpec, mergo.WithOverride)
}

func (t TiDBClusterClient) deployPostStartJob(apiRes *apiresource.TiDBCluster) error {
	var err error
	if strings.TrimSpace(apiRes.Spec.PostStart.SQL) != "" {
		_, err = t.ClientSet.BatchV1().Jobs(t.Namespace).Create(context.TODO(), &batchv1.Job{
			ObjectMeta: metav1.ObjectMeta{
				Name: fmt.Sprintf("%s-setup", apiRes.ResourceHandle.Name),
				Labels: map[string]string{
					"app.kubernetes.io/instance":  apiRes.ResourceHandle.Name,
					"app.kubernetes.io/component": "tc-setup",
				},
			},
			Spec: batchv1.JobSpec{
				Parallelism:           nil,
				ActiveDeadlineSeconds: nil,
				BackoffLimit:          pointer.Int32(0),
				Template: corev1.PodTemplateSpec{
					Spec: corev1.PodSpec{
						Containers: []corev1.Container{
							{
								Name:  "main",
								Image: config.Images.TCSetup,
								Env: []corev1.EnvVar{
									{
										Name:  "SETUP_SQL",
										Value: apiRes.Spec.PostStart.SQL,
									},
								},
								Command: []string{
									"/bin/sh",
									"-c",
									fmt.Sprintf(`set -e
while true; do
    if mysqladmin status -h %s-tidb -uroot -P4000 --connect-timeout 5
    then
        echo "connection to tidb cluster succeed"
        break
    else
        echo "connection to tidb cluster failed" >&2
    fi
    sleep 10
done
echo $SETUP_SQL | mysql -h %s-tidb -uroot -P4000
`, apiRes.ResourceHandle.Name, apiRes.ResourceHandle.Name),
								},
							},
						},
						RestartPolicy: corev1.RestartPolicyNever,
					},
				},
			},
		}, metav1.CreateOptions{})
		if err != nil {
			return errors.Annotate(err, "create tc-setup job error")
		}
	}
	return nil
}

func getComponentConfig(spec map[string]interface{}, component string) (*apiresource.GenericConfig, error) {
	res := apiresource.GenericConfig{MP: make(map[string]interface{})}
	path := []string{component, "config"}
	if component == resource.TiFlash.String() {
		path = append(path, "config")
	}
	config, found, err := unstructured.NestedMap(spec, path...)
	if err != nil {
		config, found, _ := unstructured.NestedString(spec, path...)
		if !found {
			return &apiresource.GenericConfig{}, nil
		}
		if err := res.UnmarshalTOML([]byte(config)); err != nil {
			return nil, err
		}
		return &res, nil
	}
	if found {
		return &apiresource.GenericConfig{MP: runtime.DeepCopyJSON(config)}, nil
	}
	return &res, nil
}

// yaml(map[string]interface{}) will be pruned by tidb-operator if it sees an unknown config item but toml config(string) will retain,
// we should use toml config(multiple lines format) here
//
// close: https://github.com/pingcap/test-infra/issues/420
func setComponentTOMLConfig(spec map[string]interface{}, component string, mp map[string]interface{}) error {
	config := apiresource.GenericConfig{MP: mp}
	data, err := config.MarshalTOML()
	if err != nil {
		return err
	}
	path := []string{component, "config"}
	if component == resource.TiFlash.String() {
		path = append(path, "config")
	}
	return unstructured.SetNestedField(spec, string(data), path...)
}

func mergeMap(m1 map[string]interface{}, m2 map[string]interface{}) (map[string]interface{}, bool) {
	res := make(map[string]interface{})
	var modified bool
	for key, value := range m1 {
		res[key] = value
	}
	for key, value := range m2 {
		old, exist := res[key]
		if !exist {
			res[key] = value
			modified = true
			continue
		} else {
			oldMap, oldOk := old.(map[string]interface{})
			newMap, mergeOk := value.(map[string]interface{})
			if oldOk && mergeOk {
				var modifiedRec bool
				res[key], modifiedRec = mergeMap(oldMap, newMap)
				if modifiedRec {
					modified = true
				}
			} else if res[key] != value {
				res[key] = value
				modified = true
			}
		}
	}
	return res, modified
}

func validatorSpecIfNeed(tc, tm *unstructured.Unstructured) error {
	if config.TiDBCluster.SpecValidator == nil {
		return nil
	}
	b, err := json.Marshal(map[string]interface{}{"tc": tc.Object["spec"], "tm": tm.Object["spec"]})
	if err != nil {
		return err
	}
	resp, err := http.Post(config.TiDBCluster.SpecValidator.TcValidatorURL, "application/json", bytes.NewReader(b))
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := ioutil.ReadAll(resp.Body)
		return fmt.Errorf("validator error: %s", string(body))
	}
	return nil
}

func newDefaultTiDBClusterSpec() apiresource.TiDBClusterSpec {
	return apiresource.TiDBClusterSpec{
		PD:      apiresource.TiDBClusterComponentSpec{},
		TiKV:    apiresource.TiDBClusterComponentSpec{},
		TiDB:    apiresource.TiDBClusterComponentSpec{},
		TiCDC:   apiresource.TiDBClusterComponentSpec{},
		TiFlash: apiresource.TiDBClusterComponentSpec{},
	}
}

func mustUnmarshalYaml(b []byte, o interface{}) {
	if err := yaml.Unmarshal(b, o); err != nil {
		panic(err)
	}
}

type tidbGetter struct{ handle apiresource.ResourceHandle }

func (t *tidbGetter) Get() (core.Resource, error) {
	k, err := NewKubeClientFromKubeConfig(t.handle.Testbed, os.Getenv("KUBECONFIG"))
	if err != nil {
		return nil, err
	}
	return NewTiDBClusterClient(k).Get(t.handle)
}

func init() {
	mustUnmarshalYaml(tidbClusterSpec, &defaultTiDBClusterSpec)
	mustUnmarshalYaml(tidbMonitorSpec, &defaultTiDBMonitorSpec)
	mustUnmarshalYaml(tidbNGMonitorSpec, &defaultTiDBNGMonitorSpec)

	core.RegisterProvider(resource.FQDN(TiDBClusterClient{}),
		func(handle apiresource.ResourceHandle) core.ResourceGetter { return &tidbGetter{handle: handle} })
}