kubernetes自定义调度扩展

$ kubebuilder version
Version: version.Version{KubeBuilderVersion:"1.0.7", KubernetesVendor:"1.11", GitCommit:"63bd3604767ddb5042fe76b67d097840a7a282c2", BuildDate:"2018-12-20T18:41:54Z", GoOs:"unknown", GoArch:"unknown"}

curl -sL https://github.com/kubernetes-sigs/kubebuilder/releases/download/v1.0.7/kubebuilder_1.0.7_linux_amd64.tar.gz -o kubebuilder_1.0.7_linux_amd64.tar.gz
tar -xzf kubebuilder_1.0.7_linux_amd64.tar.gz 
mv kubebuilder_1.0.7_linux_amd64 /usr/local/
export PATH=$PATH:/usr/local/kubebuilder_1.0.7_linux_amd64/bin

cd $GOPATH/src
mkdir ip-scheduler && cd ip-scheduler
#初始化项目，大概耗时耗时4-5minute
kubebuilder init --domain blueking.io --license apache2 --owner "The BlueKing Authors"
Run `dep ensure` to fetch dependencies (Recommended) [y/n]? y
#dep ensure
#Running make...
#make
#go generate ./pkg/... ./cmd/...
#go fmt ./pkg/... ./cmd/...
#go vet ./pkg/... ./cmd/...
#go run vendor/sigs.k8s.io/controller-tools/cmd/controller-gen/main.go all
#> CRD manifests generated under '/Users/Workspace/go/src/ip-scheduler/config/crds'
#> RBAC manifests generated under '/Users/Workspace/go/src/ip-scheduler/config/rbac'
#go test ./pkg/... ./cmd/... -coverprofile cover.out
#?       ip-scheduler/pkg/apis   [no test files]
#?       ip-scheduler/pkg/controller     [no test files]
#?       ip-scheduler/pkg/webhook        [no test files]
#?       ip-scheduler/cmd/manager        [no test files]
#go build -o bin/manager ip-scheduler/cmd/manager
mkdir -p cmd/scheduler

touch cmd/scheduler/main.go
touch cmd/scheduler/scheduler.go

├── Dockerfile
├── Gopkg.lock
├── Gopkg.toml
├── Makefile
├── PROJECT
├── bin
│   └── manager
├── cmd
│   ├── manager
│   └── scheduler
├── config
│   ├── crds
│   ├── default
│   ├── manager
│   └── rbac
├── cover.out
├── hack
│   └── boilerplate.go.txt
├── pkg
│   ├── apis
│   ├── controller
│   └── webhook
└── vendor
    ├── cloud.google.com
    ├── github.com
    ├── go.uber.org
    ├── golang.org
    ├── google.golang.org
    ├── gopkg.in
    ├── k8s.io
    └── sigs.k8s.io

/*
Copyright 2019 The BlueKing Authors.

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,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package main

import (
	"flag"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/go-logr/logr"
	"k8s.io/client-go/informers"
	kubeClientSet "k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
)

var (
	log     logr.Logger
	cfgFile string
	ipNum   int
)

func init() {
	logf.SetLogger(logf.ZapLogger(false))
	log = logf.Log.WithName("ip-scheduler")
	flag.StringVar(&cfgFile, "config", "./kube.yaml", "ip-scheduler kubernetes configuration")
	flag.IntVar(&ipNum, "ip-resource", 10, "ip resource for every node, only for demonstration")
}

func main() {
	flag.Parse()
	//building clientset configuration
	restConfig, err := clientcmd.BuildConfigFromFlags("", cfgFile)
	if err != nil {
		log.Error(err, "blueking ip-scheduler builds clientConfig with kubeconfig ./kube.yaml")
		os.Exit(1)
	}
	//process kubernetes clientSet for kubeInformerFactory
	clientset, err := kubeClientSet.NewForConfig(restConfig)
	if err != nil {
		log.Error(err, "bmsf-networkpolicy-controller create kubernetes clientset with restConfig failed")
		os.Exit(1)
	}
	//create SharedInformerFactory with ClientSet
	factories := informers.NewSharedInformerFactory(clientset, time.Second*180)
	s := newscheduler(clientset.CoreV1(), factories.Core().V1().Pods(), factories.Core().V1().Nodes())
	handleSignal(s)
	if err := s.run(factories); err != nil {
		log.Error(err, "ip-scheduler force to exit")
		os.Exit(1)
	}
	//wait for exit signal
	s.wait()
}

// HandleSignal handle system signal for exit
func handleSignal(s *scheduler) {
	signalChan := make(chan os.Signal, 5)
	signal.Notify(signalChan, syscall.SIGTRAP, syscall.SIGINT, syscall.SIGKILL, syscall.SIGTERM)
	go func() {
		select {
		case sig := <-signalChan:
			log.V(0).Info("blueking ip-scheduler was killed.", "signal", sig.String())
			s.stop()
			time.Sleep(time.Second * 3)
		}
	}()
}

/*
Copyright 2019 The BlueKing Authors.

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,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package main

import (
	"fmt"
	"sync"
	"time"

	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/informers"
	corev1 "k8s.io/client-go/informers/core/v1"
	coreclient "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
)

const (
	schedulerName = "blueking-ip-scheduler"
)

//newscheduler initialize custom scheduler for IP resource demonstration
func newscheduler(pc coreclient.CoreV1Interface, pods corev1.PodInformer, nodes corev1.NodeInformer) *scheduler {
	s := &scheduler{
		stopCh:       make(chan struct{}),
		podClient:    pc,
		podInformer:  pods,
		nodeInformer: nodes,
		queue:        workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
		ipResource:   make(map[string]int),
	}
	//register Pod event callback to Informer
	s.podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: s.onPodAdd,
		//UpdateFunc: s.onPodUpdate,
		DeleteFunc: s.onPodDelete,
	})
	s.nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: s.onNodeAdd,
		//UpdateFunc: s.onNodeUpdate,
		DeleteFunc: s.onNodeDelete,
	})
	return s
}

type scheduler struct {
	stopCh chan struct{}
	//client for Pod Operation
	podClient coreclient.CoreV1Interface
	//informers for event watching
	nodeInformer corev1.NodeInformer
	podInformer  corev1.PodInformer
	//queue for retry
	queue       workqueue.RateLimitingInterface
	retryLocker sync.Mutex
	//ipResource cachew
	ipResource map[string]int
	ipLocker   sync.Mutex
}

func (s *scheduler) run(factory informers.SharedInformerFactory) error {
	factory.Start(s.stopCh)
	kubeFlags := factory.WaitForCacheSync(s.stopCh)
	for t, state := range kubeFlags {
		if !state {
			return fmt.Errorf("%s informer cache init failure", t.Name())
		}
	}
	go wait.Until(s.handleQueue, time.Millisecond*100, s.stopCh)
	return nil
}

//handleQueue when ip resource limited, push to
// queue and retry again
func (s *scheduler) handleQueue() {
	keyObj, shutdown := s.queue.Get()
	if shutdown {
		return
	}
	s.retryLocker.Lock()
	defer s.retryLocker.Unlock()
	key := keyObj.(string)
	//check local cache
	obj, exist, err := s.podInformer.Informer().GetIndexer().GetByKey(key)
	if err != nil {
		log.Error(err, "retry Pod failed, get from local cache failed.", "podID", key)
		return
	}
	if !exist {
		//Pod deleted, no scheduling needed
		s.queue.Forget(keyObj)
		s.queue.Done(keyObj)
		return
	}
	pod := obj.(*v1.Pod)
	defer s.queue.Done(keyObj)
	if err := s.schedule(pod); err != nil {
		log.Error(err, "scheduler re-process Pod stil failed.", "podId", key)
		//check reschedule limit
		retry := s.queue.NumRequeues(keyObj)
		if retry < 100 {
			s.queue.AddRateLimited(keyObj)
			log.Info("Pod retry is stil effective, push back to queue again", "podID", key, "retryCnt", retry)
		} else {
			log.Info("Pod retry is beyond limit system setting, forget scheduling", "podID", key, "retryCnt", retry)
			s.queue.Forget(keyObj)
		}
	}
}

func (s *scheduler) wait() {
	<-s.stopCh
}

func (s *scheduler) stop() {
	close(s.stopCh)
	s.queue.ShutDown()
}

func (s *scheduler) onPodAdd(obj interface{}) {
	pod, ok := obj.(*v1.Pod)
	if !ok {
		return
	}
	key, err := cache.MetaNamespaceKeyFunc(obj)
	if err != nil {
		return
	}
	// Only schedule pending pods, also ignore abnormal pod
	if !(pod.Spec.NodeName == "" && pod.Spec.SchedulerName == schedulerName &&
		pod.Status.Phase != v1.PodSucceeded && pod.Status.Phase != v1.PodFailed) {
		log.V(0).Info("Pod is not under ip-scheduler control", "podId", key)
		return
	}
	log.Info("ip-scheduler Ready to schedule pod first onPodAddEvent", "podId", key)
	if err := s.schedule(pod); err != nil {
		log.Error(err, "ip-scheduler first schedule pod failed, push to retry queue to recover", "podId", key)
		//push to reschedule
		key, _ := cache.MetaNamespaceKeyFunc(pod)
		s.queue.Add(key)
	}
}

func (s *scheduler) onPodUpdate(old, cur interface{}) {

}

func (s *scheduler) onPodDelete(obj interface{}) {
	pod, ok := obj.(*v1.Pod)
	if !ok {
		return
	}
	podName := fmt.Sprintf("%s/%s", pod.GetNamespace(), pod.GetName())
	nodeName := pod.Spec.NodeName
	if len(nodeName) == 0 {
		log.Info("Pod have no Node scheduling, skip IP resource release", "podId", podName)
		return
	}
	s.retryLocker.Lock()
	defer s.retryLocker.Unlock()
	if pod.Spec.SchedulerName == schedulerName {
		//schedule management scope
		log.Info("Pod is under deletion, ip-scheduler release ip resource for this pod", "podId", podName, "nodeName", nodeName)
		s.releaseIPResource(nodeName)
		return
	}
	log.Info("Pod is not under ip-scheduler control", "podId", podName)
}

func (s *scheduler) onNodeAdd(obj interface{}) {
	node, ok := obj.(*v1.Node)
	if !ok {
		return
	}
	//check node role
	if _, isMaster := node.Labels["node-role.kubernetes.io/master"]; isMaster {
		log.Info("Node is master role, it's not under scheduling scope. skip", "nodeName", node.GetName())
		return
	}
	s.ipLocker.Lock()
	defer s.ipLocker.Unlock()
	key := node.GetName()
	if _, found := s.ipResource[key]; !found {
		log.Info("Node is adding, initialize IP resource...", "nodeName", key)
		s.ipResource[key] = ipNum
	}
}

func (s *scheduler) onNodeUpdate(old, cur interface{}) {

}

func (s *scheduler) onNodeDelete(obj interface{}) {
	node, ok := obj.(*v1.Node)
	if !ok {
		return
	}
	key := node.GetName()
	s.ipLocker.Lock()
	defer s.ipLocker.Unlock()
	if _, found := s.ipResource[key]; found {
		delete(s.ipResource, key)
		log.Info("Node is deleting, clean resource from local cache...", "nodeName", key)
	}
}

func (s *scheduler) schedule(pod *v1.Pod) error {
	nodeName := s.acquireIPResource()
	if len(nodeName) == 0 {
		log.Error(fmt.Errorf("ip-resource runs out"), "Pod can not be scheduled because of ip resource running out", "namespace", pod.GetNamespace(), "podName", pod.GetName())
		return fmt.Errorf("lack of resource")
	}
	//assign Pod to this Node
	binding := &v1.Binding{
		ObjectMeta: metav1.ObjectMeta{Namespace: pod.GetNamespace(), Name: pod.GetName(), UID: pod.UID},
		Target: v1.ObjectReference{
			Kind: "Node",
			Name: nodeName,
		},
	}
	log.Info("Attempting to bind pod to node", "namespace", pod.GetNamespace(), "podName", pod.GetName(), "nodeName", nodeName)
	if err := s.podClient.Pods(binding.Namespace).Bind(binding); err != nil {
		log.Error(err, "Pod is binding to Node failed", "namespace", pod.GetNamespace(), "podName", pod.GetName(), "nodeName", nodeName)
		s.releaseIPResource(nodeName)
		return fmt.Errorf("binding failed")
	}
	log.Info("Binding pod to node successfully", "namespace", pod.GetNamespace(), "podName", pod.GetName(), "nodeName", nodeName)
	return nil
}

//releaseIPResource release IP resource to local cache
func (s *scheduler) releaseIPResource(node string) {
	s.ipLocker.Lock()
	defer s.ipLocker.Unlock()
	ipNum, found := s.ipResource[node]
	if !found {
		log.Info("Node do not exist in local cache", "nodeName", node)
		return
	}
	s.ipResource[node] = ipNum + 1
	log.Info("Node IP resource release successfully", "nodeName", node)
}

//acquireIPResource get max IP resource node from local cache
func (s *scheduler) acquireIPResource() string {
	s.ipLocker.Lock()
	defer s.ipLocker.Unlock()
	//get max resource node
	ipNum := 0
	node := ""
	for name, ipResource := range s.ipResource {
		if ipResource > ipNum {
			node = name
			ipNum = ipResource
		}
	}
	if ipNum == 0 {
		return ""
	}
	s.ipResource[node] = ipNum - 1
	log.Info("Node IP resource acquire successfully", "nodeName", node)
	return node
}

# Build custom ip-scheduler binary
scheduler: generate fmt vet
	go build -o bin/bk-ip-scheduler ip-scheduler/cmd/scheduler

make scheduler
go generate ./pkg/... ./cmd/...
go fmt ./pkg/... ./cmd/...
go vet ./pkg/... ./cmd/...
go build -o bin/bk-ip-scheduler ip-scheduler/cmd/scheduler

apiVersion: v1
clusters:
- cluster:
    server: http://127.0.0.1:8080
  name: scheduler
contexts:
- context:
    cluster: scheduler
    user: ""
  name: scheduler
current-context: scheduler

./bk-ip-scheduler --config ./kube.yaml

apiVersion: apps/v1beta1
kind: Deployment
metadata:
  labels:
    app: blueking-demo
  name: bk-demo
  namespace: blueking
spec:
  replicas: 1
  strategy:
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 1
    type: RollingUpdate
  template:
    metadata:
      labels:
        app: blueking-demo
    spec:
      containers:
      - name: bk-demo
        command: ['python']
        args:
        - -m
        - SimpleHTTPServer
        - "8080"
        image: python:v2.7.10
        imagePullPolicy: IfNotPresent
        resources: {}
      dnsPolicy: ClusterFirst
      restartPolicy: Always
      schedulerName: blueking-ip-scheduler

kubectl create -f demo.yaml