Sync pump status to tidb-cluster.

pkg/apis/pingcap/v1alpha1/types.go

@@ -168,6 +168,7 @@ type TidbClusterStatus struct {
 	PD        PDStatus   `json:"pd,omitempty"`
 	TiKV      TiKVStatus `json:"tikv,omitempty"`
 	TiDB      TiDBStatus `json:"tidb,omitempty"`
+	Pump      PumpStatus `josn:"tidb,omitempty"`
 }
 
 // +k8s:openapi-gen=true
@@ -460,6 +461,18 @@ type TiKVFailureStore struct {
 	CreatedAt metav1.Time `json:"createdAt,omitempty"`
 }
 
+// PumpStatus is Pump status
+type PumpStatus struct {
+	Phase       MemberPhase             `json:"phase,omitempty"`
+	StatefulSet *apps.StatefulSetStatus `json:"statefulSet,omitempty"`
+	Members     map[string]PumpMember   `json:"members,omitempty"`
+}
+
+// PumpMember is Pump member
+type PumpMember struct {
+	Name string `json:"name"`
+}
+
 // +genclient
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 

pkg/controller/tidbcluster/tidb_cluster_controller.go

@@ -193,6 +193,7 @@ func NewController(
 				setInformer.Lister(),
 				svcInformer.Lister(),
 				cmInformer.Lister(),
+				podInformer.Lister(),
 			),
 			recorder,
 		),

pkg/manager/member/pump_member_manager.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tidb-operator/pkg/label"
 	"github.com/pingcap/tidb-operator/pkg/manager"
 	"github.com/pingcap/tidb-operator/pkg/util"
+	apps "k8s.io/api/apps/v1"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
@@ -64,6 +65,7 @@ type pumpMemberManager struct {
 	setLister  v1.StatefulSetLister
 	svcLister  corelisters.ServiceLister
 	cmLister   corelisters.ConfigMapLister
+	podLister  corelisters.PodLister
 }
 
 // NewPumpMemberManager returns a controller to reconcile pump clusters
@@ -73,14 +75,16 @@ func NewPumpMemberManager(
 	cmControl controller.ConfigMapControlInterface,
 	setLister v1.StatefulSetLister,
 	svcLister corelisters.ServiceLister,
-	cmLister corelisters.ConfigMapLister) manager.Manager {
+	cmLister corelisters.ConfigMapLister,
+	podLister corelisters.PodLister) manager.Manager {
 	return &pumpMemberManager{
 		setControl,
 		svcControl,
 		cmControl,
 		setLister,
 		svcLister,
 		cmLister,
+		podLister,
 	}
 }
 
@@ -91,12 +95,11 @@ func (pmm *pumpMemberManager) Sync(tc *v1alpha1.TidbCluster) error {
 	if err := pmm.syncHeadlessService(tc); err != nil {
 		return err
 	}
-	return pmm.syncStatefulSet(tc)
+	return pmm.syncPumpStatefulSetForTidbCluster(tc)
 }
 
-// syncStatefulSet syncs the pump statefulset
-// TODO: sync statefulset status of pump to tidbcluster
-func (pmm *pumpMemberManager) syncStatefulSet(tc *v1alpha1.TidbCluster) error {
+//syncPumpStatefulSetForTidbCluster sync statefulset status of pump to tidbcluster
+func (pmm *pumpMemberManager) syncPumpStatefulSetForTidbCluster(tc *v1alpha1.TidbCluster) error {
 
 	oldPumpSetTemp, err := pmm.setLister.StatefulSets(tc.Namespace).Get(controller.PumpMemberName(tc.Name))
 	if err != nil && !errors.IsNotFound(err) {
@@ -143,6 +146,27 @@ func (pmm *pumpMemberManager) syncStatefulSet(tc *v1alpha1.TidbCluster) error {
 	return nil
 }
 
+func (pmm *pumpMemberManager) syncTiDBClusterStatus(tc *v1alpha1.TidbCluster, set *apps.StatefulSet) error {
+
+	tc.Status.Pump.StatefulSet = &set.Status
+
+	upgrading, err := pmm.pumpStatefulSetIsUpgrading(set, tc)
+	if err != nil {
+		return err
+	}
+	if upgrading {
+		tc.Status.Pump.Phase = v1alpha1.UpgradePhase
+	} else {
+		tc.Status.Pump.Phase = v1alpha1.NormalPhase
+	}
+
+	pumpStatus := map[string]v1alpha1.PumpMember{}
+
+	tc.Status.Pump.Members = pumpStatus
+
+	return nil
+}
+
 func (pmm *pumpMemberManager) syncHeadlessService(tc *v1alpha1.TidbCluster) error {
 
 	newSvc := getNewPumpHeadlessService(tc)
@@ -475,6 +499,33 @@ func getPumpLogLevel(tc *v1alpha1.TidbCluster) string {
 	return logLevel
 }
 
+func (pmm *pumpMemberManager) pumpStatefulSetIsUpgrading(set *apps.StatefulSet, tc *v1alpha1.TidbCluster) (bool, error) {
+	if statefulSetIsUpgrading(set) {
+		return true, nil
+	}
+	selector, err := label.New().
+		Instance(tc.GetLabels()[label.InstanceLabelKey]).
+		Pump().
+		Selector()
+	if err != nil {
+		return false, err
+	}
+	pumpPods, err := pmm.podLister.Pods(tc.GetNamespace()).List(selector)
+	if err != nil {
+		return false, err
+	}
+	for _, pod := range pumpPods {
+		revisionHash, exist := pod.Labels[apps.ControllerRevisionHashLabelKey]
+		if !exist {
+			return false, nil
+		}
+		if revisionHash != tc.Status.Pump.StatefulSet.UpdateRevision {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 type FakePumpMemberManager struct {
 	err error
 }

pkg/manager/member/pump_member_manager_test.go

@@ -442,6 +442,7 @@ func newFakePumpMemberManager() (*pumpMemberManager, *pumpFakeControls, *pumpFak
 	svcInformer := kubeinformers.NewSharedInformerFactory(kubeCli, 0).Core().V1().Services()
 	epsInformer := kubeinformers.NewSharedInformerFactory(kubeCli, 0).Core().V1().Endpoints()
 	cmInformer := kubeinformers.NewSharedInformerFactory(kubeCli, 0).Core().V1().ConfigMaps()
+	podInformer := kubeinformers.NewSharedInformerFactory(kubeCli, 0).Core().V1().Pods()
 	setControl := controller.NewFakeStatefulSetControl(setInformer, tcInformer)
 	svcControl := controller.NewFakeServiceControl(svcInformer, epsInformer, tcInformer)
 	cmControl := controller.NewFakeConfigMapControl(cmInformer)
@@ -452,6 +453,7 @@ func newFakePumpMemberManager() (*pumpMemberManager, *pumpFakeControls, *pumpFak
 		setInformer.Lister(),
 		svcInformer.Lister(),
 		cmInformer.Lister(),
+		podInformer.Lister(),
 	}
 	controls := &pumpFakeControls{
 		svc: svcControl,