// Copyright 2018 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 member

import (
	"fmt"
	"strconv"
	"time"

	"github.com/pingcap/advanced-statefulset/client/apis/apps/v1/helper"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/controller"
	"github.com/pingcap/tidb-operator/pkg/features"
	"github.com/pingcap/tidb-operator/pkg/manager/coordinator"
	mngerutils "github.com/pingcap/tidb-operator/pkg/manager/utils"
	"github.com/pingcap/tidb-operator/pkg/manager/volumes"

	apps "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	errorutils "k8s.io/apimachinery/pkg/util/errors"
	"k8s.io/klog/v2"
	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
)

const (
	// EvictLeaderBeginTime is the key of evict Leader begin time
	EvictLeaderBeginTime = "evictLeaderBeginTime"

	// EvictLeaderFinishTime is the key of evict leader finish time
	EvictLeaderFinishTime = "evictLeaderFinishTime"

	// LeaderCountBeforeEvictLeader is the key of leader count before evict leader
	LeaderCountBeforeEvictLeader = "leaderCountBeforeEvictLeader"

	// TODO: change to use minReadySeconds in sts spec
	// See https://kubernetes.io/blog/2021/08/27/minreadyseconds-statefulsets/
	annoKeyTiKVMinReadySeconds = "tidb.pingcap.com/tikv-min-ready-seconds"
)

type TiKVUpgrader interface {
	Upgrade(meta metav1.Object, oldSet *apps.StatefulSet, newSet *apps.StatefulSet) error
}

type tikvUpgrader struct {
	deps *controller.Dependencies

	volumeModifier volumes.PodVolumeModifier
	coordinator    coordinator.CoordinatorInterface
}

// NewTiKVUpgrader returns a tikv Upgrader
func NewTiKVUpgrader(deps *controller.Dependencies, pvm volumes.PodVolumeModifier, coordinator coordinator.CoordinatorInterface) TiKVUpgrader {
	return &tikvUpgrader{
		deps:           deps,
		volumeModifier: pvm,
		coordinator:    coordinator,
	}
}

func (u *tikvUpgrader) Upgrade(meta metav1.Object, oldSet *apps.StatefulSet, newSet *apps.StatefulSet) error {
	ns := meta.GetNamespace()
	tcName := meta.GetName()

	var status *v1alpha1.TiKVStatus
	switch meta := meta.(type) {
	case *v1alpha1.TidbCluster:
		if ready, reason := isTiKVReadyToUpgrade(meta); !ready {
			klog.Infof("TidbCluster: [%s/%s], can not upgrade tikv because: %s", ns, tcName, reason)
			_, podSpec, err := GetLastAppliedConfig(oldSet)
			if err != nil {
				return err
			}
			newSet.Spec.Template.Spec = *podSpec
			return nil
		}
		status = &meta.Status.TiKV
	default:
		return fmt.Errorf("cluster[%s/%s] failed to upgrading tikv due to converting", meta.GetNamespace(), meta.GetName())
	}

	tc, _ := meta.(*v1alpha1.TidbCluster)

	// upgrade tikv without evicting leader when only one tikv is exist
	// NOTE: If `TiKVStatus.Synced`` is false, it's acceptable to use old record about peer stores
	if *oldSet.Spec.Replicas < 2 && len(tc.Status.TiKV.PeerStores) == 0 {
		klog.Infof("TiKV statefulset replicas are less than 2, skip evicting region leader for tc %s/%s", ns, tcName)
		status.Phase = v1alpha1.UpgradePhase
		mngerutils.SetUpgradePartition(newSet, 0)
		return nil
	}

	if !status.Synced {
		return fmt.Errorf("cluster: [%s/%s]'s tikv status sync failed, can not to be upgraded", ns, tcName)
	}

	status.Phase = v1alpha1.UpgradePhase
	if !templateEqual(newSet, oldSet) {
		return nil
	}

	if status.StatefulSet.UpdateRevision == status.StatefulSet.CurrentRevision {
		return nil
	}

	if oldSet.Spec.UpdateStrategy.Type == apps.OnDeleteStatefulSetStrategyType || oldSet.Spec.UpdateStrategy.RollingUpdate == nil {
		// Manually bypass tidb-operator to modify statefulset directly, such as modify tikv statefulset's RollingUpdate strategy to OnDelete strategy,
		// or set RollingUpdate to nil, skip tidb-operator's rolling update logic in order to speed up the upgrade in the test environment occasionally.
		// If we encounter this situation, we will let the native statefulset controller do the upgrade completely, which may be unsafe for upgrading tikv.
		// Therefore, in the production environment, we should try to avoid modifying the tikv statefulset update strategy directly.
		newSet.Spec.UpdateStrategy = oldSet.Spec.UpdateStrategy
		klog.Warningf("tidbcluster: [%s/%s] tikv statefulset %s UpdateStrategy has been modified manually", ns, tcName, oldSet.GetName())
		return nil
	}

	if err := isReadyForUpgrade(tc, v1alpha1.TiKVMemberType, u.coordinator); err != nil {
		return err
	}

	minReadySeconds := 0
	s, ok := tc.Annotations[annoKeyTiKVMinReadySeconds]
	if ok {
		i, err := strconv.Atoi(s)
		if err != nil {
			klog.Warningf("tidbcluster: [%s/%s] annotation %s should be an integer: %v", ns, tcName, annoKeyTiKVMinReadySeconds, err)
		} else {
			minReadySeconds = i
		}
	}

	mngerutils.SetUpgradePartition(newSet, *oldSet.Spec.UpdateStrategy.RollingUpdate.Partition)
	podOrdinals := helper.GetPodOrdinals(*oldSet.Spec.Replicas, oldSet).List()
	var canaryReplicasList []int32
	if tc.IsTiKVCanaryUpgradeEnabled() {
		canaryReplicasList = getCanaryReplicasList(tc, "tikv", int32(len(podOrdinals)), tc.Spec.TiKV.CanaryUpgrade)
	}
	for _i := len(podOrdinals) - 1; _i >= 0; _i-- {
		i := podOrdinals[_i]
		store := getStoreByOrdinal(meta.GetName(), *status, i)
		if store == nil {
			mngerutils.SetUpgradePartition(newSet, i)
			continue
		}
		podName := TikvPodName(tcName, i)
		pod, err := u.deps.PodLister.Pods(ns).Get(podName)
		if err != nil {
			return fmt.Errorf("tikvUpgrader.Upgrade: failed to get pods %s for cluster %s/%s, error: %s", podName, ns, tcName, err)
		}
		revision, exist := pod.Labels[apps.ControllerRevisionHashLabelKey]
		if !exist {
			return controller.RequeueErrorf("tidbcluster: [%s/%s]'s tikv pod: [%s] has no label: %s", ns, tcName, podName, apps.ControllerRevisionHashLabelKey)
		}

		if revision == status.StatefulSet.UpdateRevision {

			if !podutil.IsPodAvailable(pod, int32(minReadySeconds), metav1.Now()) {
				readyCond := podutil.GetPodReadyCondition(pod.Status)
				if readyCond == nil || readyCond.Status != corev1.ConditionTrue {
					return controller.RequeueErrorf("tidbcluster: [%s/%s]'s upgraded tikv pod: [%s] is not ready", ns, tcName, podName)

				}
				return controller.RequeueErrorf("tidbcluster: [%s/%s]'s upgraded tikv pod: [%s] is not available, last transition time is %v", ns, tcName, podName, readyCond.LastTransitionTime)
			}
			if store.State != v1alpha1.TiKVStateUp {
				return controller.RequeueErrorf("tidbcluster: [%s/%s]'s upgraded tikv pod: [%s] is not all ready", ns, tcName, podName)
			}
			// If pods recreated successfully, endEvictLeader for the store on this Pod.
			storeID, err := strconv.ParseUint(store.ID, 10, 64)
			if err != nil {
				return err
			}

			if err := checkTiKVReady(u.deps, tc, storeID); err != nil {
				return err
			}

			if err := endEvictLeaderbyStoreID(u.deps, tc, storeID); err != nil {
				return err
			}
			if err := checkCanaryUpgrade(tc, "tikv", pod, tc.Spec.TiKV.CanaryUpgrade, int32(len(podOrdinals)), int32(_i), canaryReplicasList); err != nil {
				if controller.IsIgnoreError(err) {
					klog.Infof("%v", err)
					return nil
				}
				return err
			}

			continue
		}

		return u.upgradeTiKVPod(tc, i, newSet)
	}

	return nil
}

func (u *tikvUpgrader) upgradeTiKVPod(tc *v1alpha1.TidbCluster, ordinal int32, newSet *apps.StatefulSet) error {
	ns := tc.GetNamespace()
	tcName := tc.GetName()
	upgradePodName := TikvPodName(tcName, ordinal)
	upgradePod, err := u.deps.PodLister.Pods(ns).Get(upgradePodName)
	if err != nil {
		return fmt.Errorf("upgradeTiKVPod: failed to get pod %s for tc %s/%s, error: %s", upgradePodName, ns, tcName, err)
	}

	done, err := u.evictLeaderBeforeUpgrade(tc, upgradePod)
	if err != nil {
		return fmt.Errorf("upgradeTiKVPod: failed to evict leader of pod %s for tc %s/%s, error: %s", upgradePodName, ns, tcName, err)
	}
	if !done {
		return controller.RequeueErrorf("upgradeTiKVPod: evicting leader of pod %s for tc %s/%s", upgradePodName, ns, tcName)
	}

	leaderCountBeforeEvictLeaderStr := upgradePod.Annotations[LeaderCountBeforeEvictLeader]
	delay := true
	if len(leaderCountBeforeEvictLeaderStr) == 0 {
		// Skip waiting when the store is not up.
		delay = false
	} else {
		leaderCountBeforeEvictLeader, err := strconv.ParseUint(leaderCountBeforeEvictLeaderStr, 10, 32)
		if err != nil {
			return err
		}
		delay = leaderCountBeforeEvictLeader > 0
	}
	if delay {
		upgradeDelay := tc.TiKVUpgradeDelayAfterEvictLeader()
		if evictLeaderFinishTimeStr, finish := upgradePod.Annotations[EvictLeaderFinishTime]; finish {
			evictLeaderFinishTime, err := time.Parse(time.RFC3339, evictLeaderFinishTimeStr)
			if err != nil {
				klog.Errorf("upgradeTiKVPod: failed to parse annotation %q to time, error: %s", EvictLeaderFinishTime, err)
				return err
			}
			now := time.Now()
			upgradeTime := evictLeaderFinishTime.Add(upgradeDelay)
			if now.Before(upgradeTime) {
				dur := upgradeTime.Sub(now)
				return controller.RequeueErrorf("upgradeTiKVPod: wait %s to upgrade pod %s for tc %s/%s", dur, upgradePodName, ns, tcName)
			}
		}
	}
	if features.DefaultFeatureGate.Enabled(features.VolumeModifying) {
		done, err = u.modifyVolumesBeforeUpgrade(tc, upgradePod)
		if err != nil {
			return fmt.Errorf("upgradeTiKVPod: failed to modify volumes of pod %s for tc %s/%s, error: %s", upgradePodName, ns, tcName, err)
		}
		if !done {
			return controller.RequeueErrorf("upgradeTiKVPod: modifying volumes of pod %s for tc %s/%s", upgradePodName, ns, tcName)
		}
	}

	mngerutils.SetUpgradePartition(newSet, ordinal)
	return nil
}

func (u *tikvUpgrader) evictLeaderBeforeUpgrade(tc *v1alpha1.TidbCluster, upgradePod *corev1.Pod) (bool, error) {
	logPrefix := fmt.Sprintf("evict leader before upgrading tikv pod %s/%s", upgradePod.Namespace, upgradePod.Name)

	storeID, err := TiKVStoreIDFromStatus(tc, upgradePod.Name)
	if err != nil {
		return false, err
	}

	// evict leader if needed
	_, evicting := upgradePod.Annotations[EvictLeaderBeginTime]
	if !evicting {
		return false, u.beginEvictLeader(tc, storeID, upgradePod)
	}

	// wait for leader eviction to complete or timeout
	evictLeaderTimeout := tc.TiKVEvictLeaderTimeout()
	if evictLeaderBeginTimeStr, evicting := upgradePod.Annotations[EvictLeaderBeginTime]; evicting {
		evictLeaderBeginTime, err := time.Parse(time.RFC3339, evictLeaderBeginTimeStr)
		if err != nil {
			klog.Errorf("%s: parse annotation %q to time failed", logPrefix, EvictLeaderBeginTime)
			return false, nil
		}
		if time.Now().After(evictLeaderBeginTime.Add(evictLeaderTimeout)) {
			klog.Infof("%s: evict leader timeout with threshold %v, so ready to upgrade", logPrefix, evictLeaderTimeout)
			return true, nil
		}
	}

	leaderCount, err := u.deps.TiKVControl.GetTiKVPodClient(tc.Namespace, tc.Name, upgradePod.Name, tc.IsTLSClusterEnabled()).GetLeaderCount()
	if err != nil {
		klog.Warningf("%s: failed to get leader count, error: %v", logPrefix, err)
		return false, nil
	}

	if leaderCount == 0 {
		_, finish := upgradePod.Annotations[EvictLeaderFinishTime]
		if !finish {
			err = u.finishEvictLeader(tc, storeID, upgradePod)
			if err != nil {
				klog.Errorf("%s: failed to finish evict leader, error: %v", logPrefix, err)
				return false, err
			}
		}
		klog.Infof("%s: leader count is 0, so ready to upgrade", logPrefix)
		return true, nil
	}

	klog.Infof("%s: leader count is %d, and wait for evictition to complete", logPrefix, leaderCount)
	return false, nil
}

func (u *tikvUpgrader) modifyVolumesBeforeUpgrade(tc *v1alpha1.TidbCluster, upgradePod *corev1.Pod) (bool, error) {
	desiredVolumes, err := u.volumeModifier.GetDesiredVolumes(tc, v1alpha1.TiKVMemberType)
	if err != nil {
		return false, err
	}

	actual, err := u.volumeModifier.GetActualVolumes(upgradePod, desiredVolumes)
	if err != nil {
		return false, err
	}

	if u.volumeModifier.ShouldModify(actual) {
		err := u.volumeModifier.Modify(actual)
		return false, err
	}

	return true, nil
}

func (u *tikvUpgrader) beginEvictLeader(tc *v1alpha1.TidbCluster, storeID uint64, pod *corev1.Pod) error {
	ns := tc.GetNamespace()
	podName := pod.GetName()

	_, setLeaderCount := pod.Annotations[LeaderCountBeforeEvictLeader]
	if !setLeaderCount {
		if store, ok := tc.Status.TiKV.Stores[fmt.Sprintf("%d", storeID)]; ok && store.State == v1alpha1.TiKVStateUp {
			err := updateAnnotationForPod(u.deps, tc, pod, LeaderCountBeforeEvictLeader, fmt.Sprintf("%d", store.LeaderCount))
			if err != nil {
				return err
			}
		}
	}

	err := controller.GetPDClient(u.deps.PDControl, tc).BeginEvictLeader(storeID)
	if err != nil {
		klog.Errorf("tikv upgrader: failed to begin evict leader: %d, %s/%s, %v",
			storeID, ns, podName, err)
		return err
	}
	klog.Infof("tikv upgrader: begin evict leader: %d, %s/%s successfully", storeID, ns, podName)
	return updateAnnotationTime(u.deps, tc, pod, EvictLeaderBeginTime)
}

func (u *tikvUpgrader) finishEvictLeader(tc *v1alpha1.TidbCluster, storeID uint64, pod *corev1.Pod) error {
	klog.Infof("tikv upgrader: finish evict leader: %d %s/%s", storeID, tc.GetNamespace(), pod.GetName())
	return updateAnnotationTime(u.deps, tc, pod, EvictLeaderFinishTime)
}

func checkTiKVReady(deps *controller.Dependencies, tc *v1alpha1.TidbCluster, storeID uint64) error {
	ns := tc.GetNamespace()
	tcName := tc.GetName()

	store, ok := tc.Status.TiKV.Stores[fmt.Sprintf("%d", storeID)]
	if !ok {
		return nil
	}
	pod, err := deps.PodLister.Pods(tc.Namespace).Get(store.PodName)
	if err != nil {
		if errors.IsNotFound(err) {
			return nil
		}
		return fmt.Errorf("tikv: failed to get pod %s/%s, error: %s", tc.Namespace, store.PodName, err)
	}

	podName := pod.GetName()

	_, err = deps.TiKVControl.GetTiKVPodClient(ns, tcName, podName, tc.IsTLSClusterEnabled()).GetLeaderCount()
	if err != nil {
		klog.Warningf("tikv upgrader: failed to get leader count, error: %v", err)
		return err
	}
	klog.Infof("tikv upgrader: tikv is ready: %d %s/%s", storeID, ns, podName)
	readyTimeStr, set := pod.Annotations[MemberReadyTime]
	if !set {
		if err := updateAnnotationTime(deps, tc, pod, MemberReadyTime); err != nil {
			return err
		}
		readyTimeStr = pod.Annotations[MemberReadyTime]
	}
	endEvictLeaderDelay := tc.TiKVEndEvictLeaderDelayAfterReady()
	readyTime, err := time.Parse(time.RFC3339, readyTimeStr)
	if err != nil {
		klog.Errorf("tikv upgrader: failed to parse annotation %q to time, error: %s", MemberReadyTime, err)
		return err
	}
	now := time.Now()
	endEvictLeaderTime := readyTime.Add(endEvictLeaderDelay)
	if now.Before(endEvictLeaderTime) {
		dur := endEvictLeaderTime.Sub(now)
		return controller.RequeueErrorf("tikv upgrader: wait %s to end evict leader of pod %s for tc %s/%s", dur, podName, ns, tcName)
	}
	return nil
}

func updateAnnotationTime(deps *controller.Dependencies, tc *v1alpha1.TidbCluster, pod *corev1.Pod, key string) error {
	return updateAnnotationForPod(deps, tc, pod, key, time.Now().Format(time.RFC3339))
}

func updateAnnotationForPod(deps *controller.Dependencies, tc *v1alpha1.TidbCluster, pod *corev1.Pod, key, value string) error {
	ns := tc.GetNamespace()
	podName := pod.GetName()

	if pod.Annotations == nil {
		pod.Annotations = map[string]string{}
	}
	pod.Annotations[key] = value
	_, err := deps.PodControl.UpdatePod(tc, pod)
	if err != nil {
		klog.Errorf("set annotation: failed to set pod %s/%s annotation %s to %s, %v",
			ns, podName, key, value, err)
		return err
	}
	klog.Infof("set annotation: set pod %s/%s annotation %s to %s successfully",
		ns, podName, key, value)
	return nil
}

// endEvictLeaderForAllStore end evict leader for all stores of a tc
func endEvictLeaderForAllStore(deps *controller.Dependencies, tc *v1alpha1.TidbCluster) error {
	storeIDs := make([]uint64, 0, len(tc.Status.TiKV.Stores)+len(tc.Status.TiKV.TombstoneStores))
	for _, stores := range []map[string]v1alpha1.TiKVStore{tc.Status.TiKV.Stores, tc.Status.TiKV.TombstoneStores} {
		for _, store := range stores {
			storeID, err := strconv.ParseUint(store.ID, 10, 64)
			if err != nil {
				return fmt.Errorf("parse store id %s to uint64 failed: %v", store.ID, err)
			}
			storeIDs = append(storeIDs, storeID)
		}
	}

	pdcli := controller.GetPDClient(deps.PDControl, tc)

	scheduelrs, err := pdcli.GetEvictLeaderSchedulersForStores(storeIDs...)
	if err != nil {
		return fmt.Errorf("get scheduler failed: %v", err)
	}
	if len(scheduelrs) == 0 {
		klog.Infof("tikv: no evict leader scheduler exists for %s/%s", tc.Namespace, tc.Name)
		return nil
	}

	errs := make([]error, 0)
	for storeID := range scheduelrs {
		if err := deleteEvictLeaderAnnotation(deps, tc, storeID); err != nil {
			klog.Error(err)
			errs = append(errs, err)
			continue
		}
		err := pdcli.EndEvictLeader(storeID)
		if err != nil {
			klog.Errorf("tikv: failed to end evict leader for store: %d of %s/%s, error: %v", storeID, tc.Namespace, tc.Name, err)
			errs = append(errs, fmt.Errorf("end evict leader for store %d failed: %v", storeID, err))
			continue
		}
		klog.Infof("tikv: end evict leader for store: %d of %s/%s successfully", storeID, tc.Namespace, tc.Name)
	}

	if len(errs) > 0 {
		return errorutils.NewAggregate(errs)
	}

	return nil
}

func endEvictLeader(deps *controller.Dependencies, tc *v1alpha1.TidbCluster, ordinal int32) error {
	store := getStoreByOrdinal(tc.GetName(), tc.Status.TiKV, ordinal)
	if store == nil {
		klog.Errorf("tikv: no store found for TiKV ordinal %v of %s/%s", ordinal, tc.Namespace, tc.Name)
		return nil
	}
	storeID, err := strconv.ParseUint(store.ID, 10, 64)
	if err != nil {
		return err
	}

	if err := checkTiKVReady(deps, tc, storeID); err != nil {
		return err
	}

	return endEvictLeaderbyStoreID(deps, tc, storeID)
}

func endEvictLeaderbyStoreID(deps *controller.Dependencies, tc *v1alpha1.TidbCluster, storeID uint64) error {
	// wait 5 second before delete evict scheduler，it is for auto test can catch these info
	if deps.CLIConfig.TestMode {
		time.Sleep(5 * time.Second)
	}

	if err := deleteEvictLeaderAnnotation(deps, tc, storeID); err != nil {
		klog.Error(err)
		return err
	}
	err := controller.GetPDClient(deps.PDControl, tc).EndEvictLeader(storeID)
	if err != nil {
		klog.Errorf("tikv: failed to end evict leader for store: %d of %s/%s, error: %v", storeID, tc.Namespace, tc.Name, err)
		return err
	}
	klog.Infof("tikv: end evict leader for store: %d of %s/%s successfully", storeID, tc.Namespace, tc.Name)
	return nil
}

func getStoreByOrdinal(name string, status v1alpha1.TiKVStatus, ordinal int32) *v1alpha1.TiKVStore {
	podName := TikvPodName(name, ordinal)
	for _, store := range status.Stores {
		if store.PodName == podName {
			return &store
		}
	}
	return nil
}

func deleteEvictLeaderAnnotation(deps *controller.Dependencies, tc *v1alpha1.TidbCluster, storeID uint64) error {
	store, ok := tc.Status.TiKV.Stores[fmt.Sprintf("%d", storeID)]
	if !ok {
		return nil
	}
	pod, err := deps.PodLister.Pods(tc.Namespace).Get(store.PodName)
	if err != nil {
		if errors.IsNotFound(err) {
			return nil
		}
		return fmt.Errorf("tikv: failed to get pod %s/%s, error: %s", tc.Namespace, store.PodName, err)
	}
	if len(pod.Annotations) == 0 {
		return nil
	}
	// If the tikv upgrade is cancelled, delete evict leader annotation.
	if _, ok := pod.Annotations[EvictLeaderBeginTime]; !ok {
		return nil
	}
	delete(pod.Annotations, EvictLeaderBeginTime)
	delete(pod.Annotations, LeaderCountBeforeEvictLeader)
	_, err = deps.PodControl.UpdatePod(tc, pod)
	if err != nil {
		return fmt.Errorf("tikv: failed to delete annotation %s and %s from pod %s/%s, err %s", EvictLeaderBeginTime, LeaderCountBeforeEvictLeader, tc.Namespace, store.PodName, err.Error())
	}
	klog.Infof("tikv: delete annotation %s and %s from pod %s/%s", EvictLeaderBeginTime, LeaderCountBeforeEvictLeader, tc.Namespace, store.PodName)
	return nil
}

func isTiKVReadyToUpgrade(tc *v1alpha1.TidbCluster) (bool, string) {
	if tc.Status.TiFlash.Phase == v1alpha1.UpgradePhase || tc.Status.TiFlash.Phase == v1alpha1.ScalePhase {
		return false, fmt.Sprintf("tiflash status is %s", tc.Status.TiFlash.Phase)
	}
	if tc.Status.PD.Phase == v1alpha1.UpgradePhase || tc.Status.PD.Phase == v1alpha1.ScalePhase {
		return false, fmt.Sprintf("pd status is %s", tc.Status.PD.Phase)
	}
	if tc.TiKVScaling() {
		return false, fmt.Sprintf("tikv status is %s", tc.Status.TiKV.Phase)
	}

	return true, ""
}

type fakeTiKVUpgrader struct{}

// NewFakeTiKVUpgrader returns a fake tikv upgrader
func NewFakeTiKVUpgrader() TiKVUpgrader {
	return &fakeTiKVUpgrader{}
}

func (u *fakeTiKVUpgrader) Upgrade(meta metav1.Object, _ *apps.StatefulSet, _ *apps.StatefulSet) error {
	tc := meta.(*v1alpha1.TidbCluster)
	tc.Status.TiKV.Phase = v1alpha1.UpgradePhase
	return nil
}