package k8s

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"math/rand"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/hashicorp/go-multierror"
	"github.com/pingcap/errors"
	pkgErrors "github.com/pkg/errors"
	prometheus "github.com/prometheus/client_golang/api"
	prometheus_api "github.com/prometheus/client_golang/api/prometheus/v1"
	"go.uber.org/zap"
	batchv1 "k8s.io/api/batch/v1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	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/schema"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/util/retry"
	"k8s.io/utils/pointer"

	apiresource "github.com/pingcap/test-infra/common/model/resource"
	logsearch "github.com/pingcap/test-infra/sdk/pkg/logs"
	"github.com/pingcap/test-infra/sdk/pkg/util/pdutil"
	"github.com/pingcap/test-infra/sdk/resource"
)

var (
	tcGVR       = schema.GroupVersionResource{Group: "pingcap.com", Version: "v1alpha1", Resource: "tidbclusters"}
	tmGVR       = schema.GroupVersionResource{Group: "pingcap.com", Version: "v1alpha1", Resource: "tidbmonitors"}
	tngmGVR     = schema.GroupVersionResource{Group: "pingcap.com", Version: "v1alpha1", Resource: "tidbngmonitorings"}
	podGroupGVR = schema.GroupVersionResource{Group: "naglfar.pingcap.com", Version: "v1", Resource: "podgroups"}
)

const (
	// normalPhase represents normal state of TiDB cluster.
	normalPhase string = "Normal"
)

var _ resource.TiDBCluster = &TiDBCluster{}

type TiDBCluster struct {
	KubeClientSet
	// api object
	apiresource.TiDBCluster
	// k8s CR object
	tc *unstructured.Unstructured // tidb-cluster CR object
}

func (t *TiDBCluster) ChaosClient() resource.ChaosClient {
	return &ChaosClient{t.KubeClientSet}
}

func (t *TiDBCluster) PrometheusClient() (prometheus.Client, error) {
	prometheusURL, err := t.ServiceURL(resource.Prometheus)
	if err != nil {
		return nil, err
	}
	client, err := prometheus.NewClient(prometheus.Config{
		Address: prometheusURL.String(),
	})
	if err != nil {
		return nil, errors.Annotate(err, fmt.Sprintf("failed to create prometheus client with url: %s", prometheusURL))
	}
	return client, nil
}

func (t *TiDBCluster) PrometheusAPI() (prometheus_api.API, error) {
	client, err := t.PrometheusClient()
	if err != nil {
		return nil, err
	}
	return prometheus_api.NewAPI(client), nil
}

func (t *TiDBCluster) PDClient() (*pdutil.Client, error) {
	pdURL, err := t.ServiceURL(resource.PDAddr)
	if err != nil {
		return nil, err
	}
	return pdutil.NewPDClient(http.DefaultClient, pdURL.String()), nil
}

func (t *TiDBCluster) Handle() apiresource.ResourceHandle {
	return t.TiDBCluster.ResourceHandle
}

func (t *TiDBCluster) SinkEndpoint() (*url.URL, error) {
	return t.ServiceURL(resource.DBAddr)
}

func (t *TiDBCluster) ServiceURL(typ resource.TiDBAddrType) (*url.URL, error) {
	switch typ {
	case resource.DBAddr, resource.DBStatus:
		if t.Spec.TiDB.Replicas == 0 {
			return nil, errors.New("tidb.replicas is zero")
		}
		if typ == resource.DBStatus {
			return url.Parse(fmt.Sprintf("http://%s-tidb.%s:10080", t.Name(), t.Namespace()))
		}
		return url.Parse(fmt.Sprintf("mysql://root:@%s-tidb.%s:4000", t.Name(), t.Namespace()))
	case resource.PDAddr:
		return url.Parse(fmt.Sprintf("http://%s-pd.%s:2379", t.Name(), t.Namespace()))
	case resource.Prometheus:
		return url.Parse(fmt.Sprintf("http://%s-prometheus.%s:9090", t.Name(), t.Namespace()))
	default:
		return nil, errors.New("service is unavailable")
	}
}

// Headless service use PeerURL to access specified pod
func (t *TiDBCluster) PeerURLs(typ resource.TiDBAddrType) ([]*url.URL, error) {
	var (
		format   string
		replicas int
		err      error
	)
	switch typ {
	case resource.DBAddr:
		format = "mysql://root:@%s-tidb-%d.%s-tidb-peer.%s:4000"
		replicas = t.Spec.TiDB.Replicas
	case resource.DBStatus:
		format = "http://%s-tidb-%d.%s-tidb-peer.%s:10080"
		replicas = t.Spec.TiDB.Replicas
	case resource.KVAddr:
		format = "http://%s-tikv-%d.%s-tikv-peer.%s:20160"
		replicas = t.Spec.TiKV.Replicas
	case resource.KVStatus:
		format = "http://%s-tikv-%d.%s-tikv-peer.%s:20180"
		replicas = t.Spec.TiKV.Replicas
	case resource.PDAddr:
		format = "http://%s-pd-%d.%s-pd-peer.%s:2379"
		replicas = t.Spec.PD.Replicas
	case resource.CDCAddr:
		format = "http://%s-ticdc-%d.%s-ticdc-peer.%s:8301"
		replicas = t.Spec.TiCDC.Replicas
	case resource.PDApi:
		format = "http://%s-pd-%d.%s-pd-peer.%s:2379/pd/api/v1"
		replicas = t.Spec.PD.Replicas
	case resource.TiFlashAddr:
		format = "http://%s-tiflash-%d.%s-tiflash-peer.%s:3930"
		replicas = t.Spec.TiFlash.Replicas
	default:
		// TODO(@mahjonp): does tiflash support peer?
		return nil, errors.New("peers are unavailable")
	}
	urls := make([]*url.URL, replicas)
	for i := 0; i < replicas; i++ {
		urls[i], err = url.Parse(fmt.Sprintf(format, t.Name(), i, t.Name(), t.Namespace()))
		if err != nil {
			return nil, err
		}
	}
	return urls, nil
}

func (t *TiDBCluster) Exec(opts resource.TiDBClusterExecOptions) (stdout string, stderr string, exitCode int, err error) {
	podName := fmt.Sprintf("%s-%s-%d", t.Name(), opts.Target.String(), opts.Peer)
	stdout, stderr, ec, err := t.ExecCommandInContainerWithFullOutput(podName, opts.Target.String(), "/bin/sh", "-c", opts.Command)
	if err != nil {
		sugar.Error("Exec error", zap.Error(err))
		err = errors.Trace(err)
	}
	if ec != nil {
		exitCode = *ec
	} else {
		sugar.Warnw("cannot get the exit code, return -1")
		exitCode = -1
	}
	return
}

func (t *TiDBCluster) UpdateConfigAsync(opts resource.TiDBClusterUpdateConfigOptions) (err error) {
	if err := t.SyncTiDBCluster(); err != nil {
		return errors.Annotate(err, "sync tidb-cluster error")
	}
	tcSpec, _ := json.Marshal(t.TiDBCluster.Spec)
	var oldRawTCSpec map[string]interface{}
	if err := json.Unmarshal(tcSpec, &oldRawTCSpec); err != nil {
		return errors.Annotate(err, "tidbcluster spec json unmarshal error")
	}
	var updateConfig interface{} = opts.Config
	if opts.Target == resource.TiFlash {
		updateConfig = map[string]interface{}{"config": updateConfig}
	}
	newGenericCFG, err := getComponentConfig(map[string]interface{}{opts.Target.String(): map[string]interface{}{"config": updateConfig}},
		opts.Target.String())
	if err != nil {
		return errors.Annotate(err, opts.Target.String()+": get new component config error")
	}
	tc, err := t.KubeClientSet.DynamicClient.Resource(tcGVR).Namespace(t.Namespace()).Get(context.TODO(), t.Name(), metav1.GetOptions{})
	if err != nil {
		return errors.Annotate(err, "get tidb-cluster object error")
	}
	oldGenericCFG, err := getComponentConfig(tc.Object["spec"].(map[string]interface{}), opts.Target.String())
	if err != nil {
		return errors.Annotate(err, opts.Target.String()+": get old component config error")
	}
	config, modified := mergeMap(oldGenericCFG.MP, newGenericCFG.MP)
	if !modified {
		return errors.New("no updated config")
	}
	err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		if err = setComponentTOMLConfig(tc.Object["spec"].(map[string]interface{}), opts.Target.String(), config); err != nil {
			return err
		}
		_, updateErr := t.KubeClientSet.DynamicClient.Resource(tcGVR).Namespace(t.Namespace()).Update(context.TODO(), tc, metav1.UpdateOptions{})
		if updateErr == nil {
			return nil
		}

		if updated, err := t.KubeClientSet.DynamicClient.Resource(tcGVR).Namespace(t.Namespace()).Get(context.TODO(), t.Name(), metav1.GetOptions{}); err != nil {
			return errors.Annotate(err, "get tidb-cluster object error")
		} else {
			tc = updated.DeepCopy()
		}

		return updateErr
	})

	if err != nil {
		return errors.Annotate(err, "update tidb-cluster config error")
	}
	return nil
}

func (t *TiDBCluster) UpdateConfig(opts resource.TiDBClusterUpdateConfigOptions) error {
	now := time.Now()
	oldDetails, err := t.getComponentRevision(t.Namespace(), t.Name(), opts.Target.String())
	if err != nil {
		return errors.Annotate(err, "get tidbcluster component revisions error")
	}
	err = t.UpdateConfigAsync(opts)
	if err != nil {
		return err
	}
	err = wait.PollImmediate(5*time.Second, opts.Timeout, func() (done bool, err error) {
		newDetails, err := t.getComponentRevision(t.Namespace(), t.Name(), opts.Target.String())
		if err != nil {
			return false, errors.Annotate(err, "get tidbcluster update status error")
		}
		return newDetails[opts.Target.String()+"CurrentRevision"] != oldDetails[opts.Target.String()+"CurrentRevision"], nil
	})
	if err != nil {
		return err
	}
	var timeout time.Duration
	if opts.Timeout != 0 {
		timeout = time.Until(now.Add(opts.Timeout))
		if timeout <= 0 {
			return fmt.Errorf("update config timeout")
		}
	}
	if err = t.WaitForReady(timeout); err != nil {
		return err
	}
	if err := t.SyncTiDBCluster(); err != nil {
		return errors.Annotate(err, "sync tidb-cluster error")
	}
	return nil
}

func (t *TiDBCluster) UpdateConfigs(opts resource.TiDBClusterUpdateConfigsOptions) error {
	now := time.Now()
	var components []string
	for component := range opts.Configs {
		components = append(components, component.String())
	}
	oldDetails, err := t.getComponentRevisions(t.Namespace(), t.Name(), components...)
	if err != nil {
		return errors.Annotate(err, "get tidbcluster component revisions error")
	}
	for component, config := range opts.Configs {
		if err = t.UpdateConfigAsync(resource.TiDBClusterUpdateConfigOptions{
			Target: component,
			Config: config,
		}); err != nil {
			return err
		}
	}
	var ctx, cancel = context.Background(), func() {}
	if opts.Timeout != 0 {
		ctx, cancel = context.WithTimeout(ctx, opts.Timeout)
	}
	defer cancel()
	for component := range opts.Configs {
		err = wait.PollImmediateUntilWithContext(ctx, 5*time.Second, func(ctx context.Context) (done bool, err error) {
			newDetails, err := t.getComponentRevision(t.Namespace(), t.Name(), component.String())
			if err != nil {
				return false, errors.Annotatef(err, "get tidbcluster component revision %s error", component.String())
			}
			return newDetails[component.String()+"CurrentRevision"] != oldDetails[component.String()+"CurrentRevision"], nil
		})
		if err != nil {
			return err
		}
	}
	var timeout time.Duration
	if opts.Timeout != 0 {
		timeout = time.Until(now.Add(opts.Timeout))
		if timeout <= 0 {
			return fmt.Errorf("update configs timeout")
		}
	}
	if err = t.WaitForReady(timeout); err != nil {
		return err
	}
	if err := t.SyncTiDBCluster(); err != nil {
		return errors.Annotate(err, "sync tidb-cluster error")
	}
	return nil
}

func (t *TiDBCluster) getComponentRevision(ns string, name string, component string) (res map[string]interface{}, err error) {
	res = make(map[string]interface{})

	sts, err := t.KubeClientSet.ClientSet.AppsV1().StatefulSets(ns).Get(context.TODO(), name+"-"+component, metav1.GetOptions{})
	res[component+"CurrentRevision"] = sts.Status.CurrentRevision
	res[component+"UpdateRevision"] = sts.Status.UpdateRevision
	return
}

func (t *TiDBCluster) getComponentRevisions(ns string, name string, components ...string) (res map[string]interface{}, err error) {
	res = make(map[string]interface{})
	stss, err := t.KubeClientSet.ClientSet.AppsV1().StatefulSets(ns).List(context.TODO(), metav1.ListOptions{})
	if err != nil {
		return nil, err
	}
componentLoop:
	for _, component := range components {
		for _, sts := range stss.Items {
			if sts.Name == fmt.Sprintf("%s-%s", name, component) {
				res[component+"CurrentRevision"] = sts.Status.CurrentRevision
				res[component+"UpdateRevision"] = sts.Status.UpdateRevision
				continue componentLoop
			}
		}
		return nil, fmt.Errorf("not found: %s-%s", name, component)
	}
	return
}

var ErrUpgradeTimeout = fmt.Errorf("upgrade the cluster timeout")

func (t *TiDBCluster) Upgrade(opts resource.TiDBClusterUpgradeOptions) error {
	now := time.Now()
	var err error
	if opts.ComponentUpgradeOption != nil {
		err = t.upgradeComponent(opts)
	} else {
		err = t.upgradeCluster(opts)
	}
	if err != nil {
		return errors.Trace(err)
	}
	if opts.Timeout == 0 {
		return errors.Trace(t.WaitForReady(opts.Timeout))
	}
	left := time.Until(now.Add(opts.Timeout))
	if left <= 0 {
		return ErrUpgradeTimeout
	}
	return errors.Trace(t.WaitForReady(left))
}

func (t *TiDBCluster) upgradeComponent(opts resource.TiDBClusterUpgradeOptions) error {
	if err := t.SyncTiDBCluster(); err != nil {
		return err
	}
	compSpecGetter, compStatusGetter, err := t.getCompInfoFns(opts.ComponentUpgradeOption.Target)
	if err != nil {
		return err
	}
	// first, we get the currentStatus of component to upgrade(it's contain the Stateful Set's status which we need to inspect the upgrade progress: currentRevision and updateRevision)
	oldStatus := compStatusGetter().Status
	err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		upgradeImage := fmt.Sprintf("%s:%s", compSpecGetter().BaseImage, opts.ComponentUpgradeOption.Version)
		if compStatusGetter().Image == upgradeImage && compStatusGetter().Phase == normalPhase {
			return nil
		}
		err = unstructured.SetNestedField(t.tc.Object, opts.ComponentUpgradeOption.Version, "spec", opts.ComponentUpgradeOption.Target.String(), "version")
		if err != nil {
			return errors.Trace(err)
		}
		_, updateErr := t.KubeClientSet.DynamicClient.Resource(tcGVR).Namespace(t.Namespace()).Update(context.TODO(), t.tc, metav1.UpdateOptions{})
		if updateErr == nil {
			return nil
		}
		if err = t.SyncTiDBCluster(); err != nil {
			return err
		}
		return updateErr
	})
	if err != nil {
		return err
	}
	err = wait.PollImmediate(5*time.Second, opts.Timeout, func() (done bool, err error) {
		err = t.SyncTiDBCluster()
		if err != nil {
			return false, err
		}
		upgradeImage := fmt.Sprintf("%s:%s", compSpecGetter().BaseImage, opts.ComponentUpgradeOption.Version)
		switch opts.ComponentUpgradeOption.Target {
		case resource.TiDB, resource.TiKV, resource.PD, resource.TiFlash:
			return compStatusGetter().Image == upgradeImage && compStatusGetter().Phase == normalPhase, nil
		case resource.TiCDC:
			// ticdc's status has no image field, so we expect an update on sts status's CurrentRevision(ensure tidb-operator has reconcile this image update: Phase has change to Upgrade)
			return compStatusGetter().Status.CurrentRevision != oldStatus.CurrentRevision && compStatusGetter().Phase == normalPhase, nil
		default:
			return false, fmt.Errorf("unknown target type: %s", opts.ComponentUpgradeOption.Target.String())
		}
	})
	if pkgErrors.Is(err, wait.ErrWaitTimeout) {
		return ErrUpgradeTimeout
	}
	return err
}

func (t *TiDBCluster) upgradeCluster(opts resource.TiDBClusterUpgradeOptions) error {
	var err error
	if err = t.SyncTiDBCluster(); err != nil {
		return err
	}
	oldStatus := t.Status
	versionUpdater := func(tc *unstructured.Unstructured) error {
		err = unstructured.SetNestedField(tc.Object, opts.Version, "spec", "version")
		if err != nil {
			return err
		}
		// upgradeCluster should clean all separate component versions on component list:
		// component list: https://github.com/pingcap/tidb-operator/blob/935aae463ff4938d5e64d6ce287ba4952b4a575b/pkg/apis/pingcap/v1alpha1/types.go#L164-L184
		for _, component := range []string{"tidb", "tikv", "pd", "ticdc", "tiflash", "pump"} {
			unstructured.RemoveNestedField(tc.Object, "spec", component, "version")
		}
		// and if OverrideComponentsUpgradeOption is set, set the override version of components
		for _, option := range opts.OverrideComponentsUpgradeOption {
			err = unstructured.SetNestedField(tc.Object, option.Version, "spec", option.Target.String(), "version")
			if err != nil {
				return err
			}
		}
		return nil
	}
	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		if err := versionUpdater(t.tc); err != nil {
			return err
		}
		_, updateErr := t.KubeClientSet.DynamicClient.Resource(tcGVR).Namespace(t.Namespace()).Update(context.TODO(), t.tc, metav1.UpdateOptions{})
		if updateErr == nil {
			return nil
		}
		if err = t.SyncTiDBCluster(); err != nil {
			return err
		}
		return updateErr
	})
	if err != nil {
		return err
	}
	now := time.Now()
	// https://github.com/pingcap/tiup/blame/8c7a48e4ee694a047094865b37a5124a24d517d7/pkg/cluster/spec/spec.go#L663-L678
	// upgrade order: "tiflash", "pd", "tikv", "pump", "tidb", "drainer", "cdc", "prometheus", "grafana", "alertmanager"
	type componentStatus struct {
		resource.TiDBCompType
		apiresource.TiDBClusterComponentStatus
	}
	for _, compOldStatus := range []componentStatus{
		{TiDBCompType: resource.TiFlash, TiDBClusterComponentStatus: oldStatus.TiFlash},
		{TiDBCompType: resource.PD, TiDBClusterComponentStatus: oldStatus.PD},
		{TiDBCompType: resource.TiKV, TiDBClusterComponentStatus: oldStatus.TiKV},
		{TiDBCompType: resource.TiDB, TiDBClusterComponentStatus: oldStatus.TiDB},
		{TiDBCompType: resource.TiCDC, TiDBClusterComponentStatus: oldStatus.TiCDC},
	} {
		component := compOldStatus.TiDBCompType
		compSpecGetter, compStatusGetter, err := t.getCompInfoFns(component)
		if err != nil {
			return err
		}
	waitUpdate:
		for {
			if err := t.SyncTiDBCluster(); err != nil {
				return err
			}
			if compSpecGetter().BaseImage == "" {
				// although tide-operator sets the baseImage to default image like pingcap/tidb when spec.version isn't empty(on crd's the default webhook),
				// we'd better check it again here
				return fmt.Errorf("component %s's baseImage isn't set, please check your testbed", component.String())
			}
			switch component {
			case resource.TiDB, resource.TiKV, resource.PD, resource.TiFlash:
				version := opts.Version
				for _, option := range opts.OverrideComponentsUpgradeOption {
					if component == option.Target {
						version = option.Version
					}
				}
				upgradeImage := fmt.Sprintf("%s:%s", compSpecGetter().BaseImage, version)
				if compStatusGetter().Image == upgradeImage && compStatusGetter().Phase == normalPhase {
					break waitUpdate
				}
			case resource.TiCDC:
				if compStatusGetter().Status.CurrentRevision != compOldStatus.Status.CurrentRevision && compStatusGetter().Phase == normalPhase {
					break waitUpdate
				}
			}
			if opts.Timeout != 0 && time.Now().After(now.Add(opts.Timeout)) {
				return ErrUpgradeTimeout
			}
			time.Sleep(5 * time.Second)
		}
	}
	return nil
}

func (t *TiDBCluster) Scale(opts resource.TiDBClusterScaleOptions) error {
	err := t.SyncTiDBCluster()
	if err != nil {
		return err
	}
	compSpecGetter, compStatusGetter, err := t.getCompInfoFns(opts.Target)
	if err != nil {
		return err
	}
	if compSpecGetter().Replicas == int(opts.Replicas) {
		return nil
	}
	err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		err = unstructured.SetNestedField(t.tc.Object, float64(opts.Replicas), "spec", opts.Target.String(), "replicas")
		if err != nil {
			return errors.Trace(err)
		}
		_, updateErr := t.KubeClientSet.DynamicClient.Resource(tcGVR).Namespace(t.Namespace()).Update(context.TODO(), t.tc, metav1.UpdateOptions{})
		if updateErr == nil {
			return nil
		}
		if err = t.SyncTiDBCluster(); err != nil {
			return err
		}
		return updateErr
	})
	if err != nil {
		return errors.Annotate(err, "scale the tidb-cluster error")
	}

	err = wait.PollImmediate(5*time.Second, opts.Timeout, func() (done bool, err error) {
		err = t.SyncTiDBCluster()
		if err != nil {
			return false, err
		}
		status := compStatusGetter()
		return uint32(status.Status.Replicas) == opts.Replicas && uint32(status.Status.ReadyReplicas) == opts.Replicas, nil
	})
	if err != nil {
		return errors.Trace(err)
	}
	if err := t.WaitForReady(opts.Timeout); err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (t *TiDBCluster) WaitForReady(timeout time.Duration) error {
	interval := 10 * time.Second
	err := wait.PollImmediate(interval, timeout, func() (done bool, err error) {
		if err := t.SyncTiDBCluster(); err != nil {
			return false, err
		}
		return t.Ready != nil && *t.Ready, nil
	})
	return errors.Trace(err)
}

func (t *TiDBCluster) SearchLog(ctx context.Context, opts resource.TiDBClusterSearchLogOptions) ([]string, error) {
	var urls []*url.URL
	var err error
	switch opts.Target {
	case resource.TiDB:
		urls, err = t.PeerURLs(resource.DBStatus)
	case resource.PD:
		urls, err = t.PeerURLs(resource.PDAddr)
	case resource.TiKV:
		urls, err = t.PeerURLs(resource.KVAddr)
	case resource.TiFlash:
		urls, err = t.PeerURLs(resource.TiFlashAddr)
	default:
		return nil, fmt.Errorf("unsupported component for log search: %s", opts.Target.String())
	}
	if err != nil {
		return nil, err
	}

	logsClient := logsearch.NewDiagnosticLogClient()
	var logs []string
	for _, url := range urls {
		port, err := strconv.ParseInt(url.Port(), 10, 64)
		if err != nil {
			return nil, err
		}
		l, err := logsClient.SearchLog(ctx, url.Hostname(), int(port), opts.StartTime, opts.EndTime, opts.Levels, opts.Patterns, opts.Limit)
		if err != nil {
			return nil, err
		}
		logs = append(logs, l...)
	}
	return logs, nil
}

func (t *TiDBCluster) ChaosSelect(selector resource.TiDBClusterSelectorOpts) ([]resource.ChaosSelector, error) {
	var selectors []resource.ChaosSelector
	if selector.TiDBClusterTiDBSelectorSpec != nil {
		spec := selector.TiDBClusterTiDBSelectorSpec
		switch {
		case spec.Peer != nil:
			pod, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).Get(context.Background(), fmt.Sprintf("%s-tidb-%d", t.Name(), *spec.Peer), metav1.GetOptions{})
			if err != nil {
				return nil, err
			}
			selectors = append(selectors, &PodSelector{pod})
		case spec.All:
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "tidb"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			for _, tidb := range pods.Items {
				pod := tidb
				selectors = append(selectors, &PodSelector{&pod})
			}
		}
	}
	if selector.TiDBClusterPDSelectorSpec != nil {
		spec := selector.TiDBClusterPDSelectorSpec
		switch {
		case spec.Peer != nil:
			pod, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).Get(context.Background(), fmt.Sprintf("%s-pd-%d", t.Name(), *spec.Peer), metav1.GetOptions{})
			if err != nil {
				return nil, err
			}
			selectors = append(selectors, &PodSelector{pod})
		case spec.All:
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "pd"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			for _, pd := range pods.Items {
				pod := pd
				selectors = append(selectors, &PodSelector{&pod})
			}
		case spec.IsLeader != nil:
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "pd"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			client, err := t.PDClient()
			if err != nil {
				return nil, err
			}
			meminfo, err := client.GetMembers()
			if err != nil {
				return nil, err
			}
			if *spec.IsLeader {
				leader := meminfo.Leader.Name
				for _, pd := range pods.Items {
					if pd.Name == leader {
						pod := pd
						selectors = append(selectors, &PodSelector{&pod})
						break
					}
				}
			} else {
				leader := meminfo.Leader.Name
				for _, pd := range pods.Items {
					if pd.Name != leader {
						pod := pd
						selectors = append(selectors, &PodSelector{&pod})
					}
				}
			}
		}
	}
	if selector.TiDBClusterTiKVSelectorSpec != nil {
		spec := selector.TiDBClusterTiKVSelectorSpec
		switch {
		case spec.Peer != nil:
			pod, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).Get(context.Background(), fmt.Sprintf("%s-tikv-%d", t.Name(), *spec.Peer), metav1.GetOptions{})
			if err != nil {
				return nil, err
			}
			selectors = append(selectors, &PodSelector{pod})
		case spec.StoreLabelSelector != nil:
			selector, err := metav1.LabelSelectorAsSelector(spec.StoreLabelSelector)
			if err != nil {
				return nil, err
			}
			addressLabels, err := t.getAddressLabels()
			if err != nil {
				return nil, err
			}
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "tikv"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}

			for address, labels := range addressLabels {
				if selector.Matches(labels) {
					for _, tikv := range pods.Items {
						pod := tikv
						// for example: tc1-tikv-2.tc1-tikv-peer.testbed-h2r9w.svc:20160
						if fmt.Sprintf("%s.%s-tikv-peer.%s.svc:20160", pod.Name, t.Name(), t.Namespace()) == address {
							selectors = append(selectors, &PodSelector{&pod})
						}
					}
				}
			}
		case spec.All:
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "tikv"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			for _, tikv := range pods.Items {
				pod := tikv
				selectors = append(selectors, &PodSelector{&pod})
			}
		}
	}
	if selector.TiDBClusterTiFlashSelectorSpec != nil {
		spec := selector.TiDBClusterTiFlashSelectorSpec
		switch {
		case spec.Peer != nil:
			pod, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).Get(context.Background(), fmt.Sprintf("%s-tiflash-%d", t.Name(), *spec.Peer), metav1.GetOptions{})
			if err != nil {
				return nil, err
			}
			selectors = append(selectors, &PodSelector{pod})
		case spec.StoreLabelSelector != nil:
			selector, err := metav1.LabelSelectorAsSelector(spec.StoreLabelSelector)
			if err != nil {
				return nil, err
			}
			addressLabels, err := t.getAddressLabels()
			if err != nil {
				return nil, err
			}
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "tiflash"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			for address, labels := range addressLabels {
				if selector.Matches(labels) {
					for _, tiflash := range pods.Items {
						pod := tiflash
						// for example: tc1-tiflash-0.tc1-tiflash-peer.testbed-h2r9w.svc:3930
						if fmt.Sprintf("%s.%s-tiflash-peer.%s.svc:3930", pod.Name, t.Name(), t.Namespace()) == address {
							selectors = append(selectors, &PodSelector{&pod})
						}
					}
				}
			}
		case spec.All:
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "tiflash"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			for _, tiflash := range pods.Items {
				pod := tiflash
				selectors = append(selectors, &PodSelector{&pod})
			}
		}
	}
	if selector.TiDBClusterCDCSelectorSpec != nil {
		spec := selector.TiDBClusterCDCSelectorSpec
		switch {
		case spec.Peer != nil:
			pod, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).Get(context.Background(), fmt.Sprintf("%s-ticdc-%d", t.Name(), *spec.Peer), metav1.GetOptions{})
			if err != nil {
				return nil, err
			}
			selectors = append(selectors, &PodSelector{pod})
		case spec.All:
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "ticdc"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			for _, cdc := range pods.Items {
				pod := cdc
				selectors = append(selectors, &PodSelector{&pod})
			}
		case spec.IsOwner != nil:
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "ticdc"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}

			ownerInfo, err := t.getCDCOwner()
			if err != nil {
				return nil, err
			}
			ownerName := strings.Split(ownerInfo.AdvertiseAddr, ".")[0]
			if *spec.IsOwner {
				for _, item := range pods.Items {
					if item.Name == ownerName {
						selectors = append(selectors, &PodSelector{&item})
						break
					}
				}
			} else {
				for _, item := range pods.Items {
					if item.Name != ownerName {
						selectors = append(selectors, &PodSelector{&item})
					}
				}
			}
		case spec.RandomSameOne:
			pods, err := t.KubeClientSet.ClientSet.CoreV1().Pods(t.Namespace()).List(context.TODO(), metav1.ListOptions{
				LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": "ticdc"}.AsSelector().String(),
			})
			if err != nil {
				return nil, err
			}
			if len(pods.Items) == 0 {
				return nil, fmt.Errorf("cdc-server pod not found")
			}
			target := rand.Intn(len(pods.Items))
			selectors = append(selectors, &PodSelector{Pod: &pods.Items[target]})
		}
	}
	return selectors, nil
}

func (t *TiDBCluster) WatchOOM(ctx context.Context, opts resource.TiDBClusterOOMWatchOptions) (<-chan resource.OOMEvent, error) {
	watcher, err := t.ClientSet.CoreV1().Pods(t.Namespace()).Watch(ctx, metav1.ListOptions{
		LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": opts.Target.String()}.AsSelector().String(),
	})
	if err != nil {
		return nil, errors.Trace(err)
	}

	resultCh := make(chan resource.OOMEvent)
	go func() {
		defer close(resultCh)
		// watcher.ResultChan() will close when ctx is canceled
		for event := range watcher.ResultChan() {
			pod, ok := event.Object.(*corev1.Pod)
			if !ok {
				continue
			}

			if len(pod.Status.ContainerStatuses) == 0 {
				continue
			}

			containerStatuses := pod.Status.ContainerStatuses
			for _, status := range containerStatuses {
				if status.Name != opts.Target.String() {
					// status.Name is name of container, it should match target component name in case there are multiple containers in the pod
					continue
				}

				if status.State.Terminated != nil &&
					(status.State.Terminated.ExitCode == 137 && // process killed by SIGKILL, could be caused by OOM
						status.State.Terminated.Reason == "OOMKilled") { // both ExitCode & Reason should be used, single condition could produce false positive result
					oomEvent := resource.OOMEvent{
						InstanceName: pod.Name,
						InstanceIP:   pod.Status.PodIP,
						HappenedAt:   status.State.Terminated.FinishedAt.Time, // status.State.Terminated.FinishedAt is the time when container is killed
					}

					select {
					case <-ctx.Done():
						return
					case resultCh <- oomEvent:
					}
				}
			}
		}
	}()

	return resultCh, nil
}

func (t *TiDBCluster) Name() string { return t.tc.GetName() }

func (t *TiDBCluster) Namespace() string { return t.tc.GetNamespace() }

// SyncTiDBCluster syncs from the k8s object(tc, pvcs) to apiresource.TiDBCluster and tc fields
func (t *TiDBCluster) SyncTiDBCluster() error {
	var errs error

	tc, err := t.DynamicClient.Resource(tcGVR).Namespace(t.KubeClientSet.Namespace).Get(context.TODO(), t.ResourceHandle.Name, metav1.GetOptions{})
	if err != nil {
		errs = multierror.Append(errs, err)
		tc = nil
	}
	pvcs, err := t.ClientSet.CoreV1().PersistentVolumeClaims(t.KubeClientSet.Namespace).List(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{"app.kubernetes.io/instance": t.ResourceHandle.Name}.AsSelector().String(),
	})
	if err != nil {
		return multierror.Append(errs, err)
	}
	// if all k8s resource created by dmcluster are not found, return an ErrorResourceNotFound error
	if len(pvcs.Items) == 0 && tc == nil {
		return resource.ErrorResourceNotFound{ResourceHandle: t.Handle()}
	}
	if errs != nil {
		return errs
	}

	t.tc = tc
	buf := new(bytes.Buffer)
	err = json.NewEncoder(buf).Encode(tc)
	if err != nil {
		return errors.Trace(err)
	}
	type SpecAndStatus struct {
		Spec   apiresource.TiDBClusterSpec   `json:"spec"`
		Status apiresource.TiDBClusterStatus `json:"status"`
	}
	var specAndStatus SpecAndStatus
	err = json.NewDecoder(buf).Decode(&specAndStatus)
	if err != nil {
		return errors.Trace(err)
	}
	t.TiDBCluster.Spec = specAndStatus.Spec
	t.TiDBCluster.Status = specAndStatus.Status
	if err = t.syncSetupJob(); err != nil {
		return errors.Trace(err)
	}
	// set the ResourceHandle fields
	t.ResourceHandle.Ready = pointer.Bool(t.isReady())
	return nil
}

func (t *TiDBCluster) syncSetupJob() error {
	job, err := t.ClientSet.BatchV1().Jobs(t.Namespace()).Get(context.TODO(), fmt.Sprintf("%s-setup", t.Name()), metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			return nil
		}
		return err
	}
	t.Spec.PostStart.SQL = job.Spec.Template.Spec.Containers[0].Env[0].Value
	t.Status.PostStart.SQL = &apiresource.SQLPostStartHandlerStatus{}
	for _, condition := range job.Status.Conditions {
		switch {
		case condition.Type == batchv1.JobComplete && condition.Status == corev1.ConditionTrue:
			t.Status.PostStart.SQL.State = apiresource.SQLPostStartHandlerStatusCompleted
			return nil
		case condition.Type == batchv1.JobFailed && condition.Status == corev1.ConditionTrue:
			t.Status.PostStart.SQL.State = apiresource.SQLPostStartHandlerStatusError
			return nil
		}
	}
	t.Status.PostStart.SQL.State = apiresource.SQLPostStartHandlerStatusRunning
	return nil
}

func (t *TiDBCluster) isReady() bool {
	if len(t.Status.Conditions) == 0 {
		return false
	}
	for _, cond := range t.Status.Conditions {
		if cond.Type == "Ready" {
			if cond.Status != string(metav1.ConditionTrue) {
				return false
			}
		}
	}
	if t.Spec.TiDB.Replicas == 0 {
		return true
	}

	if config.TiDBCluster.ReadyCheck.PingDB {
		if err := pingDB(t); err != nil {
			return false
		}
	}
	comps, err := t.ClientSet.AppsV1().StatefulSets(t.Namespace()).List(context.TODO(), metav1.ListOptions{
		LabelSelector: labels.Set{"app.kubernetes.io/instance": t.Name()}.AsSelector().String(),
	})
	if err != nil {
		sugar.Errorf("list sts error: %v", err)
		return false
	}
	for _, comp := range comps.Items {
		if comp.Status.ReadyReplicas != comp.Status.Replicas {
			return false
		}
	}
	if t.Status.PostStart.SQL != nil && t.Status.PostStart.SQL.State != apiresource.SQLPostStartHandlerStatusCompleted {
		return false
	}
	return true
}

func (t *TiDBCluster) GetSpec(comp resource.TiDBCompType) (*apiresource.TiDBClusterComponentSpec, error) {
	specGetter, _, err := t.getCompInfoFns(comp)
	if err != nil {
		return nil, err
	}
	spec := specGetter()
	return &spec, nil
}

func (t *TiDBCluster) GetStatus(comp resource.TiDBCompType) (*apiresource.TiDBClusterComponentStatus, error) {
	if err := t.SyncTiDBCluster(); err != nil {
		return nil, err
	}
	_, statusGetter, err := t.getCompInfoFns(comp)
	if err != nil {
		return nil, err
	}
	status := statusGetter()
	return &status, nil
}

func (t *TiDBCluster) getCompInfoFns(comp resource.TiDBCompType) (componentSpecGetter func() apiresource.TiDBClusterComponentSpec, componentStatusGetter func() apiresource.TiDBClusterComponentStatus, err error) {
	switch comp {
	case resource.PD:
		componentSpecGetter = func() apiresource.TiDBClusterComponentSpec { return t.Spec.PD }
		componentStatusGetter = func() apiresource.TiDBClusterComponentStatus { return t.Status.PD }
	case resource.TiKV:
		componentSpecGetter = func() apiresource.TiDBClusterComponentSpec { return t.Spec.TiKV }
		componentStatusGetter = func() apiresource.TiDBClusterComponentStatus { return t.Status.TiKV }
	case resource.TiDB:
		componentSpecGetter = func() apiresource.TiDBClusterComponentSpec { return t.Spec.TiDB }
		componentStatusGetter = func() apiresource.TiDBClusterComponentStatus { return t.Status.TiDB }
	case resource.TiCDC:
		componentSpecGetter = func() apiresource.TiDBClusterComponentSpec { return t.Spec.TiCDC }
		componentStatusGetter = func() apiresource.TiDBClusterComponentStatus { return t.Status.TiCDC }
	case resource.TiFlash:
		componentSpecGetter = func() apiresource.TiDBClusterComponentSpec { return t.Spec.TiFlash }
		componentStatusGetter = func() apiresource.TiDBClusterComponentStatus { return t.Status.TiFlash }
	default:
		err = errors.New("unknown component: " + comp.String())
	}
	return
}

func (t *TiDBCluster) GetDB(dbName string) (db *sql.DB, err error) {
	return getDB(t, dbName)
}

type tidbClusterPod struct {
	cluster  *TiDBCluster
	podName  string
	compType resource.TiDBCompType
}

func (p *tidbClusterPod) Restart() error {
	stdOut, stdErr, code, err := p.cluster.ExecCommandInContainerWithFullOutput(p.podName, p.compType.String(), "/bin/sh", "-c", "kill 1")
	if err != nil {
		return errors.Trace(err)
	}
	if *code != 0 {
		return errors.Trace(fmt.Errorf("run cmd failed with code %d, \nstdOut:\n%s, stdErr:%s\n", *code, stdOut, stdErr))
	}
	return nil
}

func (p *tidbClusterPod) CleanData() error {
	selector := labels.Set{"tidb.pingcap.com/pod-name": p.podName}.AsSelector().String()
	err := p.cluster.ClientSet.CoreV1().PersistentVolumeClaims(p.cluster.Namespace()).DeleteCollection(
		context.TODO(), metav1.DeleteOptions{}, metav1.ListOptions{LabelSelector: selector})
	if err != nil {
		return err
	}

	// TODO(@lobshunter): upgrade kubernetes version to carry statefulset controller fix: https://github.com/kubernetes/kubernetes/pull/93457
	// so code below can be simplify to a single DELETE API call
	err = p.cluster.DeletePod(p.podName)
	if err != nil {
		return err
	}
	return wait.PollInfinite(30*time.Second, func() (done bool, err error) {
		// for now we need this horrific code to ensure PVCs of pod are successfully created
		// or it can be blocked in pending state due to lack of pvc
		pod, err := p.cluster.ClientSet.CoreV1().Pods(p.cluster.Namespace()).Get(
			context.TODO(), p.podName, metav1.GetOptions{})
		if err != nil {
			return false, err
		}

		if pod.Status.Phase != corev1.PodPending {
			return true, nil
		}

		for _, volume := range pod.Spec.Volumes {
			if volume.PersistentVolumeClaim == nil {
				continue
			}

			_, err := p.cluster.ClientSet.CoreV1().PersistentVolumeClaims(p.cluster.Namespace()).Get(
				context.TODO(), volume.PersistentVolumeClaim.ClaimName, metav1.GetOptions{})
			switch {
			case apierrors.IsNotFound(err):
				err2 := p.cluster.ClientSet.CoreV1().Pods(p.cluster.Namespace()).Delete(context.Background(), pod.Name, metav1.DeleteOptions{})
				switch {
				case apierrors.IsNotFound(err2):
					return false, nil
				case err2 != nil:
					return false, err2
				}
			case err != nil:
				return false, err
			}
		}

		return false, nil
	})
}

func (t *TiDBCluster) GetNodes(comp resource.TiDBCompType) ([]resource.TiDBComponentNode, error) {
	pods, err := t.ListPods(map[string]string{"app.kubernetes.io/instance": t.Name(), "app.kubernetes.io/component": comp.String()})
	if err != nil {
		return nil, errors.Trace(err)
	}
	var nodes []resource.TiDBComponentNode
	for _, podName := range pods {
		nodes = append(nodes, &tidbClusterPod{
			cluster:  t,
			podName:  podName,
			compType: comp,
		})
	}
	return nodes, nil
}

func (t *TiDBCluster) getAddressLabels() (map[string]labels.Set, error) {
	dbURL, err := t.ServiceURL(resource.DBAddr)
	if err != nil {
		return nil, err
	}
	db, err := sql.Open("mysql", resource.BuildDSN(dbURL, "INFORMATION_SCHEMA"))
	if err != nil {
		return nil, err
	}
	rows, err := db.Query(`select ADDRESS, LABEL from TIKV_STORE_STATUS`)
	if err != nil {
		return nil, err
	}
	// address to labels
	addressLabels := make(map[string]labels.Set)

	type rowInfo struct {
		Key   string `json:"key"`
		Value string `json:"value"`
	}
	for rows.Next() {
		var address, rawLabel string
		if err := rows.Scan(&address, &rawLabel); err != nil {
			return nil, err
		}
		var r []rowInfo
		if err := json.Unmarshal([]byte(rawLabel), &r); err != nil {
			return nil, err
		}
		if _, ok := addressLabels[address]; !ok {
			addressLabels[address] = map[string]string{}
		}
		for _, item := range r {
			addressLabels[address][item.Key] = item.Value
		}
	}
	return addressLabels, nil
}

type captureInfo struct {
	ID            string `json:"id"`
	IsOwner       bool   `json:"is-owner"`
	AdvertiseAddr string `json:"address"`
}

type captureList []*captureInfo

func (t *TiDBCluster) getCDCCaptureInfo() (captureList, error) {
	pdURL, err := t.ServiceURL(resource.PDAddr)
	if err != nil {
		return nil, err
	}

	command := "/cdc cli capture list --pd=" + pdURL.String()
	stdout, stderr, ec, err := t.Exec(resource.TiDBClusterExecOptions{
		Target:  resource.TiCDC,
		Command: command,
		Peer:    0,
	})
	if err != nil {
		return nil, err
	}
	if ec != 0 {
		return nil, fmt.Errorf("get cdc capture list failed, exitCode=%+v, stderr=%+v", ec, stderr)
	}
	var result captureList
	if err := json.Unmarshal([]byte(stdout), &result); err != nil {
		return nil, err
	}
	return result, nil
}

func (t *TiDBCluster) getCDCOwner() (*captureInfo, error) {
	allCaptures, err := t.getCDCCaptureInfo()
	if err != nil {
		return nil, err
	}
	for _, item := range allCaptures {
		if item.IsOwner {
			return item, nil
		}
	}
	return nil, fmt.Errorf("get cdc owner failed")
}