a tidb-scheduler bug fix and emit event when scheduled failed

pkg/scheduler/predicates/ha.go

@@ -14,7 +14,9 @@
 package predicates
 
 import (
+	"errors"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -27,18 +29,23 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
+	kubescheme "k8s.io/client-go/kubernetes/scheme"
+	eventv1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
 )
 
 type ha struct {
 	lock          sync.Mutex
 	kubeCli       kubernetes.Interface
 	cli           versioned.Interface
 	podListFn     func(ns, instanceName, component string) (*apiv1.PodList, error)
+	podGetFn      func(ns, podName string) (*apiv1.Pod, error)
 	pvcGetFn      func(ns, pvcName string) (*apiv1.PersistentVolumeClaim, error)
 	tcGetFn       func(ns, tcName string) (*v1alpha1.TidbCluster, error)
 	pvcListFn     func(ns, instanceName, component string) (*apiv1.PersistentVolumeClaimList, error)
 	updatePVCFn   func(*apiv1.PersistentVolumeClaim) error
 	acquireLockFn func(*apiv1.Pod) (*apiv1.PersistentVolumeClaim, *apiv1.PersistentVolumeClaim, error)
+	recorder      record.EventRecorder
 }
 
 // NewHA returns a Predicate
@@ -48,11 +55,19 @@ func NewHA(kubeCli kubernetes.Interface, cli versioned.Interface) Predicate {
 		cli:     cli,
 	}
 	h.podListFn = h.realPodListFn
+	h.podGetFn = h.realPodGetFn
 	h.pvcGetFn = h.realPVCGetFn
 	h.tcGetFn = h.realTCGetFn
 	h.pvcListFn = h.realPVCListFn
 	h.updatePVCFn = h.realUpdatePVCFn
 	h.acquireLockFn = h.realAcquireLock
+
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(glog.Infof)
+	eventBroadcaster.StartRecordingToSink(&eventv1.EventSinkImpl{
+		Interface: eventv1.New(h.kubeCli.CoreV1().RESTClient()).Events("")})
+	h.recorder = eventBroadcaster.NewRecorder(kubescheme.Scheme, apiv1.EventSource{Component: "tidb-scheduler"})
+
 	return h
 }
 
@@ -143,11 +158,16 @@ func (h *ha) Filter(instanceName string, pod *apiv1.Pod, nodes []apiv1.Node) ([]
 	}
 
 	if len(minNodeNames) == 0 {
-		return nil, fmt.Errorf("can't find a node from: %v, nodeMap: %v", nodes, nodeMap)
+		msg := fmt.Sprintf("can't scheduled to nodes: %v, because these pods had been scheduled to nodes: %v", GetNodeNames(nodes), nodeMap)
+		h.recorder.Event(pod, apiv1.EventTypeWarning, "FailedScheduling", msg)
+		return nil, errors.New(msg)
 	}
 	return getNodeFromNames(nodes, minNodeNames), nil
 }
 
+// kubernetes scheduling is parallel, to achieve HA, we must ensure the scheduling is serial,
+// so when a pod is scheduling, we set an annotation to its PVC, other pods can't be scheduled at this time,
+// delete the PVC's annotation when the pod is scheduled(PVC is bound and the pod's nodeName is set)
 func (h *ha) realAcquireLock(pod *apiv1.Pod) (*apiv1.PersistentVolumeClaim, *apiv1.PersistentVolumeClaim, error) {
 	ns := pod.GetNamespace()
 	component := pod.Labels[label.ComponentLabelKey]
@@ -180,7 +200,12 @@ func (h *ha) realAcquireLock(pod *apiv1.Pod) (*apiv1.PersistentVolumeClaim, *api
 	if schedulingPVC == currentPVC {
 		return schedulingPVC, currentPVC, nil
 	}
-	if schedulingPVC.Status.Phase != apiv1.ClaimBound {
+	schedulingPodName := getPodNameFromPVC(schedulingPVC)
+	schedulingPod, err := h.podGetFn(ns, schedulingPodName)
+	if err != nil {
+		return schedulingPVC, currentPVC, err
+	}
+	if schedulingPVC.Status.Phase != apiv1.ClaimBound || schedulingPod.Spec.NodeName == "" {
 		return schedulingPVC, currentPVC, fmt.Errorf("waiting for Pod %s/%s scheduling", ns, strings.TrimPrefix(schedulingPVC.GetName(), component))
 	}
 
@@ -199,6 +224,10 @@ func (h *ha) realPodListFn(ns, instanceName, component string) (*apiv1.PodList,
 	})
 }
 
+func (h *ha) realPodGetFn(ns, podName string) (*apiv1.Pod, error) {
+	return h.kubeCli.CoreV1().Pods(ns).Get(podName, metav1.GetOptions{})
+}
+
 func (h *ha) realPVCListFn(ns, instanceName, component string) (*apiv1.PersistentVolumeClaimList, error) {
 	selector := label.New().Instance(instanceName).Component(component).Labels()
 	return h.kubeCli.CoreV1().PersistentVolumeClaims(ns).List(metav1.ListOptions{
@@ -242,3 +271,16 @@ func getReplicasFrom(tc *v1alpha1.TidbCluster, component string) int32 {
 func pvcName(component, podName string) string {
 	return fmt.Sprintf("%s-%s", component, podName)
 }
+
+func GetNodeNames(nodes []apiv1.Node) []string {
+	nodeNames := make([]string, 0)
+	for _, node := range nodes {
+		nodeNames = append(nodeNames, node.GetName())
+	}
+	sort.Strings(nodeNames)
+	return nodeNames
+}
+
+func getPodNameFromPVC(pvc *apiv1.PersistentVolumeClaim) string {
+	return strings.TrimPrefix(pvc.Name, fmt.Sprintf("%s-", pvc.Labels[label.ComponentLabelKey]))
+}