Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Serial scheduling #266

Merged
merged 4 commits into from
Jan 21, 2019
Merged

Serial scheduling #266

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8d3542e
Serial scheduling
weekface Jan 15, 2019
8a8b61d
address comment
weekface Jan 16, 2019
39999cf
address comment
weekface Jan 17, 2019
91a2c4a
make check happy
weekface Jan 21, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions charts/tidb-operator/templates/scheduler-rbac.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ rules:
verbs: ["get"]
- apiGroups: [""]
resources: ["persistentvolumeclaims"]
verbs: ["get"]
verbs: ["get", "list", "update"]
---
kind: ClusterRoleBinding
apiVersion: rbac.authorization.k8s.io/v1beta1
Expand Down Expand Up @@ -114,7 +114,7 @@ rules:
verbs: ["get"]
- apiGroups: [""]
resources: ["persistentvolumeclaims"]
verbs: ["get"]
verbs: ["get", "list", "update"]
---
kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1beta1
Expand Down
2 changes: 2 additions & 0 deletions pkg/label/label.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,8 @@ const (
AnnPodNameKey string = "tidb.pingcap.com/pod-name"
// AnnPVCDeferDeleting is pvc defer deletion annotation key used in PVC for defer deleting PVC
AnnPVCDeferDeleting = "tidb.pingcap.com/pvc-defer-deleting"
// AnnPVCPodScheduling is pod scheduling annotation key, it represents whether the pod is scheduling
AnnPVCPodScheduling = "tidb.pingcap.com/pod-scheduling"

// PDLabelVal is PD label value
PDLabelVal string = "pd"
Expand Down
3 changes: 3 additions & 0 deletions pkg/manager/meta/reclaim_policy_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,9 @@ func (rpm *reclaimPolicyManager) Sync(tc *v1alpha1.TidbCluster) error {
}

for _, pvc := range pvcs {
if pvc.Spec.VolumeName == "" {
continue
}
pv, err := rpm.pvLister.Get(pvc.Spec.VolumeName)
if err != nil {
return err
Expand Down
55 changes: 35 additions & 20 deletions pkg/manager/meta/reclaim_policy_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,11 +34,12 @@ import (
func TestReclaimPolicyManagerSync(t *testing.T) {
g := NewGomegaWithT(t)
type testcase struct {
name string
pvcHasLabels bool
updateErr bool
err bool
changed bool
name string
pvcHasLabels bool
pvcHasVolumeName bool
updateErr bool
err bool
changed bool
}

testFn := func(test *testcase, t *testing.T) {
Expand All @@ -50,6 +51,9 @@ func TestReclaimPolicyManagerSync(t *testing.T) {
if !test.pvcHasLabels {
pvc1.Labels = nil
}
if !test.pvcHasVolumeName {
pvc1.Spec.VolumeName = ""
}

rpm, fakePVControl, pvcIndexer, pvIndexer := newFakeReclaimPolicyManager()
err := pvcIndexer.Add(pvc1)
Expand Down Expand Up @@ -82,25 +86,36 @@ func TestReclaimPolicyManagerSync(t *testing.T) {

tests := []testcase{
{
name: "normal",
pvcHasLabels: true,
updateErr: false,
err: false,
changed: true,
name: "normal",
pvcHasLabels: true,
pvcHasVolumeName: true,
updateErr: false,
err: false,
changed: true,
},
{
name: "pvc don't have labels",
pvcHasLabels: false,
pvcHasVolumeName: true,
updateErr: false,
err: false,
changed: false,
},
{
name: "pvc don't have labels",
pvcHasLabels: false,
updateErr: false,
err: false,
changed: false,
name: "pvc don't have volumeName",
pvcHasLabels: false,
pvcHasVolumeName: false,
updateErr: false,
err: false,
changed: false,
},
{
name: "update failed",
pvcHasLabels: true,
updateErr: true,
err: true,
changed: false,
name: "update failed",
pvcHasLabels: true,
pvcHasVolumeName: true,
updateErr: true,
err: true,
changed: false,
},
}

Expand Down
95 changes: 89 additions & 6 deletions pkg/scheduler/predicates/ha.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@ package predicates
import (
"fmt"
"strings"
"sync"
"time"

"github.com/golang/glog"
"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1alpha1"
Expand All @@ -28,11 +30,15 @@ import (
)

type ha struct {
kubeCli kubernetes.Interface
cli versioned.Interface
podListFn func(ns, instanceName, component string) (*apiv1.PodList, error)
pvcGetFn func(ns, pvcName string) (*apiv1.PersistentVolumeClaim, error)
tcGetFn func(ns, tcName string) (*v1alpha1.TidbCluster, error)
lock sync.Mutex
kubeCli kubernetes.Interface
cli versioned.Interface
podListFn func(ns, instanceName, component string) (*apiv1.PodList, 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)
}

// NewHA returns a Predicate
Expand All @@ -44,6 +50,9 @@ func NewHA(kubeCli kubernetes.Interface, cli versioned.Interface) Predicate {
h.podListFn = h.realPodListFn
h.pvcGetFn = h.realPVCGetFn
h.tcGetFn = h.realTCGetFn
h.pvcListFn = h.realPVCListFn
h.updatePVCFn = h.realUpdatePVCFn
h.acquireLockFn = h.realAcquireLock
return h
}

Expand All @@ -55,6 +64,9 @@ func (h *ha) Name() string {
// 2. return these nodes that have least pods and its pods count is less than (replicas+1)/2 to kube-scheduler
// 3. let kube-scheduler to make the final decision
func (h *ha) Filter(instanceName string, pod *apiv1.Pod, nodes []apiv1.Node) ([]apiv1.Node, error) {
h.lock.Lock()
defer h.lock.Unlock()

ns := pod.GetNamespace()
podName := pod.GetName()
component := pod.Labels[label.ComponentLabelKey]
Expand All @@ -63,9 +75,12 @@ func (h *ha) Filter(instanceName string, pod *apiv1.Pod, nodes []apiv1.Node) ([]
if len(nodes) == 0 {
return nil, fmt.Errorf("kube nodes is empty")
}
if _, _, err := h.acquireLockFn(pod); err != nil {
return nil, err
}

if len(nodes) == 1 {
pvcName := fmt.Sprintf("%s-%s", component, podName)
pvcName := pvcName(component, podName)
pvc, err := h.pvcGetFn(ns, pvcName)
if err != nil {
return nil, err
Expand Down Expand Up @@ -127,13 +142,69 @@ func (h *ha) Filter(instanceName string, pod *apiv1.Pod, nodes []apiv1.Node) ([]
return getNodeFromNames(nodes, minNodeNames), nil
}

func (h *ha) realAcquireLock(pod *apiv1.Pod) (*apiv1.PersistentVolumeClaim, *apiv1.PersistentVolumeClaim, error) {
ns := pod.GetNamespace()
component := pod.Labels[label.ComponentLabelKey]
instanceName := pod.Labels[label.InstanceLabelKey]
podName := pod.GetName()
pvcList, err := h.pvcListFn(ns, instanceName, component)
if err != nil {
return nil, nil, err
}

currentPVCName := pvcName(component, podName)
var currentPVC *apiv1.PersistentVolumeClaim
var schedulingPVC *apiv1.PersistentVolumeClaim
for i := 0; i < len(pvcList.Items); i++ {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
for i := 0; i < len(pvcList.Items); i++ {
for _, pvc := range pvcList.Items {

Copy link
Contributor Author

@weekface weekface Jan 16, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use this style to avoid the golang common mistakes

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is not use closures running as goroutines，If there is such a scene, then the way to use range can also avoid this problem.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

currentPVC and schedulingPVC are pointers, the pvc in the for range block is a variable, if we use the for range style, then these two variables will get the same address.

Copy link
Contributor Author

@weekface weekface Jan 17, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example:

package main

import "fmt"

func main() {
	arr := []string{"a", "b", "c"}

	var pa *string
	var pb *string
	for _, item := range arr {
		if item == "a" {
			pa = &item
		}
		if item == "b" {
			pb = &item
		}
	}
	fmt.Println(pa)
	fmt.Println(pb)
	fmt.Println(pa == pb)

	var ppa *string
	var ppb *string
	for i := 0; i < len(arr); i++ {
		if arr[i] == "a" {
			ppa = &arr[i]
		}
		if arr[i] == "b" {
			ppb = &arr[i]
		}
	}
	fmt.Println(ppa)
	fmt.Println(ppb)
	fmt.Println(ppa == ppb)
}

Outputs:

0xc00005e1c0
0xc00005e1c0
true
0xc000060180
0xc000060190
false

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can create a new variable to fix this problem. such as follow:

for i := 0; i < len(pvcList.Items); i++ { 
    pvc := pvcList.Items[i]
    ...
}
------
for _, pvc := range pvcList.Items {
    pvc := pvc
 ...
}

Copy link
Contributor Author

@weekface weekface Jan 17, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Style for i:=0; i<n; i++ is equivalent to the for range style, but using for i:=0; i<n; i++ is more clear at this situation.

pvc := pvc looks weird.

pvc := pvcList.Items[i]
if pvc.GetName() == currentPVCName {
currentPVC = &pvcList.Items[i]
}
if pvc.Annotations[label.AnnPVCPodScheduling] != "" && schedulingPVC == nil {
schedulingPVC = &pvcList.Items[i]
}
}

if currentPVC == nil {
return schedulingPVC, currentPVC, fmt.Errorf("can't find current Pod %s/%s's PVC", ns, podName)
}
if schedulingPVC == nil {
return schedulingPVC, currentPVC, h.setCurrentPodScheduling(currentPVC)
}
if schedulingPVC == currentPVC {
return schedulingPVC, currentPVC, nil
}
if schedulingPVC.Status.Phase != apiv1.ClaimBound {
return schedulingPVC, currentPVC, fmt.Errorf("waiting for Pod %s/%s scheduling", ns, strings.TrimPrefix(schedulingPVC.GetName(), component))
}

delete(schedulingPVC.Annotations, label.AnnPVCPodScheduling)
err = h.updatePVCFn(schedulingPVC)
if err != nil {
return schedulingPVC, currentPVC, err
}
return schedulingPVC, currentPVC, h.setCurrentPodScheduling(currentPVC)
}

func (h *ha) realPodListFn(ns, instanceName, component string) (*apiv1.PodList, error) {
selector := label.New().Instance(instanceName).Component(component).Labels()
return h.kubeCli.CoreV1().Pods(ns).List(metav1.ListOptions{
LabelSelector: labels.SelectorFromSet(selector).String(),
})
}

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{
LabelSelector: labels.SelectorFromSet(selector).String(),
})
}

func (h *ha) realUpdatePVCFn(pvc *apiv1.PersistentVolumeClaim) error {
_, err := h.kubeCli.CoreV1().PersistentVolumeClaims(pvc.GetNamespace()).Update(pvc)
return err
}

func (h *ha) realPVCGetFn(ns, pvcName string) (*apiv1.PersistentVolumeClaim, error) {
return h.kubeCli.CoreV1().PersistentVolumeClaims(ns).Get(pvcName, metav1.GetOptions{})
}
Expand All @@ -142,6 +213,14 @@ func (h *ha) realTCGetFn(ns, tcName string) (*v1alpha1.TidbCluster, error) {
return h.cli.PingcapV1alpha1().TidbClusters(ns).Get(tcName, metav1.GetOptions{})
}

func (h *ha) setCurrentPodScheduling(pvc *apiv1.PersistentVolumeClaim) error {
if pvc.Annotations == nil {
pvc.Annotations = map[string]string{}
}
pvc.Annotations[label.AnnPVCPodScheduling] = time.Now().Format(time.RFC3339)
return h.updatePVCFn(pvc)
}

func getTCNameFromPod(pod *apiv1.Pod, component string) string {
return strings.TrimSuffix(pod.GenerateName, fmt.Sprintf("-%s-", component))
}
Expand All @@ -153,3 +232,7 @@ func getReplicasFrom(tc *v1alpha1.TidbCluster, component string) int32 {

return tc.Spec.TiKV.Replicas
}

func pvcName(component, podName string) string {
return fmt.Sprintf("%s-%s", component, podName)
}
Loading