Koord-Queue

介绍

Koord-Queue 是 Koordinator 生态系统的原生 Kubernetes 作业队列管理系统。它管理多个队列间的作业准入和排序，与 Koordinator 的 ElasticQuota 深度集成，实现资源公平性和多租户隔离。主要功能包括：

• 多队列管理，支持 Priority、Block 和 Intelligent 排队策略。
• 深度 ElasticQuota 集成，避免重复配额配置，实现弹性资源共享。
• 预调度，通过在作业创建 Pod 之前排队来减少调度器压力。
• 多框架支持，包括 TFJob、PyTorchJob、Spark、Argo Workflow、Ray 和原生 Kubernetes Job。
• 准入检查框架，兼容 Kueue 的 AdmissionCheck API。

配置

前置要求

• Kubernetes >= 1.22
• Koordinator >= 1.5（用于 ElasticQuota 集成）

安装

使用 Helm 安装 Koord-Queue：

1. # 方式一：从 Helm 仓库安装
2. helm repo add koordinator-sh https://koordinator-sh.github.io/charts/
3. helm install koord-queue koordinator-sh/koord-queue --version 1.8.0 \
4. --namespace koord-queue \
5. --create-namespace

验证安装：

1. $ kubectl get deployment -n koord-queue
2. NAME                       READY   UP-TO-DATE   AVAILABLE   AGE
3. koord-queue-controllers    1/1     1            1           30s
4. koord-queue                1/1     1            1           30s

6. $ kubectl get crd | grep scheduling.x-k8s.io
7. queues.scheduling.x-k8s.io          2024-01-01T00:00:00Z
8. queueunits.scheduling.x-k8s.io      2024-01-01T00:00:00Z

配置项

Koord-Queue 默认使用 ElasticQuotaV2 模式。

默认配置

1. # 镜像仓库（默认：阿里云北京）
2. global:
3. imagePrefix: registry.cn-beijing.aliyuncs.com

5. controller:
6. image:
7. repository: koordinator-sh/koord-queue
8. tag: v1.8.0

10. extension:
11. koord-queue-controllers:
12. repository: koordinator-sh/koord-queue-controllers
13. tag: v1.8.0
14. batchjob:
15. enable: true    # 原生 Kubernetes Job 支持
16. tf:
17. enable: false
18. pytorch:
19. enable: false
20. argo:
21. enable: false
22. spark:
23. enable: false
24. ray:
25. enable: false
26. mpi:
27. enable: false

29. pluginConfigs:
30. apiVersion: scheduling.k8s.io/v1
31. kind: KoordQueueConfiguration
32. plugins:
33. - name: Priority
34. - name: ElasticQuotaV2

ElasticQuotaV2 模式（默认）

使用独立的 ElasticQuota CR（scheduling.sigs.k8s.io/v1alpha1）。这是 Koordinator 用户的推荐模式。

使用 Koord-Queue

使用 ElasticQuota 快速开始

此示例使用 Koordinator 的 ElasticQuota 进行弹性资源管理。

创建 ElasticQuota

1. apiVersion: scheduling.sigs.k8s.io/v1alpha1
2. kind: ElasticQuota
3. metadata:
4. name: team-a
5. namespace: default
6. labels:
7. quota.scheduling.koordinator.sh/parent: ""
8. quota.scheduling.koordinator.sh/is-parent: "false"
9. spec:
10. max:
11. cpu: "4"
12. memory: 8Gi
13. min:
14. cpu: "4"
15. memory: 8Gi

1. $ kubectl apply -f elastic-quota.yaml

Queue 自动创建

使用 ElasticQuotaV2 时，插件会为每个 ElasticQuota 资源自动创建 Queue CR，位于 koord-queue 命名空间中。自动创建的 Queue 与 ElasticQuota 同名（例如 team-a），默认 priority: 1000、queuePolicy: Priority。无需为每个 ElasticQuota 手动创建 Queue。

如需自定义 Queue 策略，可在 ElasticQuota 上设置 koord-queue/queue-policy 标签：

1. metadata:
2. labels:
3. koord-queue/queue-policy: Priority  # 选项：Priority、Block、Intelligent

提交作业并验证排队行为

Koord-Queue 的 Job Extensions 会自动为提交的作业创建 QueueUnit 资源。要提交一个由 Koord-Queue 管理的 Kubernetes Job，需设置 spec.suspend: true 并添加配额标签。将以下双文档 YAML 保存为 jobs.yaml 并一次性提交：

1. apiVersion: batch/v1
2. kind: Job
3. metadata:
4. name: my-job
5. namespace: default
6. labels:
7. quota.scheduling.koordinator.sh/name: team-a
8. spec:
9. suspend: true
10. template:
11. spec:
12. containers:
13. - name: test
14. image: busybox:stable
15. command: ["/bin/sh", "-c", "sleep 30"]
16. resources:
17. requests:
18. cpu: "4"
19. memory: 8Gi
20. limits:
21. cpu: "4"
22. memory: 8Gi
23. restartPolicy: Never
24. ---
25. apiVersion: batch/v1
26. kind: Job
27. metadata:
28. name: my-job-blocked
29. namespace: default
30. labels:
31. quota.scheduling.koordinator.sh/name: team-a
32. spec:
33. suspend: true
34. template:
35. spec:
36. containers:
37. - name: test
38. image: busybox:stable
39. command: ["/bin/sh", "-c", "sleep 30"]
40. resources:
41. requests:
42. cpu: "4"
43. memory: 8Gi
44. limits:
45. cpu: "4"
46. memory: 8Gi
47. restartPolicy: Never

1. $ kubectl apply -f jobs.yaml

team-a 配额的 max.cpu 为 "4"、max.memory 为 "8Gi"，正好供一个作业使用。ElasticQuotaV2 插件基于 Pod 实际运行时的资源消耗来统计配额用量。当 my-job 的 Pod 进入 Running 状态并占满配额后，my-job-blocked 将被阻塞在队列中：

1. # 先等待 my-job 的 Pod 进入 Running 状态
2. $ kubectl wait --for=condition=Ready pod -l job-name=my-job -n default --timeout=120s

4. $ kubectl get queueunit my-job-blocked -n default
5. NAME             PHASE   PRIORITY   ADMISSIONS   JOBTYPE
6. my-job-blocked                                   Job

QueueUnit 保持 Enqueued 状态，因为 team-a 已达到 max 配额上限。当 my-job 完成并释放资源后，my-job-blocked 将自动出队执行。

不同作业类型的暂停方式

不同类型的作业使用不同的字段进行暂停：

Argo Workflow 示例：

对于 Argo Workflow，Koord-Queue 使用名为 koord-queue-suspend 的特殊暂停模板。工作流必须满足以下条件才能被队列管理：

1. 包含名为 koord-queue-suspend 的模板，且有 suspend 字段
2. 工作流有处于 Running 状态的 suspend 节点，或者 spec.suspend 设置为 true

1. apiVersion: argoproj.io/v1alpha1
2. kind: Workflow
3. metadata:
4. name: my-workflow
5. annotations:
6. koord-queue/min-resources: |
7. cpu: 5
8. memory: 5Gi
9. spec:
10. suspend: true
11. templates:
12. # 添加此暂停模板用于队列管理
13. - name: koord-queue-suspend
14. suspend: {}
15. # 你的实际工作流模板
16. - name: main
17. container:
18. image: python:3.9
19. command: [python, -c, "print('Hello from workflow')"]
20. entrypoint: main

工作原理：

当提交工作流时，Koord-Queue 通过以下方式检查是否应该管理它：

• 扫描所有模板，查找带有 suspend 字段的 koord-queue-suspend 模板
• 检查是否有任何工作流节点类型为 Suspend 且状态为 Running
• 或者检查 spec.suspend 是否设置为 true

当 QueueUnit 出队时，Extension Server 将移除暂停条件，允许工作流继续执行。

TFJob 示例：

对于 TFJob，设置 spec.runPolicy.suspend: true 启用队列管理：

1. apiVersion: kubeflow.org/v1
2. kind: TFJob
3. metadata:
4. labels:
5. quota.scheduling.koordinator.sh/name: team-a-queue
6. spec:
7. runPolicy:
8. suspend: true

PyTorchJob 示例：

对于 PyTorchJob，设置 spec.runPolicy.suspend: true 启用队列管理：

1. apiVersion: kubeflow.org/v1
2. kind: PyTorchJob
3. metadata:
4. labels:
5. quota.scheduling.koordinator.sh/name: team-a-queue
6. spec:
7. runPolicy:
8. suspend: true

Kubeflow Jobs 的工作原理：

当提交 TFJob 或 PyTorchJob 时：

1. Job extension 检测到带有 spec.runPolicy.suspend: true 的新作业
2. 自动创建对应的 QueueUnit
3. 作业在队列中等待直到资源可用
4. 出队时，Extension Server 设置 spec.runPolicy.suspend: false，允许作业创建 Pod 并开始训练

使用 Queue

Queue Spec

Queue 优先级

默认情况下，Koord-Queue 的 Job Extensions 会自动从作业的 Pod 模板中派生 QueueUnit 的优先级：它读取 spec.template.spec.priorityClassName 和 spec.template.spec.priority。如果找到 PriorityClass 对象，则使用其 .value 作为 QueueUnit 的优先级；否则使用 spec.template.spec.priority 中的原始整数值。

你也可以在 QueueUnit 创建后手动修改其 spec.priority 来覆盖默认值，从而影响出队顺序。

Priority Queue

优先级值更高的作业优先出队。优先级相同的作业中，较早创建的作业优先出队：

设置 QueueUnit 的优先级：

1. apiVersion: scheduling.x-k8s.io/v1alpha1
2. kind: QueueUnit
3. metadata:
4. name: high-priority-job
5. namespace: default
6. spec:
7. queue: priority-queue
8. priority: 200
9. consumerRef:
10. apiVersion: batch/v1
11. kind: Job
12. name: important-job
13. namespace: default
14. resource:
15. cpu: "2"
16. memory: 4Gi

队列策略

Koord-Queue 支持三种队列策略来控制作业的出队和调度。

Priority 策略（优先级策略）

排序规则：队列单元按优先级值（降序）排序，然后按创建时间戳（升序）排序。优先级高的作业总是优先出队。相同优先级的作业按 FIFO 顺序处理。

行为特点：

• 优先级值更高的作业优先出队
• 相同优先级时，较早创建的作业先调度
• 失败的作业会被重新加入队列并重试
• 支持抢占：低优先级作业可以被抢占，为高优先级作业腾出空间

调度行为： Priority 策略不是严格的优先级调度。当高优先级作业被阻塞（如配额耗尽）时，调度器会跳过它们继续扫描。可调度的低优先级作业可以在被阻塞的高优先级作业之前出队。这提高了吞吐量并防止调度器停滞。

与 Block 策略的关键区别：

• Priority：乐观调度 - 配额接近限制时仍继续调度，阻塞的作业会被跳过
• Block：保守调度 - 配额达到限制时严格阻塞作业

适用场景：

• 多租户环境，不同优先级级别
• 生产作业应该抢占开发作业

配置示例：

1. apiVersion: scheduling.sigs.k8s.io/v1alpha1
2. kind: ElasticQuota
3. metadata:
4. name: priority-queue
5. labels:
6. koord-queue/queue-policy: Priority
7. spec:
8. max:
9. cpu: "10"
10. memory: 20Gi

Block 策略（阻塞策略）

排序规则：与 Priority 策略相同 - 按优先级（降序）然后时间戳（升序）排序。

行为特点：

• 严格资源阻塞：当配额达到限制时，使用该配额的后续作业会被阻塞
• 与 Priority 策略（会跳过阻塞的高优先级作业，让后面的可调度作业先出队）不同，Block 策略严格执行优先级顺序
• 防止资源过度分配
• 被阻塞的队列单元在调度时会被跳过，直到资源可用

关键区别：

• Priority 策略：不是严格的优先级调度 - 允许低优先级作业在被阻塞的高优先级作业之前出队
• Block 策略：严格的优先级调度 - 被阻塞的高优先级作业必须等待，防止低优先级作业绕过它们

适用场景：

• 资源受限的环境
• 需要保证资源可用性的生产工作负载
• 需要严格执行资源限制的多租户隔离场景

配置示例：

1. apiVersion: scheduling.sigs.k8s.io/v1alpha1
2. kind: ElasticQuota
3. metadata:
4. name: block-queue
5. labels:
6. koord-queue/queue-policy: Block
7. spec:
8. max:
9. cpu: "10"
10. memory: 20Gi

Intelligent 策略（智能策略）

排序规则：使用 双队列机制，配置优先级阈值（默认：4）：

• 高优先级队列：优先级 >= 阈值的作业
• 按优先级（降序）然后时间戳（升序）排序
• 重试行为：失败时，重试同一作业（FIFO 模式）
• 低优先级队列：优先级 < 阈值的作业
• 按优先级（降序）然后时间戳（升序）排序
• 重试行为：失败时，移动到下一作业（Round-Robin 模式）

行为特点：

• 优先处理高优先级作业：高优先级队列总是先被检查
• 高优先级作业获得重试保证：失败的高优先级作业立即重试
• 低优先级作业使用轮询：失败的低优先级作业让出给下一个作业
• 可通过 annotation 配置阈值：koord-queue/priority-threshold

适用场景：

• 混合负载，既有关键作业又有批量作业
• 需要既执行优先级又保证公平调度的环境

配置示例：

1. apiVersion: scheduling.sigs.k8s.io/v1alpha1
2. kind: ElasticQuota
3. metadata:
4. name: intelligent-queue
5. labels:
6. koord-queue/queue-policy: Intelligent
7. annotations:
8. koord-queue/priority-threshold: "5"
9. spec:
10. max:
11. cpu: "10"
12. memory: 20Gi

策略对比

配置队列策略

可以通过两种方式设置队列策略：

1. 通过 ElasticQuota 标签（推荐）：

1. metadata:
2. labels:
3. koord-queue/queue-policy: Priority  # 选项：Priority、Block、Intelligent

1. 通过 Queue CR（高级配置）：

1. apiVersion: scheduling.x-k8s.io/v1alpha1
2. kind: Queue
3. metadata:
4. name: my-queue
5. namespace: koord-queue
6. spec:
7. queuePolicy: Intelligent
8. priority: 1000
9. annotations:
10. koord-queue/priority-threshold: "5"

高级调优 Annotations：

• koord-queue/priority-threshold：设置 Intelligent 策略的阈值（默认：4）
• koord-queue/max-depth：限制调度时考虑的最大作业数量
• koord-queue/wait-for-pods-running：等待 Pod 进入 Running 状态后再出队下一作业

使用 QueueUnit

QueueUnit Spec

QueueUnit Status

使用 AdmissionCheck

准入检查 （开发中）

注意：准入检查控制器尚未包含在本版本中。本节描述了未来使用的计划 API。

Queue 可以要求在 QueueUnit 释放之前必须通过准入检查。这对于与外部资源供应系统集成非常有用。

1. apiVersion: scheduling.x-k8s.io/v1alpha1
2. kind: Queue
3. metadata:
4. name: checked-queue
5. namespace: koord-queue
6. spec:
7. queuePolicy: Priority
8. admissionChecks:
9. - name: prov-req-check
10. labelSelector:
11. matchLabels:
12. requires-provisioning: "true"

当 QueueUnit 被预留时，准入检查控制器会处理每个配置的检查。QueueUnit 仅在所有检查报告 Ready 状态时才转换为 Dequeued。

可观测性

监控

Koord-Queue 暴露 Prometheus 指标用于监控：

1. # 端口转发到控制器
2. $ kubectl port-forward -n koord-queue svc/koord-queue 10259:10259

4. # 获取指标
5. $ curl http://localhost:10259/metrics

如果启用了可视化服务器（enableVisibilityServer: true），可以通过 REST API 查询队列状态：

1. $ curl http://koord-queue-visibility:8090/api/queues

调试

查看控制器日志了解调度决策：

1. $ kubectl logs -n koord-queue deployment/koord-queue-controller -f --tail=100

检查 QueueUnit 状态获取调度详情：

1. $ kubectl describe queueunit <name> -n <namespace>

查看 Kubernetes 事件获取调度相关消息：

1. $ kubectl get events -n <namespace> --field-selector reason=Scheduling