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。它就像一个独立的“媒婆”，拥有自己的“客户”名单和“婚配”标准。

工作流程如下：

1. 监听 Pod 创建事件： 自定义调度器通过 Kubernetes API Server 监听 Pod 的创建事件，筛选出需要自己调度的 Pod。通常，我们会通过 schedulerName 字段来指定 Pod 使用哪个调度器。
2. Node 筛选： 自定义调度器根据自定义的策略，从集群中选择合适的 Node。这个过程通常包括两个阶段：
   • Predicate： 过滤掉不满足条件的 Node。例如，如果 Pod 需要 GPU 资源，则过滤掉没有 GPU 资源的 Node。
   • Priority： 对满足条件的 Node 进行评分，选择得分最高的 Node。例如，可以根据 Node 的空闲资源、网络延迟等指标来评分。
3. 绑定 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 找到最合适的“家”！ 😉