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

	"github.com/juju/errors"
	"github.com/ngaut/log"
	cmapi "github.com/pingcap/tidb-cloud-manager/pkg/api"
	"github.com/pingcap/tidb-cloud-manager/pkg/client"
	tiapi "github.com/pingcap/tidb-cloud-manager/pkg/tidbinstance/api"
	tcapi "github.com/pingcap/private-tidb-operator/pkg/tidbcluster/api"
	tsapi "github.com/pingcap/private-tidb-operator/pkg/tidbset/api"
	"github.com/pingcap/private-tidb-operator/pkg/util/label"
	"k8s.io/api/apps/v1beta1"
	batchv1 "k8s.io/api/batch/v1"
	"k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/intstr"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
)

var (
	keyFunc = cache.DeletionHandlingMetaNamespaceKeyFunc
	// AlertmanagerURL is alertmanager URL for prometheus
	AlertmanagerURL string
	// controllerKind contains the schema.GroupVersionKind for this controller type
	controllerKind = cmapi.SchemeGroupVersion.WithKind("TidbInstance")

	restartedDuration = 30 * time.Second

	// Whether to expose privileged TiDB SVC
	ExposePrivilegedTiDBSVC bool

	// RBACEnabled indicates whether to enable RBAC, if k8s cluster enables RBAC, this must be true
	RBACEnabled bool

	propBackGround = metav1.DeletePropagationBackground
	deleteOptions  = &metav1.DeleteOptions{PropagationPolicy: &propBackGround}
)

const (
	saName            = "tidb-prometheus" // serviceAccount name for monitor
	clusterName       = "tidb-cluster"
	promRetentionDays = 30 // prometheus default retention days
)

// TiDBInstanceController is the controller for TiDBInstance
type TiDBInstanceController struct {
	ns         string
	cli        client.Interface
	controller cache.Controller
	store      cache.Store
	queue      *workqueue.Type
}

// NewTiDBInstanceController creates a new TiDBInstance Controller
func NewTiDBInstanceController(ns string, cli client.Interface) *TiDBInstanceController {
	tic := &TiDBInstanceController{
		ns:    ns,
		cli:   cli,
		queue: workqueue.New(),
	}
	tic.store, tic.controller = cache.NewInformer(
		&cache.ListWatch{
			ListFunc: cache.ListFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
				return tic.cli.PingcapV1().TidbInstances(metav1.NamespaceAll).List(opts)
			}),
			WatchFunc: cache.WatchFunc(func(opts metav1.ListOptions) (watch.Interface, error) {
				return tic.cli.PingcapV1().TidbInstances(metav1.NamespaceAll).Watch(opts)
			}),
		},
		&tiapi.TidbInstance{},
		30*time.Second,
		cache.ResourceEventHandlerFuncs{
			AddFunc:    tic.enqueueTiDBInstance,
			UpdateFunc: tic.updateTiDBInstance,
			DeleteFunc: tic.enqueueTiDBInstance,
		},
	)
	return tic
}

// Run method runs a TiDBInstance Controller
func (tic *TiDBInstanceController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer tic.queue.ShutDown()
	log.Infof("Starting TidbInstance controller")
	go tic.controller.Run(wait.NeverStop)
	go wait.Until(tic.worker, time.Second, stopCh)
	<-stopCh
	log.Infof("Shutting down TidbInstance controller")
}

func (tic *TiDBInstanceController) enqueueTiDBInstance(obj interface{}) {
	key, err := keyFunc(obj)
	if err != nil {
		log.Errorf("can't get key for obj: %v, error: %v", obj, err)
		return
	}

	tic.queue.Add(key)
}

func (tic *TiDBInstanceController) worker() {
	for {
		func() {
			key, quit := tic.queue.Get()
			if quit {
				return
			}
			defer tic.queue.Done(key)
			if err := tic.syncTiDBInstance(key.(string)); err != nil {
				log.Error(err)
			}
		}()
	}
}

func (tic *TiDBInstanceController) syncTiDBInstance(key string) error {
	startTime := time.Now()
	defer func() {
		log.Infof("Finished syncing tidb instance %q (%v)", key, time.Now().Sub(startTime))
	}()

	obj, exists, err := tic.store.GetByKey(key)
	if err != nil {
		errMsg := fmt.Sprintf("obj: %v is not a *TidbInstance", obj)
		log.Errorf(errMsg)
		return errors.New(errMsg)
	}
	if !exists {
		log.Infof("TidbInstance has been deleted %v", key)
		ns, name, err := cache.SplitMetaNamespaceKey(key)
		if err != nil {
			log.Error(err)
			return err
		}
		_, err = tic.cli.PingcapV1().TidbInstances(ns).Get(name, metav1.GetOptions{})
		if err != nil {
			if apierrors.IsNotFound(err) { // make sure tidbinstance is actually deleted
				return tic.deleteTiDBInstanceResource(ns)
			}
			return err
		}
		return nil
	}

	ti, ok := obj.(*tiapi.TidbInstance)
	if !ok {
		log.Errorf("obj: %v is not a *TidbInstance", obj)
		return fmt.Errorf("object %+v is not a *TidbInstance", obj)
	}

	if ti.Spec.Deleted {
		log.Info("TidbInstance.Spec.Deleted is equal to true")

		// set TidbCluster.Spec.Deleted equal to true
		tic.tryToDeleteTidbCluster(ti)
		// delete privileged TiDB deployment and service
		tic.tryToDeletePrivilegedTiDB(ti)

		// delete all monitor services: prometheus, grafana and deployment
		tic.tryToDeleteMonitor(ti)

		// delete all jobs
		tic.tryToDeleteJobs(ti)

		return nil
	}

	if RBACEnabled {
		if err := tic.ensureRBAC(ti.GetNamespace(), saName); err != nil {
			log.Errorf("failed to sync RBAC settings: %v", err)
		}
	}

	ns := ti.GetNamespace()
	if !ti.Status.Initialized {
		jobName := fmt.Sprintf("%s-set-tidb-password", clusterName)
		job, err := tic.cli.BatchV1().Jobs(ns).Get(jobName, metav1.GetOptions{})
		if err != nil {
			log.Errorf("failed to get job %s/%s: %v", ns, jobName, err)
		} else if job.Status.Succeeded != 1 {
			log.Infof("tidb cluster initialization hasn't finished: %v", job.Status)
		} else {
			ti.Status.Initialized = true
			_, err = tic.cli.PingcapV1().TidbInstances(ns).Update(ti)
			if err != nil {
				errMsg := fmt.Errorf("failed to update TidbInstance %s/%s initialization status: %v", ns, ti.GetName())
				log.Error(errMsg)
				return errMsg
			}
		}
	}

	if err := tic.syncTiDBCluster(ti); err != nil {
		log.Errorf("failed to sync TidbCluster for %s: %v", ti.Name, err)
	}

	if err := tic.syncMonitor(ti); err != nil {
		log.Errorf("failed to sync monitor for %s: %v", ti.Name, err)
	}

	return nil
}

func (tic *TiDBInstanceController) tryToDeleteJobs(ti *tiapi.TidbInstance) {
	ns := ti.GetName()
	jobList, err := tic.cli.BatchV1().Jobs(ns).List(label.ClusterListOptions(clusterName))
	if err != nil {
		log.Errorf("failed to list jobs in cluster [%s/%s], %v", ns, clusterName, err)
		return
	}

	// TODO use wait.Poll to ensure jobs are deleted?
	for _, job := range jobList.Items {
		jobName := job.GetName()
		if err := tic.cli.BatchV1().Jobs(ns).Delete(jobName, nil); err != nil {
			log.Errorf("delete Job: [%s/%s] failed, %v", ns, jobName, err)
		} else {
			log.Infof("delete Job: [%s/%s] successfully", ns, jobName)
		}
	}
}

func (tic *TiDBInstanceController) tryToDeleteMonitor(ti *tiapi.TidbInstance) {
	ns := ti.Name
	svcNames := []string{
		"prometheus",
		"grafana",
	}
	deployName := getName("monitor")

	for _, svcName := range svcNames {
		name := getName(svcName)
		if err := tic.cli.CoreV1().Services(ns).Delete(name, nil); err != nil {
			log.Errorf("delete Service: [%s/%s] failed, %v", ns, name, err)
		} else {
			log.Infof("delete Service: [%s/%s] successfully", ns, name)
		}
	}

	if RBACEnabled {
		if err := tic.cli.RbacV1beta1().Roles(ns).Delete(saName, nil); err != nil {
			log.Errorf("delete Role %s/%s failed: %v", ns, saName, err)
		} else {
			log.Infof("delete Role %s/%s successfully", ns, saName)
		}
		if err := tic.cli.RbacV1beta1().RoleBindings(ns).Delete(saName, nil); err != nil {
			log.Errorf("delete RoleBinding %s/%s failed: %v", ns, saName, err)
		} else {
			log.Infof("delete RoleBinding %s/%s successfully", ns, saName)
		}
		if err := tic.cli.CoreV1().ServiceAccounts(ns).Delete(saName, nil); err != nil {
			log.Errorf("delete ServiceAccount %s/%s failed: %v", ns, saName, err)
		} else {
			log.Infof("delete ServiceAccount %s/%s successfully", ns, saName)
		}
	}

	if err := tic.cli.AppsV1beta1().Deployments(ns).Delete(deployName, deleteOptions); err != nil {
		log.Errorf("delete Deployment: [%s/%s] failed, %v", ns, deployName, err)
	} else {
		log.Infof("delete Deployment: [%s/%s] successfully", ns, deployName)
	}
}

func (tic *TiDBInstanceController) tryToDeletePrivilegedTiDB(ti *tiapi.TidbInstance) {
	ns := ti.Name
	name := getName("privileged-tidb")
	if err := tic.cli.AppsV1beta1().Deployments(ns).Delete(name, deleteOptions); err != nil {
		log.Errorf("delete Deployment: %s/%s failed: %v", ns, name, err)
		return
	}

	log.Infof("delete Deployment: %s/%s successfully", ns, name)

	if err := tic.cli.CoreV1().Services(ns).Delete(name, nil); err != nil {
		log.Errorf("delete Service: %s/%s failed: %v", ns, name, err)
		return
	}

	log.Infof("delete Service: %s/%s successfully", ns, name)
}

func (tic *TiDBInstanceController) tryToDeleteTidbCluster(ti *tiapi.TidbInstance) {
	ns := ti.Name
	tc, err := tic.cli.PingcapV1().TidbClusters(ns).Get(clusterName, metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			log.Warnf("TidbInstance: [%s/%s] is graceful deleted, and its TidbCluster: [%s/%s] doesn't exist, deleting", ns, ns, ns, clusterName)
			err := tic.deleteTiDBInstanceResource(ns)
			if err != nil {
				log.Errorf("delete TidbInstance: [%s]'s resources failed, %v", ns, err)
			}
		} else {
			log.Errorf("get TidbCluster: [%s/%s] failed, %v", ns, clusterName, err)
		}

		return
	}

	if tc.ReallyGracefulDeleted() {
		log.Infof("TidbCluster: [%s/%s] is already graceful deleted, skipping", ns, clusterName)
		return
	}

	tc.Spec.State = tcapi.StateGracefulDeleted
	_, err = tic.cli.PingcapV1().TidbClusters(ns).Update(tc)
	if err != nil {
		log.Errorf("failed to update TidbCluster: %s/%s", ns, clusterName)
		return
	}

	log.Infof("TidbCluster: %s/%s's Spec.Deleted is set to true successfully", ti.Name, clusterName)
}

// TODO: implement this method
func (tic *TiDBInstanceController) syncTiDBInstanceStatus() error {
	return nil
}

func (tic *TiDBInstanceController) updateTiDBInstance(old, cur interface{}) {
	tic.enqueueTiDBInstance(cur)
}

func (tic *TiDBInstanceController) deleteTiDBInstanceResource(ns string) error {
	if err := tic.cli.CoreV1().Namespaces().Delete(ns, nil); err != nil {
		log.Errorf("failed to delete namespace [%s], %v", ns, err)
		return errors.Trace(err)
	}

	log.Infof("delete namespace: [%s] successfully", ns)
	return nil
}

// upgrade order is pd -> tikv -> tidb
func (tic *TiDBInstanceController) syncTiDBCluster(instance *tiapi.TidbInstance) error {
	ns := instance.Name

	tc, err := tic.cli.PingcapV1().TidbClusters(ns).Get(clusterName, metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			return tic.createTiDBCluster(instance)
		}

		log.Errorf("failed to get TidbCluster %s/%s: %v", ns, clusterName, err)
		return errors.Trace(err)
	}

	// start to update config and pushgateway-image annotation
	config, err := tic.genRenderedConfig(instance)
	if err != nil {
		log.Errorf("failed to render TidbInstance: [%s/%s] config: %v", instance.Namespace, instance.Name, err)
		return errors.Trace(err)
	}
	tc.Spec.Config = config
	tc.Annotations = updatePushGatewayAnn(tc.Annotations, instance.Spec.TiKV)
	_, err = tic.cli.PingcapV1().TidbClusters(instance.Name).Update(tc)
	if err != nil {
		log.Errorf("failed to upgrade TidbCluster: %s/%s's config, %v", ns, clusterName, err)
		return errors.Trace(err)
	}

	// tidbinstance is not graceful deleted and tidbcluster is graceful deleted
	if tc.ReallyGracefulDeleted() {
		log.Warnf("TidbCluster: [%s/%s] is graceful deleted, restoring", ns, clusterName)
		tc.Spec.State = tcapi.StateRestore
		tc, err = tic.cli.PingcapV1().TidbClusters(ns).Update(tc)
		if err != nil {
			log.Errorf("failed to update TidbCluster: [%s/%s], %v", ns, clusterName, err)
		}
		log.Infof("update TidbCluster: [%s/%s] successfully", ns, clusterName)
	}

	if err := tic.ensurePrivilegedTiDBExist(instance); err != nil {
		log.Errorf("failed to ensure privileged TiDB existence: %v", err)
		return errors.Trace(err)
	}

	if ShouldResize(instance, tc) {
		tc.Spec.PD.Size = instance.Spec.PD.Size
		tc.Spec.TiDB.Size = instance.Spec.TiDB.Size
		tc.Spec.TiKV.Size = instance.Spec.TiKV.Size

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

		log.Infof("TidbCluster %s/%s resized", ns, clusterName)

		return nil
	}

	if instance.PDUpgrading() {
		log.Infof("upgrading TidbInstance: %s's PD", objectName(instance.ObjectMeta))
		err = tic.tryToUpgradePD(instance, tc)
		if err != nil {
			log.Errorf("failed to upgrade TidbCluster: %s/%s's PD, %v", ns, clusterName, err)
			return errors.Trace(err)
		}
	} else if instance.TiKVUpgrading() {
		log.Infof("upgrading TidbInstance: %s's TiKV", objectName(instance.ObjectMeta))
		err = tic.tryToUpgradeTiKV(instance, tc)
		if err != nil {
			log.Errorf("failed to upgrade TidbCluster: %s/%s's TiKV, %v", ns, clusterName, err)
			return errors.Trace(err)
		}
	} else if instance.TiDBUpgrading() {
		log.Infof("upgrading TidbInstance: %s's TiDB", objectName(instance.ObjectMeta))
		err = tic.tryToUpgradeTiDB(instance, tc)
		if err != nil {
			log.Errorf("failed to upgrade TidbCluster: %s/%s's TiDB, %v", ns, clusterName, err)
			return errors.Trace(err)
		}
	}

	return nil
}

func objectName(om metav1.ObjectMeta) string {
	return fmt.Sprintf("%s/%s", om.Namespace, om.Name)
}

func (tic *TiDBInstanceController) createTiDBCluster(instance *tiapi.TidbInstance) error {
	ns := instance.Name

	config, err := tic.genRenderedConfig(instance)
	if err != nil {
		log.Errorf("failed to render TidbInstance: [%s/%s] config: %v", instance.Namespace, instance.Name, err)
		return errors.Trace(err)
	}

	clusterSpec := tcapi.ClusterSpec{
		PD: tcapi.PDSpec{
			assembleMemberSpec(PDContainerType, instance.Spec.PD),
		},
		TiDB: tcapi.TiDBSpec{
			MemberSpec: assembleMemberSpec(TiDBContainerType, instance.Spec.TiDB),
		},
		TiKV: tcapi.TiKVSpec{
			assembleMemberSpec(TiKVContainerType, instance.Spec.TiKV),
		},
		Paused:  false,
		Config:  config,
		Service: instance.Spec.ServiceType,
	}

	if HasBinlogEnabled(instance) {
		binlogMemberSpec := assembleMemberSpec(BinlogContainerType, instance.Spec.TiDB)
		clusterSpec.TiDB.Binlog = &binlogMemberSpec
	}

	tc := &tcapi.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:   clusterName,
			Labels: instance.Labels,
			Annotations: map[string]string{
				"pushgateway-image": GetImageByTiPodSpec(PushgatewayContainerType, instance.Spec.TiKV),
			},
		},
		Spec: clusterSpec,
	}

	if _, err := tic.cli.PingcapV1().TidbClusters(ns).Create(tc); err != nil {
		log.Errorf("failed to create TidbCluster %s/%s: %v", ns, clusterName, err)
		return errors.Trace(err)
	}

	log.Infof("TidbCluster %s/%s created", ns, clusterName)

	return nil
}

func (tic *TiDBInstanceController) tryToUpgradePD(instance *tiapi.TidbInstance, tc *tcapi.TidbCluster) error {
	ns := instance.Name
	pdImage := GetImageByTiPodSpec(PDContainerType, instance.Spec.PD)

	if tc.Spec.PD.Image != pdImage {
		log.Infof("setting TidbCluster: %s's PD Image to: %s", objectName(tc.ObjectMeta), pdImage)
		tc.Spec.PD.Image = pdImage
		_, err := tic.cli.PingcapV1().TidbClusters(instance.Name).Update(tc)
		if err != nil {
			log.Errorf("failed to upgrade TidbCluster: %s/%s's PD, %v", ns, clusterName, err)
			return errors.Trace(err)
		}
	}

	pdPodList, err := tic.cli.CoreV1().Pods(ns).List(
		metav1.ListOptions{
			LabelSelector: labels.SelectorFromSet(
				label.New().Cluster(clusterName).PD().Labels(),
			).String(),
		},
	)
	if err != nil {
		log.Errorf("Error when list PD pods: %v", err)
		return errors.Trace(err)
	}

	tsName := fmt.Sprintf("%s-pdset", clusterName)
	ts, err := tic.cli.PingcapV1().TidbSets(ns).Get(tsName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("Error when get TidbSet: %s, %v", tsName, err)
		return errors.Trace(err)
	}

	if tic.allUpgraded(pdPodList.Items, instance, ts, PDContainerType, pdImage) {
		log.Infof("setting TidbInstance: %s's .Status.PDStatus.Upgrading = false", objectName(instance.ObjectMeta))
		instance.Status.PDStatus.UpgradeState = tiapi.Upgraded
		instance.Status.PDStatus.UpgradeEndTime = metav1.Now()
		_, err := tic.cli.PingcapV1().TidbInstances(ns).Update(instance)
		if err != nil {
			log.Errorf("Error when update TidbInstance: %s", objectName(instance.ObjectMeta))
			return errors.Trace(err)
		}
	}

	return nil
}

// PodType represents pod type
type PodType string

const (
	// PDPodType is pd pod type
	PDPodType PodType = "pd"

	// TiDBPodType is tidb pod type
	TiDBPodType PodType = "tidb"

	// TiKVPodType is tikv pod type
	TiKVPodType PodType = "tikv"

	// MonitorPodType is monitor pod type
	MonitorPodType PodType = "monitor"

	// UnknownPodType is unknown pod type
	UnknownPodType PodType = "unknown"
)

// ContainerType represents container type
type ContainerType string

const (
	// PDContainerType is pd container type
	PDContainerType ContainerType = "pd"

	// TiDBContainerType is tidb container type
	TiDBContainerType ContainerType = "tidb"

	// BinlogContainerType is tidb binlog container type
	BinlogContainerType ContainerType = "tidb-binlog"

	// TiKVContainerType is tikv container type
	TiKVContainerType ContainerType = "tikv"

	// PushgatewayContainerType is pushgateway container type
	PushgatewayContainerType ContainerType = "pushgateway"

	// GrafanaContainerType is grafana container type
	GrafanaContainerType ContainerType = "grafana"

	// PrometheusContainerType is prometheus container type
	PrometheusContainerType ContainerType = "prometheus"

	// InstallerContainerType is tidb dashboard installer type
	InstallerContainerType ContainerType = "tidb-dashboard-installer"

	// TiDBToolsContainerType is tidb-tools container type
	TiDBToolsContainerType ContainerType = "tidb-tools"

	// TiDBEnterpriseToolsContainerType is tidb-enterprise-tools container type
	TiDBEnterpriseToolsContainerType ContainerType = "tidb-enterprise-tools"
)

func (tic *TiDBInstanceController) allUpgraded(pods []v1.Pod, ti *tiapi.TidbInstance, ts *tsapi.TidbSet, ctype ContainerType, image string) bool {
	for _, pod := range pods {
		var c *v1.Container
		for _, container := range pod.Spec.Containers {
			if string(ctype) == container.Name {
				c = &container
				break
			}
		}
		if c == nil {
			return false
		}
		if !(c.Image == image && ts.IsReady(&pod)) {
			return false
		}

		var cs *v1.ContainerStatus
		cses := pod.Status.ContainerStatuses
		if len(cses) == 0 {
			return false
		}

		for _, css := range cses {
			if string(ctype) == css.Name {
				cs = &css
			}
		}
		if cs == nil {
			return false
		}

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

		switch ctype {
		case PDContainerType:
			if running.StartedAt.Before(ti.Status.PDStatus.UpgradeBeginTime) {
				return false
			}
		case TiDBContainerType, BinlogContainerType:
			if running.StartedAt.Before(ti.Status.TiDBStatus.UpgradeBeginTime) {
				return false
			}
		case TiKVContainerType:
			if running.StartedAt.Before(ti.Status.TiKVStatus.UpgradeBeginTime) {
				return false
			}
		}

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

	return true
}

func (tic *TiDBInstanceController) tryToUpgradeTiKV(instance *tiapi.TidbInstance, tc *tcapi.TidbCluster) error {
	ns := instance.Name
	tikvImage := GetImageByTiPodSpec(TiKVContainerType, instance.Spec.TiKV)

	if tc.Spec.TiKV.Image != tikvImage {
		log.Infof("setting TidbCluster: %s's TiKV Image to: %s", objectName(tc.ObjectMeta), tikvImage)
		tc.Spec.TiKV.Image = tikvImage
		_, err := tic.cli.PingcapV1().TidbClusters(instance.Name).Update(tc)
		if err != nil {
			log.Errorf("failed to upgrade TidbCluster: %s/%s's TiKV, %v", ns, clusterName, err)
			return errors.Trace(err)
		}
	}

	tikvPodList, err := tic.cli.CoreV1().Pods(ns).List(
		metav1.ListOptions{
			LabelSelector: labels.SelectorFromSet(
				label.New().Cluster(clusterName).TiKV().Labels(),
			).String(),
		},
	)
	if err != nil {
		log.Errorf("Error when list TiKV pods: %v", err)
		return errors.Trace(err)
	}

	tsName := fmt.Sprintf("%s-tikvset", clusterName)
	ts, err := tic.cli.PingcapV1().TidbSets(ns).Get(tsName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("Error when get TidbSet: %s, %v", tsName, err)
		return errors.Trace(err)
	}

	if tic.allUpgraded(tikvPodList.Items, instance, ts, TiKVContainerType, tikvImage) {
		log.Infof("setting TidbInstance: %s's .Status.TiKVStatus.Upgrading = false", objectName(instance.ObjectMeta))
		instance.Status.TiKVStatus.UpgradeState = tiapi.Upgraded
		instance.Status.TiKVStatus.UpgradeEndTime = metav1.Now()
		_, err := tic.cli.PingcapV1().TidbInstances(ns).Update(instance)
		if err != nil {
			log.Errorf("Error when update TidbInstance: %s", objectName(instance.ObjectMeta))
			return errors.Trace(err)
		}
	}

	return nil
}

func (tic *TiDBInstanceController) tryToUpgradeTiDB(instance *tiapi.TidbInstance, tc *tcapi.TidbCluster) error {
	ns := instance.Name
	tidbImage := GetImageByTiPodSpec(TiDBContainerType, instance.Spec.TiDB)

	if tc.Spec.TiDB.Image != tidbImage {
		log.Infof("setting TidbCluster: %s's TiDB Image to: %s", objectName(tc.ObjectMeta), tidbImage)
		tc.Spec.TiDB.Image = tidbImage
	}

	binlogImage := GetImageByTiPodSpec(BinlogContainerType, instance.Spec.TiDB)
	if binlogImage != "" && (tc.Spec.TiDB.Binlog == nil || tc.Spec.TiDB.Binlog.Image != binlogImage) {
		log.Infof("setting TidbCluster: %s's TiDB binlog Image to: %s", objectName(tc.ObjectMeta), binlogImage)
		if tc.Spec.TiDB.Binlog == nil {
			log.Debugf("enable TiDB binlog: [%s/%s], image: [%s]", instance.Namespace, instance.Name, binlogImage)
			binlogMemberSpec := assembleMemberSpec(BinlogContainerType, instance.Spec.TiDB)
			tc.Spec.TiDB.Binlog = &binlogMemberSpec
		} else {
			log.Debugf("update TiDB binlog: [%s/%s], image: [%s]", instance.Namespace, instance.Name, binlogImage)
			tc.Spec.TiDB.Binlog.Image = binlogImage
		}
	}

	_, err := tic.cli.PingcapV1().TidbClusters(instance.Name).Update(tc)
	if err != nil {
		log.Errorf("failed to upgrade TidbCluster: %s/%s's TiDB, %v", ns, clusterName, err)
		return errors.Trace(err)
	}

	tidbPodList, err := tic.cli.CoreV1().Pods(ns).List(
		metav1.ListOptions{
			LabelSelector: labels.SelectorFromSet(
				label.New().Cluster(clusterName).TiDB().Labels(),
			).String(),
		},
	)
	if err != nil {
		log.Errorf("Error when list TiDB pods: %v", err)
		return errors.Trace(err)
	}

	tsName := fmt.Sprintf("%s-tidbset", clusterName)
	ts, err := tic.cli.PingcapV1().TidbSets(ns).Get(tsName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("Error when get TidbSet: %s, %v", tsName, err)
		return errors.Trace(err)
	}
	binlogAllUpdate := false
	if binlogImage == "" || tic.allUpgraded(tidbPodList.Items, instance, ts, BinlogContainerType, binlogImage) {
		// always set binlogAllUpdate true when binlogImage is empty
		binlogAllUpdate = true
	}

	if tic.allUpgraded(tidbPodList.Items, instance, ts, TiDBContainerType, tidbImage) && binlogAllUpdate {
		log.Infof("setting TidbInstance: %s's .Status.TiDBStatus.Upgrading = false", objectName(instance.ObjectMeta))
		instance.Status.TiDBStatus.UpgradeState = tiapi.Upgraded
		instance.Status.TiDBStatus.UpgradeEndTime = metav1.Now()
		_, err := tic.cli.PingcapV1().TidbInstances(ns).Update(instance)
		if err != nil {
			log.Errorf("Error when update TidbInstance: %s", objectName(instance.ObjectMeta))
			return errors.Trace(err)
		}
	}

	return nil
}

func (tic *TiDBInstanceController) ensurePrivilegedTiDBExist(instance *tiapi.TidbInstance) error {
	ns := instance.Name
	deploymentName := getName("privileged-tidb")
	_, err := tic.cli.AppsV1beta1().Deployments(ns).Get(deploymentName, metav1.GetOptions{})
	if err != nil {
		if !apierrors.IsNotFound(err) {
			log.Errorf("failed to get deployment %s/%s: %v", ns, deploymentName, err)
			return err
		}
		ts, err := tic.cli.PingcapV1().TidbSets(ns).Get(getName("tidbset"), metav1.GetOptions{})
		if err != nil {
			return err
		}
		if len(ts.Status.TiDBStatus) == 0 {
			log.Infof("TiDB not bootstrapped, skip creating privileged tidb")
			return nil
		}
		for _, status := range ts.Status.TiDBStatus {
			if status.IP != "" {
				if err := tic.createPrivilegedTiDB(instance); err != nil {
					return err
				}
				return nil
			}
		}
		log.Infof("TiDB not bootstrapped, skip creating privileged tidb")
	}
	return nil
}

func (tic *TiDBInstanceController) createPrivilegedTiDB(instance *tiapi.TidbInstance) error {
	ns := instance.Name
	image := GetImageByTiPodSpec(TiDBContainerType, instance.Spec.TiDB)
	ownerRef := generateOwnerRef(instance)
	l := label.New().Cluster(clusterName).App("privileged-tidb").Labels()
	name := getName("privileged-tidb")
	cmd := []string{
		"/tidb-server",
		"--run-ddl=false",
		"--store=tikv",
		"--host=0.0.0.0",
		fmt.Sprintf("--path=%s:2379", getName("pd")),
		"--config=/etc/tidb/tidb.toml",
	}
	deploy := &v1beta1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:            name,
			Namespace:       ns,
			Labels:          l,
			OwnerReferences: []metav1.OwnerReference{ownerRef},
		},
		Spec: v1beta1.DeploymentSpec{
			Template: v1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Name:   name,
					Labels: l,
				},
				Spec: v1.PodSpec{
					Containers: []v1.Container{
						{
							Name:    "privileged-tidb",
							Image:   image,
							Command: cmd,
							Ports: []v1.ContainerPort{
								{
									Name:          "mysql-client",
									ContainerPort: int32(4000),
									Protocol:      v1.ProtocolTCP,
								},
							},
							VolumeMounts: []v1.VolumeMount{
								{
									Name:      "timezone",
									ReadOnly:  true,
									MountPath: "/etc/localtime",
								},
								{
									Name:      "config",
									ReadOnly:  true,
									MountPath: "/etc/tidb",
								},
							},
						},
					},
					RestartPolicy: v1.RestartPolicyAlways,
					Volumes: []v1.Volume{
						{
							Name:         "timezone",
							VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: "/etc/localtime"}},
						},
						{
							Name: "config",
							VolumeSource: v1.VolumeSource{
								ConfigMap: &v1.ConfigMapVolumeSource{
									LocalObjectReference: v1.LocalObjectReference{Name: fmt.Sprintf("%s-config", clusterName)},
									Items:                []v1.KeyToPath{{Key: "privileged-tidb-config", Path: "tidb.toml"}},
								},
							},
						},
					},
				},
			},
		},
	}
	if _, err := tic.cli.AppsV1beta1().Deployments(ns).Create(deploy); err != nil {
		log.Errorf("failed to create deployment %s/%s: %v", ns, name, err)
		return err
	}
	svc := &v1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:            name,
			Namespace:       ns,
			Labels:          l,
			OwnerReferences: []metav1.OwnerReference{ownerRef},
		},
		Spec: v1.ServiceSpec{
			Ports: []v1.ServicePort{
				{
					Name:       "mysql-client",
					Port:       4000,
					TargetPort: intstr.FromInt(4000),
					Protocol:   v1.ProtocolTCP,
				},
			},
			Selector: l,
		},
	}
	if ExposePrivilegedTiDBSVC {
		svc.Spec.Type = v1.ServiceTypeNodePort
	}
	if _, err := tic.cli.CoreV1().Services(ns).Create(svc); err != nil && !apierrors.IsAlreadyExists(err) {
		log.Errorf("failed to create service %s/%s: %v", ns, name, err)
		return err
	}
	return nil
}

func (tic *TiDBInstanceController) syncMonitor(instance *tiapi.TidbInstance) error {
	ns := instance.Name

	if err := tic.syncMonitorSecret(instance); err != nil {
		log.Errorf("failed to sync monitor secret in %s: %v", ns, err)
		return err
	}
	if err := tic.syncMonitorJob(instance); err != nil {
		log.Errorf("failed to sync configure monitor job in %s: %v", ns, err)
		return err
	}
	if err := tic.syncMonitorService(instance); err != nil {
		log.Errorf("failed to sync monitor service in %s: %v", ns, err)
		return err
	}
	if err := tic.syncMonitorDeployment(instance); err != nil {
		log.Errorf("failed to sync monitor deployment in %s: %v", ns, err)
		return err
	}
	return nil
}

func (tic *TiDBInstanceController) syncMonitorSecret(instance *tiapi.TidbInstance) error {
	ns := instance.Name
	username, ok := instance.Spec.Secrets["grafana-username"]
	if !ok {
		username = "admin"
	}
	username_enc := base64.StdEncoding.EncodeToString([]byte(username))

	password, ok := instance.Spec.Secrets["grafana-password"]
	if !ok {
		password = "admin"
	}
	password_enc := base64.StdEncoding.EncodeToString([]byte(password))

	secret := &v1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      getSecretName(ns),
			Namespace: ns,
		},
		Data: map[string][]byte{
			"grafana-username": []byte(username_enc),
			"grafana-password": []byte(password_enc),
		},
	}

	_, err := tic.cli.CoreV1().Secrets(ns).Create(secret)
	if err != nil && !apierrors.IsAlreadyExists(err) {
		log.Errorf("failed to create grafana secret: [%s/%s]; %v", ns, secret.GetName(), err)
		return err
	}

	return nil
}

func (tic *TiDBInstanceController) syncMonitorJob(instance *tiapi.TidbInstance) error {
	ns := instance.Name
	jobName := getName("configure-grafana")
	jobLabel := label.New().Cluster(clusterName).App("configure-grafana").Labels()
	job, err := tic.cli.BatchV1().Jobs(ns).Get(jobName, metav1.GetOptions{})
	if err != nil {
		if !apierrors.IsNotFound(err) {
			log.Errorf("failed to get job %s/%s: %v", ns, jobName, err)
			return err
		}
		job = &batchv1.Job{
			ObjectMeta: metav1.ObjectMeta{
				Name:   jobName,
				Labels: jobLabel,
			},
			Spec: batchv1.JobSpec{
				Template: v1.PodTemplateSpec{
					ObjectMeta: metav1.ObjectMeta{
						Name: jobName,
					},
					Spec: v1.PodSpec{
						RestartPolicy: v1.RestartPolicyOnFailure,
						Containers: []v1.Container{
							{
								Name:  string(InstallerContainerType),
								Image: GetImageByTiPodSpec(InstallerContainerType, instance.Spec.Monitor),
								Args:  []string{fmt.Sprintf("%s:3000", getName("grafana"))},
							},
						},
					},
				},
			},
		}
		if _, err := tic.cli.BatchV1().Jobs(ns).Create(job); err != nil {
			log.Errorf("failed to create job %s/%s: %v", ns, jobName, err)
			return errors.Trace(err)
		}
	}

	return nil
}

func (tic *TiDBInstanceController) syncMonitorService(instance *tiapi.TidbInstance) error {
	ns := instance.Name
	ownerRef := generateOwnerRef(instance)
	svcLabel := label.New().Cluster(clusterName).App("tidb-monitor").Labels()
	services := map[string]int{
		"prometheus": 9090,
		"grafana":    3000,
	}
	for kind, port := range services {
		svc := &v1.Service{
			ObjectMeta: metav1.ObjectMeta{
				Name:            getName(kind),
				Namespace:       ns,
				Labels:          svcLabel,
				OwnerReferences: []metav1.OwnerReference{ownerRef},
			},
			Spec: v1.ServiceSpec{
				Ports: []v1.ServicePort{
					{
						Name:       kind,
						Port:       int32(port),
						TargetPort: intstr.FromInt(port),
						Protocol:   v1.ProtocolTCP,
					},
				},
				Selector: svcLabel,
			},
		}
		if kind == "prometheus" || kind == "grafana" {
			svc.Spec.Type = v1.ServiceTypeNodePort
		}
		if _, err := tic.cli.CoreV1().Services(ns).Create(svc); err != nil && !apierrors.IsAlreadyExists(err) {
			return err
		}
	}
	return nil
}

func (tic *TiDBInstanceController) syncMonitorDeployment(instance *tiapi.TidbInstance) error {
	ns := instance.Name
	ownerRef := generateOwnerRef(instance)
	deploymentName := getName("monitor")
	configName := getName("config")
	deploymentLabel := label.New().Cluster(clusterName).App("tidb-monitor").Labels()
	reserveDays := promRetentionDays
	if instance.Spec.MonitorReserveDays > 0 {
		reserveDays = instance.Spec.MonitorReserveDays
	}

	deploy, err := tic.cli.AppsV1beta1().Deployments(ns).Get(deploymentName, metav1.GetOptions{})
	if err != nil {
		if !apierrors.IsNotFound(err) {
			log.Errorf("failed to get TiDB monitor %s/%s: %v", ns, deploymentName, err)
			return err
		}

		prometheusCS := getContainerSpec(PrometheusContainerType, instance.Spec.Monitor)
		grafanaCS := getContainerSpec(GrafanaContainerType, instance.Spec.Monitor)

		deploy = &v1beta1.Deployment{
			ObjectMeta: metav1.ObjectMeta{
				Name:            deploymentName,
				Namespace:       ns,
				Labels:          deploymentLabel,
				OwnerReferences: []metav1.OwnerReference{ownerRef},
			},
			Spec: v1beta1.DeploymentSpec{
				Template: v1.PodTemplateSpec{
					ObjectMeta: metav1.ObjectMeta{
						Name:   deploymentName,
						Labels: deploymentLabel,
					},
					Spec: v1.PodSpec{
						Affinity: affinityForNodeSelector(instance.Spec.Monitor.NodeSelector),
						Containers: []v1.Container{
							{
								Name:  string(PrometheusContainerType),
								Image: GetImageByTiPodSpec(PrometheusContainerType, instance.Spec.Monitor),
								Command: []string{
									"/bin/prometheus",
									"--config.file=/etc/prometheus/prometheus.yml",
									fmt.Sprintf("--storage.tsdb.retention=%dd", reserveDays),
									"--storage.tsdb.path=/prometheus",
								},
								Ports: []v1.ContainerPort{
									{
										Name:          string(PrometheusContainerType),
										ContainerPort: int32(9090),
										Protocol:      v1.ProtocolTCP,
									},
								},
								VolumeMounts: []v1.VolumeMount{
									{Name: "timezone", ReadOnly: true, MountPath: "/etc/localtime"},
									{Name: "prometheus-config", ReadOnly: true, MountPath: "/etc/prometheus"},
									{Name: "prometheus-data", MountPath: "/prometheus"},
								},
								Resources: *getResourceRequirements(prometheusCS),
							},
							{
								Name:  string(GrafanaContainerType),
								Image: GetImageByTiPodSpec(GrafanaContainerType, instance.Spec.Monitor),
								Ports: []v1.ContainerPort{
									{
										Name:          string(GrafanaContainerType),
										ContainerPort: int32(3000),
										Protocol:      v1.ProtocolTCP,
									},
								},
								VolumeMounts: []v1.VolumeMount{
									{Name: "timezone", ReadOnly: true, MountPath: "/etc/localtime"},
									{Name: "grafana-data", MountPath: "/data"},
								},
								Env: []v1.EnvVar{
									{Name: "GF_PATHS_DATA", Value: "/data"},
								},
								Resources: *getResourceRequirements(grafanaCS),
							},
						},
						Volumes: []v1.Volume{
							{Name: "timezone", VolumeSource: v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: "/etc/localtime"}}},
							{Name: "prometheus-data", VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}},
							{Name: "grafana-data", VolumeSource: v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}},
							{Name: "prometheus-config", VolumeSource: v1.VolumeSource{
								ConfigMap: &v1.ConfigMapVolumeSource{
									LocalObjectReference: v1.LocalObjectReference{Name: configName},
									Items: []v1.KeyToPath{
										{Key: "prometheus-config", Path: "prometheus.yml"},
										{Key: "alert-rules-config", Path: "alert.rules"},
									},
								},
							}},
						},
					},
				},
			},
		}
		if RBACEnabled {
			deploy.Spec.Template.Spec.ServiceAccountName = saName
		}
		if _, err := tic.cli.AppsV1beta1().Deployments(ns).Create(deploy); err != nil {
			log.Errorf("failed to create tidb monitor %s/%s: %v", ns, deploymentName, err)
			return err
		}
		return nil
	}
	return nil
}