Kubernetes 自定义调度器（Custom Scheduler）开发：满足特殊业务需求

Kubernetes 自定义调度器：为你的Pod量身定制“鹊桥” 🌉

各位观众老爷们，大家好！我是你们的老朋友，一个在 Kubernetes 的世界里摸爬滚打多年的老码农。今天，咱们要聊聊 Kubernetes 里的一个高级玩法——自定义调度器（Custom Scheduler）。

想象一下，Kubernetes 就像一个大型的后宫，里面住满了各种各样的 Pod，它们都等着被“皇上”（kube-scheduler）翻牌子，安排到合适的“寝宫”（Node）里去。但是，有时候皇上的眼光不咋地，总是把美女安排到偏远冷宫，把壮汉安排到绣花房，这可不行啊！这时候，就需要我们这些“红娘”出马，为这些 Pod 量身定制一个更靠谱的“鹊桥”，也就是自定义调度器。

为什么要自定义调度器？ 🤔

Kubernetes 自带的 kube-scheduler 已经很强大了，但它就像一个通用的“媒婆”，只能满足大多数情况的需求。对于一些特殊业务场景，我们需要更加精细的调度策略。以下是一些典型的应用场景：

资源感知调度： 假设你的应用需要大量的 GPU 资源，而 kube-scheduler 只知道 Node 的 CPU 和内存情况。你需要一个自定义调度器，能够感知 Node 的 GPU 资源，并将 Pod 调度到具有足够 GPU 资源的 Node 上。
拓扑感知调度： 你的应用对网络延迟非常敏感，需要将相关的 Pod 调度到同一机架或者同一可用区。kube-scheduler 无法感知 Node 的拓扑信息，你需要一个自定义调度器来实现拓扑感知调度。
亲和性/反亲和性调度： 你希望某些 Pod 尽可能地调度到同一 Node 上，以减少网络开销；或者希望某些 Pod 尽可能地分散到不同的 Node 上，以提高应用的可用性。虽然 kube-scheduler 提供了亲和性和反亲和性策略，但有时候你需要更加复杂的策略。
成本优化调度： 你希望将 Pod 调度到成本较低的 Node 上，例如空闲的预留实例或者竞价实例。kube-scheduler 无法感知 Node 的成本信息，你需要一个自定义调度器来实现成本优化调度。
自定义负载均衡： 你需要根据自定义的负载指标（例如请求延迟、错误率）来调度 Pod，而不是仅仅依赖 CPU 和内存的使用率。

总而言之，当 kube-scheduler 无法满足你的特殊业务需求时，自定义调度器就是你的救星！它可以让你更灵活地控制 Pod 的调度过程，优化资源利用率，提高应用的性能和可用性。

自定义调度器的原理：当红娘也需要资格证 📜

自定义调度器的本质就是一个独立的程序，它通过 Kubernetes API Server 监听 Pod 的创建事件，并根据自定义的调度策略来选择合适的 Node。它就像一个独立的“媒婆”，拥有自己的“客户”名单和“婚配”标准。

工作流程如下：

监听 Pod 创建事件： 自定义调度器通过 Kubernetes API Server 监听 Pod 的创建事件，筛选出需要自己调度的 Pod。通常，我们会通过 schedulerName 字段来指定 Pod 使用哪个调度器。
Node 筛选： 自定义调度器根据自定义的策略，从集群中选择合适的 Node。这个过程通常包括两个阶段：
- Predicate： 过滤掉不满足条件的 Node。例如，如果 Pod 需要 GPU 资源，则过滤掉没有 GPU 资源的 Node。
- Priority： 对满足条件的 Node 进行评分，选择得分最高的 Node。例如，可以根据 Node 的空闲资源、网络延迟等指标来评分。
绑定 Pod 到 Node： 自定义调度器通过 Kubernetes API Server 将 Pod 绑定到选定的 Node。这个过程就像“鹊桥”搭好之后，将牛郎织女送过去一样。

关键组件：

API Server： Kubernetes 的大脑，提供 API 接口，供自定义调度器监听事件和更新状态。
Informer： 用于缓存 Kubernetes 资源（例如 Pod、Node）的状态，提高调度器的性能。
Predicate 函数： 用于过滤不满足条件的 Node。
Priority 函数： 用于对满足条件的 Node 进行评分。
Binder： 用于将 Pod 绑定到选定的 Node。

重要概念：

SchedulerName： Pod 的一个字段，用于指定 Pod 使用哪个调度器。
Annotations： 用于在 Pod 上添加自定义的元数据，例如调度策略、资源需求等。
Taints 和 Tolerations： 用于控制 Pod 是否能够调度到具有特定 Taint 的 Node 上。

自定义调度器的实现：手把手教你搭“鹊桥” 🛠️

接下来，咱们来手把手地实现一个简单的自定义调度器。为了方便演示，我们使用 Go 语言来实现。

1. 项目初始化：

首先，创建一个新的 Go 项目，并引入 Kubernetes 相关的依赖：

mkdir custom-scheduler
cd custom-scheduler
go mod init custom-scheduler
go get k8s.io/[email protected]
go get k8s.io/[email protected]
go get k8s.io/[email protected]

2. 编写调度器代码：

创建一个 main.go 文件，并添加以下代码：

package main

import (
    "context"
    "flag"
    "fmt"
    "os"
    "time"

    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/util/wait"
    "k8s.io/client-go/informers"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/klog/v2"
)

var (
    kubeconfig = flag.String("kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
    schedulerName = flag.String("scheduler-name", "my-scheduler", "Name of the scheduler to filter pods.")
)

func main() {
    klog.InitFlags(nil)
    flag.Parse()

    // 1. 构建 Kubernetes 客户端
    config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
    if err != nil {
        klog.Fatalf("Error building kubeconfig: %s", err.Error())
    }
    kubeClient, err := kubernetes.NewForConfig(config)
    if err != nil {
        klog.Fatalf("Error building kubernetes clientset: %s", err.Error())
    }

    // 2. 创建 Informer 工厂
    factory := informers.NewSharedInformerFactory(kubeClient, 0)
    podInformer := factory.Core().V1().Pods()
    nodeInformer := factory.Core().V1().Nodes()

    // 3. 添加事件处理函数
    podInformer.Informer().AddEventHandler(&podEventHandler{
        kubeClient:    kubeClient,
        nodeLister:    nodeInformer.Lister(),
        schedulerName: *schedulerName,
    })

    // 4. 启动 Informer 工厂
    stop := make(chan struct{})
    defer close(stop)
    factory.Start(stop)
    factory.WaitForCacheSync(stop)

    // 5. 保持程序运行
    wait.Until(func() {
        klog.Info("Custom scheduler is running...")
    }, time.Second, stop)
}

type podEventHandler struct {
    kubeClient    kubernetes.Interface
    nodeLister    corev1.NodeLister
    schedulerName string
}

func (p *podEventHandler) AddFunc(obj interface{}) {
    pod := obj.(*corev1.Pod)
    if pod.Spec.SchedulerName != p.schedulerName {
        return
    }

    klog.Infof("Found a pod to schedule: %s/%s", pod.Namespace, pod.Name)

    // 1. 筛选 Node
    nodes, err := p.nodeLister.List(labels.Everything())
    if err != nil {
        klog.Errorf("Error listing nodes: %s", err.Error())
        return
    }

    var selectedNode *corev1.Node
    for _, node := range nodes {
        // 简单的 Predicate: 检查 Node 是否 Ready
        for _, condition := range node.Status.Conditions {
            if condition.Type == corev1.NodeReady && condition.Status == corev1.ConditionTrue {
                selectedNode = node
                break
            }
        }
        if selectedNode != nil {
            break // 找到一个 Ready 的 Node 就停止
        }
    }

    if selectedNode == nil {
        klog.Errorf("No ready node found for pod: %s/%s", pod.Namespace, pod.Name)
        return
    }

    // 2. 绑定 Pod 到 Node
    err = p.bindPodToNode(pod, selectedNode.Name)
    if err != nil {
        klog.Errorf("Error binding pod to node: %s", err.Error())
        return
    }

    klog.Infof("Successfully scheduled pod %s/%s to node %s", pod.Namespace, pod.Name, selectedNode.Name)
}

func (p *podEventHandler) UpdateFunc(oldObj, newObj interface{}) {}
func (p *podEventHandler) DeleteFunc(obj interface{}) {}

func (p *podEventHandler) bindPodToNode(pod *corev1.Pod, nodeName string) error {
    binding := &corev1.Binding{
        ObjectMeta: metav1.ObjectMeta{
            Name:      pod.Name,
            Namespace: pod.Namespace,
        },
        Target: corev1.ObjectReference{
            Kind: "Node",
            Name: nodeName,
        },
    }

    err := p.kubeClient.CoreV1().Pods(pod.Namespace).Bind(context.TODO(), binding, metav1.CreateOptions{})
    return err
}

代码解释：

构建 Kubernetes 客户端： 使用 clientcmd 和 kubernetes 包构建 Kubernetes 客户端，用于与 Kubernetes API Server 通信。
创建 Informer 工厂： 使用 informers 包创建 Informer 工厂，用于监听 Pod 和 Node 的创建事件。
添加事件处理函数： 创建一个 podEventHandler 结构体，并实现 AddFunc、UpdateFunc、DeleteFunc 方法，用于处理 Pod 的创建、更新和删除事件。
- 在 AddFunc 方法中，我们首先判断 Pod 的 schedulerName 是否与我们自定义的调度器名称一致，如果一致，则说明该 Pod 需要由我们来调度。
- 然后，我们通过 nodeLister 获取所有 Node 的列表，并遍历这些 Node，找到一个 Ready 状态的 Node。
- 最后，我们使用 bindPodToNode 方法将 Pod 绑定到选定的 Node。
启动 Informer 工厂： 启动 Informer 工厂，开始监听 Pod 和 Node 的创建事件。
保持程序运行： 使用 wait.Until 函数保持程序运行，直到收到停止信号。

3. 编译和运行：

go build -o custom-scheduler
./custom-scheduler --kubeconfig=/path/to/your/kubeconfig --scheduler-name=my-scheduler

4. 创建 Pod：

创建一个 Pod 的 YAML 文件，并指定 schedulerName 为 my-scheduler：

apiVersion: v1
kind: Pod
metadata:
  name: my-pod
spec:
  schedulerName: my-scheduler
  containers:
  - name: nginx
    image: nginx

5. 验证：

使用 kubectl apply -f pod.yaml 命令创建 Pod。然后，使用 kubectl describe pod my-pod 命令查看 Pod 的状态，可以看到 Pod 已经被调度到某个 Node 上，并且 schedulerName 为 my-scheduler。

恭喜你！你已经成功地实现了一个简单的自定义调度器！🎉

高级技巧：让你的“鹊桥”更智能 🧠

上面的例子只是一个非常简单的自定义调度器，它只是简单地将 Pod 调度到第一个 Ready 状态的 Node 上。在实际应用中，我们需要更加复杂的调度策略。

1. Predicate 函数：

Predicate 函数用于过滤掉不满足条件的 Node。例如，我们可以编写一个 Predicate 函数来检查 Node 是否具有足够的 CPU 和内存资源：

func checkNodeResources(node *corev1.Node, pod *corev1.Pod) bool {
    cpuRequest := pod.Spec.Containers[0].Resources.Requests.Cpu()
    memoryRequest := pod.Spec.Containers[0].Resources.Requests.Memory()

    cpuAllocatable := node.Status.Allocatable.Cpu()
    memoryAllocatable := node.Status.Allocatable.Memory()

    if cpuAllocatable.Cmp(cpuRequest) < 0 || memoryAllocatable.Cmp(memoryRequest) < 0 {
        return false
    }

    return true
}

2. Priority 函数：

Priority 函数用于对满足条件的 Node 进行评分，选择得分最高的 Node。例如，我们可以编写一个 Priority 函数来根据 Node 的空闲资源进行评分：

func calculateNodeScore(node *corev1.Node) int {
    cpuAllocatable := node.Status.Allocatable.Cpu().MilliValue()
    memoryAllocatable := node.Status.Allocatable.Memory().MilliValue()

    return int(cpuAllocatable + memoryAllocatable)
}

3. 使用 Taints 和 Tolerations：

Taints 和 Tolerations 可以用于控制 Pod 是否能够调度到具有特定 Taint 的 Node 上。例如，我们可以给某些 Node 添加一个 Taint，表示这些 Node 专门用于运行特定的应用。然后，我们可以在 Pod 上添加一个 Toleration，表示该 Pod 可以容忍这个 Taint。

4. 集成外部数据源：

自定义调度器可以集成外部数据源，例如数据库、监控系统等，来获取更多的信息，从而做出更智能的调度决策。例如，我们可以从数据库中获取 Node 的成本信息，然后根据成本信息来调度 Pod。

5. 使用 Kubernetes 调度框架：

Kubernetes 提供了调度框架，可以让你更方便地开发自定义调度器。调度框架提供了一系列的扩展点，例如 PreFilter、Filter、PreScore、Score、Bind 等，你可以通过实现这些扩展点来实现自定义的调度逻辑。

注意事项：避免“鹊桥”变成“奈何桥” 💀

自定义调度器虽然强大，但也需要谨慎使用，否则可能会导致一些问题：

性能问题： 自定义调度器需要监听 Kubernetes API Server 的事件，并进行复杂的计算，可能会对 API Server 的性能产生影响。因此，需要对调度器进行性能优化，例如使用 Informer 缓存数据，减少 API Server 的访问次数。
一致性问题： 自定义调度器需要与 kube-scheduler 协同工作，可能会出现一致性问题。例如，kube-scheduler 可能会将 Pod 调度到自定义调度器已经选择的 Node 上。因此，需要仔细设计调度策略，避免冲突。
复杂性问题： 自定义调度器需要编写大量的代码，并且需要对 Kubernetes 的内部机制有深入的了解，开发难度较高。

总结：

自定义调度器是 Kubernetes 的一个高级特性，可以让你更灵活地控制 Pod 的调度过程，优化资源利用率，提高应用的性能和可用性。但是，自定义调度器也需要谨慎使用，需要仔细设计调度策略，避免出现性能问题、一致性问题和复杂性问题。

希望这篇文章能够帮助你了解 Kubernetes 自定义调度器，并在实际应用中发挥它的威力！记住，好的“红娘”能让“有情人终成眷属”，而好的自定义调度器，能让你的 Pod 找到最合适的“家”！ 😉