本文將講述如何使用scheduler-framework擴展原生調度器
目的： 在prefilter階段檢查pod是否添加有dely註釋，如果未達到對應時間則不調度
分析需要實現的method
註冊插件
WithPlugin返回一個註冊選項，由此我們可以看出，我們的插件需要實現framework.PluginFactory 接口
<code>func WithPlugin(name string, factory framework.PluginFactory) Option {
return func(registry framework.Registry) error {
return registry.Register(name, factory)
}
}
/<code>該接口接收傳入的附加參數和FrameworkHandle，關於FrameworkHandle的作用請查看之前文章
<code>type PluginFactory = func(configuration *runtime.Unknown, f FrameworkHandle) (Plugin, error)
/<code>
PreFilterPlugin接口
<code>// PreFilterPlugin is an interface that must be implemented by "prefilter" plugins.
// These plugins are called at the beginning of the scheduling cycle.
type PreFilterPlugin interface {
Plugin
// PreFilter is called at the beginning of the scheduling cycle. All PreFilter
// plugins must return success or the pod will be rejected.
PreFilter(ctx context.Context, state *CycleState, p *v1.Pod) *Status
// PreFilterExtensions returns a PreFilterExtensions interface if the plugin implements one,
// or nil if it does not. A Pre-filter plugin can provide extensions to incrementally
// modify its pre-processed info. The framework guarantees that the extensions
// AddPod/RemovePod will only be called after PreFilter, possibly on a cloned
// CycleState, and may call those functions more than once before calling
// Filter again on a specific node.
PreFilterExtensions() PreFilterExtensions
}
// Plugin is the parent type for all the scheduling framework plugins.
type Plugin interface {
Name() string
}
/<code>
PreFilterExtensions()方法返回PreFilterExtensions接口
<code>type PreFilterExtensions interface {
// AddPod is called by the framework while trying to evaluate the impact
// of adding podToAdd to the node while scheduling podToSchedule.
AddPod(ctx context.Context, state *CycleState, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status
// RemovePod is called by the framework while trying to evaluate the impact
// of removing podToRemove from the node while scheduling podToSchedule.
RemovePod(ctx context.Context, state *CycleState, podToSchedule *v1.Pod, podToRemove *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status
}
/<code>我們需要實現五個method:
- Name 返回插件名稱
- PreFilter 對pod進行篩選
- PreFilterExtensions prefilter擴展功能，評估add/removepod的影響，如果不實現可返回nil
- AddPod 評估添加pod到node的影響
- RemovePod 評估刪除pod到node的影響
代碼實現
實現註冊相關
<code>const Name = "test"
var _ framework.PreFilterPlugin = &TestPlugin{}
type Args struct {
KubeConfig string `json:"kubeconfig,omitempty"`
Master string `json:"master,omitempty"`
}
type TestPlugin struct {
handle framework.FrameworkHandle
Args *Args
}
func New(rargs *runtime.Unknown, handle framework.FrameworkHandle) (framework.Plugin, error) {
args := &Args{}
if err := framework.DecodeInto(rargs, args); err != nil {
return nil, err
}
klog.Info(args)
return &TestPlugin{
handle: handle,
Args: args,
}, nil
}
/<code>
實現prefilter接口
<code># 返回名稱，任何plugin都需要實現
func (self *TestPlugin) Name() string {
return Name
}
# 實現PreFilter method
func (self *TestPlugin) PreFilter(ctx context.Context, state *framework.CycleState, p *v1.Pod) *framework.Status {
klog.Error("into controller test")
state.Write()
var dtime int64
var err error
# 判斷是否有延遲字段
if v, ok := p.Annotations["delay"]; ok {
if dtime, err = strconv.ParseInt(v, 10, 64); err != nil {
return nil
}
# 距離當前大於延時間，則調度
if time.Now().Unix()-p.CreationTimestamp.Unix() >= dtime {
klog.Infof("scheduler: %s/%s", p.Namespace, p.Name)
return nil
}
# 否則不調度
klog.Infof("not reatch scheduler time: %s/%s", p.Namespace, p.Name)
return framework.NewStatus(framework.Skip, "not reatch scheduler time")
}
return nil
}
# PreFilterExtensions AddPod method,返回nil即success
func (self *TestPlugin) AddPod(ctx context.Context, state *framework.CycleState, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *framework.Status {
return nil
}
# PreFilterExtensions RemovePod,返回nil即success
func (self *TestPlugin) RemovePod(ctx context.Context, state *framework.CycleState, podToSchedule *v1.Pod, podToRemove *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *framework.Status {
return nil
}
# 這裡也可以返回nil
func (self *TestPlugin) PreFilterExtensions() framework.PreFilterExtensions {
return self
}
/<code>
測試
- 測試時使用kubeadm，停止原有kubelet
<code># 將static pod配置移走，kubelet會自動停止
mv /etc/kubernetes/manifests/kube-scheduler.yaml ./
/<code>
- 編譯運行
<code>go build
./test-scheduler-framework --config=config.yaml
/<code>
- 測試pod
創建deploy 配置如下 apiVersion: apps/v1 kind: Deployment metadata: labels: run: busybox name: busybox namespace: default spec: replicas: 0 selector: matchLabels: run: busybox template: metadata: annotations: delay: "15" labels: run: busybox spec: containers: - args: - sleep - "123456" image: busybox name: busybox - 擴容觀察pod狀態
<code>$ kubectl scale deploy busybox --replicas=1
$ kubectl get pods -l run=busybox -w
NAME READY STATUS RESTARTS AGE
busybox-8d8554fc8-f9899 0/1 Pending 0 3s
busybox-8d8554fc8-f9899 0/1 Pending 0 85s
busybox-8d8554fc8-f9899 0/1 ContainerCreating 0 85s
busybox-8d8554fc8-f9899 0/1 ContainerCreating 0 86s
busybox-8d8554fc8-f9899 1/1 Running 0 90s
/<code>
- 觀察調度器日誌
<code>E1226 18:19:11.071899 92593 type.go:59] into controller test
I1226 18:19:11.071909 92593 type.go:70] not reatch scheduler time: default/busybox-8d8554fc8-f9899
E1226 18:19:11.071923 92593 framework.go:287] error while running "test" prefilter plugin for pod "busybox-8d8554fc8-f9899": not reatch scheduler time
E1226 18:19:11.071941 92593 factory.go:469] Error scheduling default/busybox-8d8554fc8-f9899: error while running "test" prefilter plugin for pod "busybox-8d8554fc8-f9899": not reatch scheduler time; retrying
E1226 18:19:11.071970 92593 scheduler.go:638] error selecting node for pod: error while running "test" prefilter plugin for pod "busybox-8d8554fc8-f9899": not reatch scheduler time
E1226 18:20:36.064233 92593 type.go:59] into controller test
I1226 18:20:36.064258 92593 type.go:67] scheduler: default/busybox-8d8554fc8-f9899
/<code>
可以看到確實延時調度了，但是因為重新調度本身有時間間隔(30s)，所以並不是我們設置的值
分享到:
閱讀更多 有點技術 的文章
