Merge pull request #139 from yiwenshao/add-preemption

implement preemption for galaxy-ipam

Author: chenchun

pkg/api/k8s/schedulerapi/schedulerapi.go

@@ -98,6 +98,43 @@ type LabelPreference struct {
 	Presence bool `json:"presence"`
 }
 
+// ExtenderPreemptionResult represents the result returned by preemption phase of extender.
+type ExtenderPreemptionResult struct {
+	NodeNameToMetaVictims map[string]*MetaVictims
+}
+
+// ExtenderPreemptionArgs represents the arguments needed by the extender to preempt pods on nodes.
+type ExtenderPreemptionArgs struct {
+	// Pod being scheduled
+	Pod *corev1.Pod
+	// Victims map generated by scheduler preemption phase
+	// Only set NodeNameToMetaVictims if Extender.NodeCacheCapable == true. Otherwise, only set NodeNameToVictims.
+	NodeNameToVictims     map[string]*Victims
+	NodeNameToMetaVictims map[string]*MetaVictims
+}
+
+// Victims represents:
+//   pods:  a group of pods expected to be preempted.
+//   numPDBViolations: the count of violations of PodDisruptionBudget
+type Victims struct {
+	Pods             []*corev1.Pod
+	NumPDBViolations int64
+}
+
+// MetaPod represent identifier for a v1.Pod
+type MetaPod struct {
+	UID string
+}
+
+// MetaVictims represents:
+//   pods:  a group of pods expected to be preempted.
+//     Only Pod identifiers will be sent and user are expect to get v1.Pod in their own way.
+//   numPDBViolations: the count of violations of PodDisruptionBudget
+type MetaVictims struct {
+	Pods             []*MetaPod
+	NumPDBViolations int64
+}
+
 // ExtenderArgs represents the arguments needed by the extender to filter/prioritize
 // nodes for a pod.
 type ExtenderArgs struct {

pkg/ipam/context/context.go

@@ -42,13 +42,15 @@ type IPAMContext struct {
 	DynamicClient dynamic.Interface
 
 	PodLister         corev1lister.PodLister
+	NodeLister        corev1lister.NodeLister
 	StatefulSetLister appv1.StatefulSetLister
 	DeploymentLister  appv1.DeploymentLister
 	PoolLister        list.PoolLister
 	ExtensionLister   extensionlister.CustomResourceDefinitionLister
 
-	PodInformer coreinformer.PodInformer
-	FIPInformer galaxyinformer.FloatingIPInformer
+	PodInformer  coreinformer.PodInformer
+	NodeInformer coreinformer.NodeInformer
+	FIPInformer  galaxyinformer.FloatingIPInformer
 
 	informerFactory    informers.SharedInformerFactory
 	crdInformerFactory crdInformer.SharedInformerFactory
@@ -66,6 +68,7 @@ func NewIPAMContext(client kubernetes.Interface, galaxyClient crd_clientset.Inte
 	}
 	ctx.informerFactory = informers.NewSharedInformerFactoryWithOptions(ctx.Client, time.Minute)
 	ctx.PodInformer = ctx.informerFactory.Core().V1().Pods()
+	ctx.NodeInformer = ctx.informerFactory.Core().V1().Nodes()
 	statefulsetInformer := ctx.informerFactory.Apps().V1().StatefulSets()
 	deploymentInformer := ctx.informerFactory.Apps().V1().Deployments()
 	ctx.crdInformerFactory = crdInformer.NewSharedInformerFactory(ctx.GalaxyClient, 0)
@@ -76,6 +79,7 @@ func NewIPAMContext(client kubernetes.Interface, galaxyClient crd_clientset.Inte
 	extensionInformer.Informer() // call Informer to actually create an informer
 
 	ctx.PodLister = ctx.PodInformer.Lister()
+	ctx.NodeLister = ctx.NodeInformer.Lister()
 	ctx.StatefulSetLister = statefulsetInformer.Lister()
 	ctx.DeploymentLister = deploymentInformer.Lister()
 	ctx.PoolLister = poolInformer.Lister()

pkg/ipam/floatingip/ipam_crd.go

@@ -338,7 +338,6 @@ func (ci *crdIpam) ConfigurePool(floatIPs []*FloatingIPPool) error {
 		glog.Infof("Configure pool done, %d fip pool, %d unallocated, %d allocated", len(ci.FloatingIPs),
 			len(ci.unallocatedFIPs), len(ci.allocatedFIPs))
 	}()
-
 	sort.Sort(FloatingIPSlice(floatIPs))
 	ips, err := ci.listFloatingIPs()
 	if err != nil {

pkg/ipam/schedulerplugin/ipam.go

@@ -161,7 +161,7 @@ func (p *FloatingIPPlugin) getNodeSubnetfromIPAM(node *corev1.Node) (*net.IPNet,
 		return nil, errors.New("FloatingIPPlugin:UnknowNode")
 	}
 	if ipNet := p.ipam.NodeSubnet(nodeIP); ipNet != nil {
-		glog.V(4).Infof("node %s %s %s", node.Name, nodeIP.String(), ipNet.String())
+		glog.V(4).Infof("node subnet cache %s %s %s", node.Name, nodeIP.String(), ipNet.String())
 		p.nodeSubnet[node.Name] = ipNet
 		return ipNet, nil
 	} else {

pkg/ipam/schedulerplugin/preempt.go

@@ -0,0 +1,59 @@
+package schedulerplugin
+
+import (
+	glog "k8s.io/klog"
+	"tkestack.io/galaxy/pkg/api/galaxy/constant"
+	"tkestack.io/galaxy/pkg/api/k8s/schedulerapi"
+)
+
+func fillNodeNameToMetaVictims(args *schedulerapi.ExtenderPreemptionArgs) {
+	if len(args.NodeNameToVictims) != 0 && len(args.NodeNameToMetaVictims) == 0 {
+		args.NodeNameToMetaVictims = map[string]*schedulerapi.MetaVictims{}
+		for node, victim := range args.NodeNameToVictims {
+			metaVictim := &schedulerapi.MetaVictims{
+				Pods:             []*schedulerapi.MetaPod{},
+				NumPDBViolations: victim.NumPDBViolations,
+			}
+			for _, pod := range victim.Pods {
+				metaPod := &schedulerapi.MetaPod{
+					UID: string(pod.UID),
+				}
+				metaVictim.Pods = append(metaVictim.Pods, metaPod)
+			}
+			args.NodeNameToMetaVictims[node] = metaVictim
+		}
+	}
+}
+
+func (p *FloatingIPPlugin) Preempt(args *schedulerapi.ExtenderPreemptionArgs) map[string]*schedulerapi.MetaVictims {
+	fillNodeNameToMetaVictims(args)
+	policy := parseReleasePolicy(&args.Pod.ObjectMeta)
+	if policy == constant.ReleasePolicyPodDelete {
+		return args.NodeNameToMetaVictims
+	}
+	subnetSet, err := p.getSubnet(args.Pod)
+	if err != nil {
+		glog.Errorf("unable to get pod subnets: %v", err)
+		return args.NodeNameToMetaVictims
+	}
+	glog.V(4).Infof("subnet for pod %v is %v", args.Pod.Name, subnetSet)
+	for nodeName := range args.NodeNameToMetaVictims {
+		node, err := p.NodeLister.Get(nodeName)
+		if err != nil {
+			glog.Errorf("unable to list node: %v", err)
+			delete(args.NodeNameToMetaVictims, nodeName)
+			continue
+		}
+		subnet, err := p.getNodeSubnet(node)
+		if err != nil {
+			glog.Errorf("unable to get node %v subnet: %v", nodeName, err)
+			delete(args.NodeNameToMetaVictims, nodeName)
+			continue
+		}
+		if !subnetSet.Has(subnet.String()) {
+			glog.V(4).Infof("remove node %v with subnet %v from victim", node.Name, subnet.String())
+			delete(args.NodeNameToMetaVictims, nodeName)
+		}
+	}
+	return args.NodeNameToMetaVictims
+}

pkg/ipam/schedulerplugin/preempt_test.go

@@ -0,0 +1,27 @@
+package schedulerplugin
+
+import (
+	"gotest.tools/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"testing"
+	"tkestack.io/galaxy/pkg/api/k8s/schedulerapi"
+)
+
+func TestFillNodeNameToMetaVictims(t *testing.T) {
+	args := &schedulerapi.ExtenderPreemptionArgs{
+		NodeNameToVictims: map[string]*schedulerapi.Victims{
+			"nod1": {
+				Pods: []*corev1.Pod{
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							UID: "id1",
+						},
+					},
+				},
+			},
+		},
+	}
+	fillNodeNameToMetaVictims(args)
+	assert.Equal(t, len(args.NodeNameToMetaVictims), 1)
+}

pkg/ipam/server/server.go

@@ -220,6 +220,7 @@ func (s *Server) startServer() {
 		Writes(schedulerapi.HostPriorityList{}))
 	ws.Route(ws.POST("/bind").To(s.bind).Reads(schedulerapi.ExtenderBindingArgs{}).
 		Writes(schedulerapi.ExtenderBindingResult{}))
+	ws.Route(ws.POST("/preempt").To(s.preempt).Reads(schedulerapi.ExtenderPreemptionArgs{}))
 	health := new(restful.WebService)
 	health.Route(health.GET("/healthy").To(s.healthy))
 	container := restful.NewContainer()
@@ -385,6 +386,26 @@ func (s *Server) bind(request *restful.Request, response *restful.Response) {
 	_ = response.WriteEntity(result)
 }
 
+func (s *Server) preempt(request *restful.Request, response *restful.Response) {
+	args := new(schedulerapi.ExtenderPreemptionArgs)
+	if err := request.ReadEntity(&args); err != nil {
+		glog.Error(err)
+		_ = response.WriteError(http.StatusInternalServerError, err)
+		return
+	}
+	glog.V(5).Infof("POST preempt %v+", *args)
+	start := time.Now()
+	glog.V(3).Infof("preempt for pod %v/%v with %v/%v victims start at %d+",
+		args.Pod.Namespace, args.Pod.Name, len(args.NodeNameToMetaVictims), len(args.NodeNameToVictims), start.Nanosecond())
+	nodeNameToMetaVictims := s.plugin.Preempt(args)
+	glog.V(3).Infof("preempt for pod %v/%v with %v/%v victims start at %d-",
+		args.Pod.Namespace, args.Pod.Name, len(args.NodeNameToMetaVictims), len(args.NodeNameToVictims), start.Nanosecond())
+	glog.V(5).Infof("POST preempt %v-", *args)
+	_ = response.WriteEntity(schedulerapi.ExtenderPreemptionResult{
+		NodeNameToMetaVictims: nodeNameToMetaVictims,
+	})
+}
+
 func (s *Server) healthy(request *restful.Request, response *restful.Response) {
 	response.WriteHeader(http.StatusOK)
 	_, _ = response.Write([]byte("ok"))

yaml/scheduler-policy.yaml

@@ -17,6 +17,7 @@ data:
           "httpTimeout": 70000000000,
           "filterVerb": "filter",
           "BindVerb": "bind",
+          "preemptVerb": "preempt",
           "weight": 1,
           "enableHttps": false,
           "managedResources": [
@@ -27,4 +28,4 @@ data:
           ]
         }
       ]
-    }
+    }