// 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 (
	"fmt"
	"reflect"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	tsctl "github.com/pingcap/tidb-operator/pkg/tidbset/controller"
	"github.com/pingcap/tidb-operator/pkg/util"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
)

func (tcc *TiDBClusterController) syncTidbSetFor(tc *v1.TidbCluster) error {
	ns := tc.GetNamespace()
	newPDSet, err := tcc.getNewPdSetFor(tc)
	if err != nil {
		return errors.Trace(err)
	}
	newTiKVSet, err := tcc.getNewTikvSetFor(tc)
	if err != nil {
		return errors.Trace(err)
	}
	newTiDBSet, err := tcc.getNewTidbSetFor(tc)
	if err != nil {
		return errors.Trace(err)
	}

	sets := tcc.cli.PingcapV1().TidbSets(ns)
	oldPDSet, err := sets.Get(pdSetNameFor(tc), metav1.GetOptions{})
	if err != nil {
		if !apierrors.IsNotFound(err) {
			return errors.Trace(err)
		}
		oldPDSet = newPDSet
	}
	oldTiKVSet, err := sets.Get(tikvSetNameFor(tc), metav1.GetOptions{})
	if err != nil {
		if !apierrors.IsNotFound(err) {
			return errors.Trace(err)
		}
		oldTiKVSet = newTiKVSet
	}
	oldTiDBSet, err := sets.Get(tidbSetNameFor(tc), metav1.GetOptions{})
	if err != nil {
		if !apierrors.IsNotFound(err) {
			return errors.Trace(err)
		}
		oldTiDBSet = newTiDBSet
	}

	pdShouldUpgrade := tc.Spec.PD.Image != oldPDSet.ImageFor(v1.PDContainerType)
	tikvShouldUpgrade := tc.Spec.TiKV.Image != oldTiKVSet.ImageFor(v1.TiKVContainerType)
	tidbShouldUpgrade := tc.Spec.TiDB.Image != oldTiDBSet.ImageFor(v1.TiDBContainerType)

	if pdShouldUpgrade || tc.PDIsUpgrading() {
		if pdShouldUpgrade {
			if err := tcc.setUpgradingFor(tc, v1.PDContainerType); err != nil {
				return errors.Trace(err)
			}
		}
		tcc.revertTiKVSetImage(newTiKVSet, oldTiKVSet)
		tcc.revertTiDBSetImage(newTiDBSet, oldTiDBSet)
	} else if tikvShouldUpgrade || tc.TiKVIsUpgrading() {
		if tikvShouldUpgrade {
			if err := tcc.setUpgradingFor(tc, v1.TiKVContainerType); err != nil {
				return errors.Trace(err)
			}
		}
		tcc.revertTiDBSetImage(newTiDBSet, oldTiDBSet)
	} else if tidbShouldUpgrade {
		if err := tcc.setUpgradingFor(tc, v1.TiDBContainerType); err != nil {
			return errors.Trace(err)
		}
	}

	if err := tcc.updateTidbSet(newPDSet); err != nil {
		return errors.Trace(err)
	}
	if err := tcc.updateTidbSet(newTiKVSet); err != nil {
		return errors.Trace(err)
	}
	return tcc.updateTidbSet(newTiDBSet)
}

func (tcc *TiDBClusterController) syncUpgradeStatusesFor(tc *v1.TidbCluster) error {
	err := tcc.syncUpgradeStatusFor(tc, v1.PDContainerType)
	if err != nil {
		return errors.Trace(err)
	}
	err = tcc.syncUpgradeStatusFor(tc, v1.TiKVContainerType)
	if err != nil {
		return errors.Trace(err)
	}
	return tcc.syncUpgradeStatusFor(tc, v1.TiDBContainerType)
}

func (tcc *TiDBClusterController) syncUpgradeStatusFor(tc *v1.TidbCluster, ctype v1.ContainerType) error {
	if !tc.PDIsUpgrading() && !tc.TiKVIsUpgrading() && !tc.TiDBIsUpgrading() {
		return nil
	}

	ns := tc.GetNamespace()
	clusterName := tc.GetName()
	pods, err := tcc.getPodsOf(tc, ctype)
	if err != nil {
		return errors.Trace(err)
	}

	ts, err := tcc.cli.PingcapV1().TidbSets(ns).Get(fmt.Sprintf("%s-%sset", clusterName, ctype.String()), metav1.GetOptions{})
	if err != nil {
		return errors.Trace(err)
	}

	if allUpgraded(tc, ts, pods, ctype) {
		switch ctype {
		case v1.PDContainerType:
			tc.Status.PDClusterStatus.Upgrading = false
			tc.Status.PDClusterStatus.LastTransitionTime = metav1.Now()
		case v1.TiKVContainerType:
			tc.Status.TiKVClusterStatus.Upgrading = false
			tc.Status.TiKVClusterStatus.LastTransitionTime = metav1.Now()
		case v1.TiDBContainerType:
			tc.Status.TiDBClusterStatus.Upgrading = false
			tc.Status.TiDBClusterStatus.LastTransitionTime = metav1.Now()
		}

		_, err := tcc.cli.PingcapV1().TidbClusters(ns).Update(tc)
		if err != nil {
			log.Errorf("failed to update TidbCluster: [%s/%s], %v", ns, clusterName, err)
			return errors.Trace(err)
		}
		log.Infof("update TidbCluster: [%s/%s] successfullyl", ns, clusterName)
		return nil
	}

	return nil
}

func (tcc *TiDBClusterController) getPodsOf(tc *v1.TidbCluster, ctype v1.ContainerType) ([]corev1.Pod, error) {
	ns := tc.GetNamespace()
	clusterName := tc.GetName()
	listOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Cluster(clusterName).App(ctype.String()).Labels(),
		).String(),
	}
	podList, err := tcc.kubeCli.CoreV1().Pods(ns).List(listOptions)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return podList.Items, nil
}

func (tcc *TiDBClusterController) setUpgradingFor(tc *v1.TidbCluster, ctype v1.ContainerType) error {
	ns := tc.GetNamespace()
	clusterName := tc.GetName()
	switch ctype {
	case v1.PDContainerType:
		tc.Status.PDClusterStatus.Upgrading = true
		tc.Status.PDClusterStatus.LastTransitionTime = metav1.Now()
	case v1.TiKVContainerType:
		tc.Status.TiKVClusterStatus.Upgrading = true
		tc.Status.TiKVClusterStatus.LastTransitionTime = metav1.Now()
	case v1.TiDBContainerType:
		tc.Status.TiDBClusterStatus.Upgrading = true
		tc.Status.TiDBClusterStatus.LastTransitionTime = metav1.Now()
	}
	_, err := tcc.cli.PingcapV1().TidbClusters(ns).Update(tc)
	if err != nil {
		log.Errorf("failed to update TidbCluster: [%s/%s], %v", ns, clusterName, err)
		return errors.Trace(err)
	}
	log.Infof("update TidbCluster: [%s/%s] successfullyl", ns, clusterName)
	return nil
}

func (tcc *TiDBClusterController) revertTiKVSetImage(newTiKVSet, oldTiKVSet *v1.TidbSet) {
	oldImage := tsctl.GetImageByPodSpec(v1.TiKVContainerType, &oldTiKVSet.Spec.Template.Spec)
	tsctl.SetImageByPodSpec(v1.TiKVContainerType, &newTiKVSet.Spec.Template.Spec, oldImage)
}

func (tcc *TiDBClusterController) revertTiDBSetImage(newTiDBSet, oldTiDBSet *v1.TidbSet) {
	oldImage := tsctl.GetImageByPodSpec(v1.TiDBContainerType, &oldTiDBSet.Spec.Template.Spec)
	tsctl.SetImageByPodSpec(v1.TiDBContainerType, &newTiDBSet.Spec.Template.Spec, oldImage)
}

func (tcc *TiDBClusterController) getNewPdSetFor(tc *v1.TidbCluster) (*v1.TidbSet, error) {
	ns := tc.GetNamespace()
	clusterName := tc.GetName()
	cmds := []string{
		"/entrypoint.sh",
	}
	args := []string{
		"--data-dir=/var/lib/tidb/pd",
		"--name=$(MY_POD_NAME)",
		"--peer-urls=http://0.0.0.0:2380",
		fmt.Sprintf("--advertise-peer-urls=http://$(MY_POD_NAME).%s-pd-peer.%s.svc:2380", clusterName, ns),
		"--client-urls=http://0.0.0.0:2379",
		fmt.Sprintf("--advertise-client-urls=http://$(MY_POD_NAME).%s-pd-peer.%s.svc:2379", clusterName, ns),
	}
	volMounts := []corev1.VolumeMount{
		{Name: v1.TiDBVolumeName, MountPath: "/var/lib/tidb"},
		{Name: "timezone", ReadOnly: true, MountPath: "/etc/localtime"},
		{Name: "labels", ReadOnly: true, MountPath: "/etc/k8s-pod-info"},
	}
	vols := []corev1.Volume{
		{Name: v1.TiDBVolumeName, VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "PVC-NAME"}}},
		{Name: "timezone", VolumeSource: corev1.VolumeSource{HostPath: &corev1.HostPathVolumeSource{Path: "/etc/localtime"}}},
		{Name: "labels", VolumeSource: corev1.VolumeSource{
			DownwardAPI: &corev1.DownwardAPIVolumeSource{
				Items: []corev1.DownwardAPIVolumeFile{
					{
						Path:     "labels",
						FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.labels"},
					},
				},
			},
		}},
	}
	cmds = append(cmds, "--config=/etc/tidb/pd.toml")
	volMounts = append(volMounts, corev1.VolumeMount{Name: "config", ReadOnly: true, MountPath: "/etc/tidb"})
	vols = append(vols, corev1.Volume{
		Name: "config",
		VolumeSource: corev1.VolumeSource{
			ConfigMap: &corev1.ConfigMapVolumeSource{
				LocalObjectReference: corev1.LocalObjectReference{Name: tc.GetConfigMapName()},
				Items:                []corev1.KeyToPath{{Key: "pd-config", Path: "pd.toml"}},
			},
		}},
	)

	ann := map[string]string{}
	if tc.Spec.PD.Requests != nil {
		size := tc.Spec.PD.Requests.Storage
		if q, err := resource.ParseQuantity(size); err != nil {
			log.Warnf("cant' get storage size: %s for TidbCluster: %s/%s, %v", size, ns, clusterName, err)
		} else {
			ann[v1.AnnotationStorageSize] = q.String()
		}
	}

	pdset := &v1.TidbSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:            fmt.Sprintf("%s-pdset", clusterName),
			Namespace:       ns,
			Labels:          label.New().Cluster(clusterName).PD().Labels(),
			OwnerReferences: []metav1.OwnerReference{getOwnerRef(tc)},
			Annotations:     util.AnnPaused(ann, tc),
		},
		Spec: v1.TidbSetSpec{
			Replicas: int32(tc.Spec.PD.Size),
			Selector: label.New().Cluster(clusterName).PD().Selector(),
			Template: &corev1.PodTemplateSpec{
				Spec: corev1.PodSpec{
					Affinity: util.AffinityForNodeSelector(
						ns,
						tc.Spec.PD.NodeSelectorRequired,
						label.New().Cluster(clusterName).PD(),
						tc.Spec.PD.NodeSelector,
					),
					Subdomain: pdPeerSvcName(clusterName),
					Containers: []corev1.Container{
						{
							Name:    v1.PDContainerType.String(),
							Image:   tc.Spec.PD.Image,
							Command: cmds,
							Args:    args,
							Ports: []corev1.ContainerPort{
								{
									Name:          "server",
									ContainerPort: int32(2380),
									Protocol:      corev1.ProtocolTCP,
								},
								{
									Name:          "client",
									ContainerPort: int32(2379),
									Protocol:      corev1.ProtocolTCP,
								},
								{
									Name:          "metrics", // This is for metrics pull
									ContainerPort: int32(2379),
									Protocol:      corev1.ProtocolTCP,
								},
							},
							VolumeMounts: volMounts,
							Resources:    util.ResourceRequirement(tc.Spec.PD.ContainerSpec),
							Env: []corev1.EnvVar{
								{
									Name: "MY_POD_NAME",
									ValueFrom: &corev1.EnvVarSource{
										FieldRef: &corev1.ObjectFieldSelector{
											FieldPath: "metadata.name",
										},
									},
								},
							},
						},
					},
					RestartPolicy: corev1.RestartPolicyAlways,
					Volumes:       vols,
				},
			},
		},
	}
	return pdset, nil
}

func (tcc *TiDBClusterController) getNewTikvSetFor(tc *v1.TidbCluster) (*v1.TidbSet, error) {
	ns := tc.GetNamespace()
	clusterName := tc.GetName()
	cmds := []string{
		"/entrypoint.sh",
	}
	args := []string{
		fmt.Sprintf("--pd=%s-pd:2379", clusterName),
		"--addr=0.0.0.0:20160",
		fmt.Sprintf("--advertise-addr=$(MY_POD_NAME).%s-tikv-peer.%s.svc:20160", clusterName, ns),
		"--data-dir=/var/lib/tidb/tikv",
		tikvCapacityArgs(tc.Spec.TiKV.Limits),
	}
	volMounts := []corev1.VolumeMount{
		{Name: v1.TiDBVolumeName, MountPath: "/var/lib/tidb"},
		{Name: "timezone", ReadOnly: true, MountPath: "/etc/localtime"},
		{Name: "labels", ReadOnly: true, MountPath: "/etc/k8s-pod-info"},
	}
	vols := []corev1.Volume{
		{Name: "timezone", VolumeSource: corev1.VolumeSource{HostPath: &corev1.HostPathVolumeSource{Path: "/etc/localtime"}}},
		{Name: v1.TiDBVolumeName, VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "PVC-NAME"}}},
		{Name: "labels", VolumeSource: corev1.VolumeSource{
			DownwardAPI: &corev1.DownwardAPIVolumeSource{
				Items: []corev1.DownwardAPIVolumeFile{
					{
						Path:     "labels",
						FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.labels"},
					},
				},
			},
		}},
	}
	cmds = append(cmds, "--config=/etc/tidb/tikv.toml")
	volMounts = append(volMounts, corev1.VolumeMount{Name: "config", ReadOnly: true, MountPath: "/etc/tidb"})
	vols = append(vols, corev1.Volume{
		Name: "config",
		VolumeSource: corev1.VolumeSource{
			ConfigMap: &corev1.ConfigMapVolumeSource{
				LocalObjectReference: corev1.LocalObjectReference{Name: tc.GetConfigMapName()},
				Items:                []corev1.KeyToPath{{Key: "tikv-config", Path: "tikv.toml"}},
			},
		},
	})

	ann := map[string]string{}

	if tc.Spec.TiKV.Requests != nil {
		size := tc.Spec.TiKV.Requests.Storage
		if q, err := resource.ParseQuantity(size); err != nil {
			log.Warnf("cant' get storage size: %s for TidbCluster: %s/%s, %v", size, ns, clusterName, err)
		} else {
			ann[v1.AnnotationStorageSize] = q.String()
		}
	}

	tikvset := &v1.TidbSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:            fmt.Sprintf("%s-tikvset", clusterName),
			Namespace:       ns,
			Labels:          label.New().Cluster(clusterName).TiKV(),
			OwnerReferences: []metav1.OwnerReference{getOwnerRef(tc)},
			Annotations:     util.AnnPaused(ann, tc),
		},
		Spec: v1.TidbSetSpec{
			Replicas: int32(tc.Spec.TiKV.Size),
			Selector: label.New().Cluster(clusterName).TiKV().Selector(),
			Template: &corev1.PodTemplateSpec{
				Spec: corev1.PodSpec{
					Affinity: util.AffinityForNodeSelector(
						ns,
						tc.Spec.TiKV.NodeSelectorRequired,
						label.New().Cluster(clusterName).TiKV(),
						tc.Spec.TiKV.NodeSelector,
					),
					Subdomain: tikvPeerSvcName(clusterName),
					Containers: []corev1.Container{
						{
							Name:    v1.TiKVContainerType.String(),
							Image:   tc.Spec.TiKV.Image,
							Command: cmds,
							Args:    args,
							Ports: []corev1.ContainerPort{
								{
									Name:          "server",
									ContainerPort: int32(20160),
									Protocol:      corev1.ProtocolTCP,
								},
							},
							VolumeMounts: volMounts,
							Resources:    util.ResourceRequirement(tc.Spec.TiKV.ContainerSpec),
							Env: []corev1.EnvVar{
								{
									Name: "MY_POD_NAME",
									ValueFrom: &corev1.EnvVarSource{
										FieldRef: &corev1.ObjectFieldSelector{
											FieldPath: "metadata.name",
										},
									},
								},
							},
						},
						{
							Name:  v1.PushGatewayContainerType.String(),
							Image: getPushgatewayImage(tc),
							Ports: []corev1.ContainerPort{
								{
									Name:          "metrics",
									ContainerPort: int32(9091),
									Protocol:      corev1.ProtocolTCP,
								},
							},
							VolumeMounts: []corev1.VolumeMount{{Name: "timezone", ReadOnly: true, MountPath: "/etc/localtime"}},
							Resources: corev1.ResourceRequirements{
								Requests: map[corev1.ResourceName]resource.Quantity{
									corev1.ResourceCPU:    resource.MustParse("50m"),
									corev1.ResourceMemory: resource.MustParse("50Mi"),
								},
								Limits: map[corev1.ResourceName]resource.Quantity{
									corev1.ResourceCPU:    resource.MustParse("100m"),
									corev1.ResourceMemory: resource.MustParse("100Mi"),
								},
							},
						},
					},
					RestartPolicy: corev1.RestartPolicyAlways,
					Volumes:       vols,
				},
			},
		},
	}
	return tikvset, nil
}

func (tcc *TiDBClusterController) getNewTidbSetFor(tc *v1.TidbCluster) (*v1.TidbSet, error) {
	ns := tc.GetNamespace()
	clusterName := tc.GetName()
	cmName := tc.GetConfigMapName()
	cm, err := tcc.kubeCli.CoreV1().ConfigMaps(ns).Get(cmName, metav1.GetOptions{})
	if err != nil {
		return nil, errors.Trace(err)
	}

	var enableBinlog bool
	if tc.Spec.TiDB.Binlog != nil && len(cm.Data["pump-config"]) != 0 {
		enableBinlog = true
	}

	tidbImage := tc.Spec.TiDB.Image
	tidbRR := util.ResourceRequirement(tc.Spec.TiDB.ContainerSpec)

	containers := []corev1.Container{
		getTiDBContainer(clusterName, tidbImage, tidbRR, enableBinlog),
	}
	if enableBinlog {
		binlogImage := tc.Spec.TiDB.Binlog.Image
		binlogRR := util.ResourceRequirement(*tc.Spec.TiDB.Binlog)
		binlogContainer := getBinlogContainer(clusterName, binlogImage, binlogRR)
		containers = append(containers, binlogContainer)
	}
	vols := getTiDBVolumes(tc.GetConfigMapName(), enableBinlog)

	tidbset := &v1.TidbSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:            fmt.Sprintf("%s-tidbset", clusterName),
			Namespace:       ns,
			Labels:          label.New().Cluster(clusterName).TiDB(),
			OwnerReferences: []metav1.OwnerReference{getOwnerRef(tc)},
			Annotations:     util.AnnPaused(nil, tc),
		},
		Spec: v1.TidbSetSpec{
			Replicas: int32(tc.Spec.TiDB.Size),
			Selector: label.New().Cluster(clusterName).TiDB().Selector(),
			Template: &corev1.PodTemplateSpec{
				Spec: corev1.PodSpec{
					Affinity: util.AffinityForNodeSelector(
						ns,
						tc.Spec.TiDB.NodeSelectorRequired,
						nil,
						tc.Spec.TiDB.NodeSelector,
					),
					Containers:    containers,
					RestartPolicy: corev1.RestartPolicyAlways,
					Volumes:       vols,
				},
			},
		},
	}
	return tidbset, nil
}

func (tcc *TiDBClusterController) updateTidbSet(newTidbSet *v1.TidbSet) error {
	ns := newTidbSet.GetNamespace()
	tsName := newTidbSet.GetObjectMeta().GetName()
	ts, err := tcc.cli.PingcapV1().TidbSets(ns).Get(tsName, metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			_, err = tcc.cli.PingcapV1().TidbSets(ns).Create(newTidbSet)
			if err != nil {
				log.Errorf("TidbSet[%s/%s] created failed: %v", ns, tsName, err)
				return errors.Trace(err)
			}
			log.Infof("TidbSet[%s/%s] created successfully", ns, tsName)
			return nil
		}
		return errors.Trace(err)
	}

	if ts.Annotations == nil {
		ts.Annotations = make(map[string]string)
	}

	if ts.Paused() == newTidbSet.Paused() && reflect.DeepEqual(ts.Spec, newTidbSet.Spec) {
		return nil
	}

	log.Infof("TidbSet: [%s/%s] Replicas before = %d after = %d",
		ns, tsName, ts.Spec.Replicas, newTidbSet.Spec.Replicas)
	ts.Spec = newTidbSet.Spec
	ts.Annotations[label.AnnPaused] = newTidbSet.Annotations[label.AnnPaused]

	_, err = tcc.cli.PingcapV1().TidbSets(ns).Update(ts)
	if err != nil {
		log.Errorf("TidbSet[%s/%s] updated failed: %v", ns, tsName, err)
		return errors.Trace(err)
	}

	log.Infof("TidbSet[%s/%s] updated successfully", ns, tsName)
	return nil
}

func getTiDBVolumes(configmapName string, enableBinlog bool) []corev1.Volume {
	items := []corev1.KeyToPath{{Key: "tidb-config", Path: "tidb.toml"}}
	if enableBinlog {
		items = append(items, corev1.KeyToPath{Key: "pump-config", Path: "pump.toml"})
	}
	vols := []corev1.Volume{
		{Name: "timezone", VolumeSource: corev1.VolumeSource{HostPath: &corev1.HostPathVolumeSource{Path: "/etc/localtime"}}},
		{Name: "config", VolumeSource: corev1.VolumeSource{
			ConfigMap: &corev1.ConfigMapVolumeSource{
				LocalObjectReference: corev1.LocalObjectReference{
					Name: configmapName,
				},
				Items: items,
			},
		},
		},
		{Name: "labels", VolumeSource: corev1.VolumeSource{
			DownwardAPI: &corev1.DownwardAPIVolumeSource{
				Items: []corev1.DownwardAPIVolumeFile{
					{
						Path:     "labels",
						FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.labels"},
					},
				},
			},
		}},
	}
	if enableBinlog {
		vols = append(vols, corev1.Volume{Name: "varrun", VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}})
	}
	return vols
}

func getTiDBContainer(clusterName string, image string, rr corev1.ResourceRequirements, enableBinlog bool) corev1.Container {
	cmds := []string{
		"/entrypoint.sh",
	}
	args := []string{
		"--store=tikv",
		"--host=0.0.0.0",
		fmt.Sprintf("--path=%s-pd:2379", clusterName),
		"--config=/etc/tidb/tidb.toml",
	}
	volMounts := []corev1.VolumeMount{
		{Name: "timezone", ReadOnly: true, MountPath: "/etc/localtime"},
		{Name: "config", ReadOnly: true, MountPath: "/etc/tidb"},
		{Name: "labels", ReadOnly: true, MountPath: "/etc/k8s-pod-info"},
	}
	if enableBinlog {
		args = append(args, "--binlog-socket=/var/run/tidb/pump.sock")
		volMounts = append(volMounts, corev1.VolumeMount{Name: "varrun", MountPath: "/var/run/tidb"})
	}
	return corev1.Container{
		Name:    v1.TiDBContainerType.String(),
		Image:   image,
		Command: cmds,
		Args:    args,
		Ports: []corev1.ContainerPort{
			{
				Name:          "server",
				ContainerPort: int32(4000),
				Protocol:      corev1.ProtocolTCP,
			},
			{
				Name:          "status", // pprof, status, metrics
				ContainerPort: int32(10080),
				Protocol:      corev1.ProtocolTCP,
			},
			{
				Name:          "metrics", // This is used for metrics pull
				ContainerPort: int32(10080),
				Protocol:      corev1.ProtocolTCP,
			},
		},
		VolumeMounts: volMounts,
		Resources:    rr,
	}
}

func getBinlogContainer(clusterName string, image string, rr corev1.ResourceRequirements) corev1.Container {
	cmds := []string{
		"/pump",
	}
	args := []string{
		"--addr=0.0.0.0:8250",
		"--advertise-addr=$(MY_POD_IP):8250",
		fmt.Sprintf("--pd-urls=http://%s-pd:2379", clusterName),
		"--config=/etc/tidb/pump.toml",
	}
	volMounts := []corev1.VolumeMount{
		{Name: "timezone", ReadOnly: true, MountPath: "/etc/localtime"},
		{Name: "varrun", MountPath: "/var/run/tidb"},
		{Name: "config", ReadOnly: true, MountPath: "/etc/tidb"},
	}

	return corev1.Container{
		Name:    v1.BinlogContainerType.String(),
		Image:   image,
		Command: cmds,
		Args:    args,
		Ports: []corev1.ContainerPort{
			{
				Name:          "pump",
				ContainerPort: int32(8250),
				Protocol:      corev1.ProtocolTCP,
			},
			{
				Name:          "metrics", // This is used for metrics pull
				ContainerPort: int32(8250),
				Protocol:      corev1.ProtocolTCP,
			},
		},
		VolumeMounts: volMounts,
		Resources:    rr,
	}
}

func allUpgraded(tc *v1.TidbCluster, ts *v1.TidbSet, pods []corev1.Pod, ctype v1.ContainerType) bool {
	var newImage string
	switch ctype {
	case v1.PDContainerType:
		newImage = tc.Spec.PD.Image
	case v1.TiKVContainerType:
		newImage = tc.Spec.TiKV.Image
	case v1.TiDBContainerType:
		newImage = tc.Spec.TiDB.Image
	}
	for _, pod := range pods {
		var container *corev1.Container
		for _, c := range pod.Spec.Containers {
			if c.Name == ctype.String() {
				container = &c
				break
			}
		}
		if container == nil {
			return false
		}
		if container.Image != newImage {
			return false
		}
		if !ts.IsReady(&pod) {
			return false
		}

		var cs *corev1.ContainerStatus
		for _, c := range pod.Status.ContainerStatuses {
			if c.Name == ctype.String() {
				cs = &c
				break
			}
		}
		if cs == nil {
			return false
		}

		running := cs.State.Running
		if running == nil {
			return false
		}

		switch ctype {
		case v1.PDContainerType:
			if running.StartedAt.Before(&tc.Status.PDClusterStatus.LastTransitionTime) {
				return false
			}
		case v1.TiKVContainerType:
			if running.StartedAt.Before(&tc.Status.TiKVClusterStatus.LastTransitionTime) {
				return false
			}
		case v1.TiDBContainerType:
			if running.StartedAt.Before(&tc.Status.TiDBClusterStatus.LastTransitionTime) {
				return false
			}
		}

		if running.StartedAt.Add(30 * time.Second).After(time.Now()) {
			return false
		}
	}

	return true

}