第 2 篇｜KubeRay 协调：从 Pod 变化到 Reconcile

本文最后更新于：39 分钟前

前言

第一篇里，RayJob default/demo-job 创建了专用 RayCluster，示例名称为 default/demo-job-abcde，包含一个 head Pod 和 compute 组的两个 worker Pod。RayJob Controller 要等集群就绪，才会创建 submitter Job。假设最后一个 worker Pod 已经进入 Running，Ready 条件也刚变为 True，这个变化怎样传到 Operator，又怎样让作业继续提交？

这次变化要经过两次协调：RayCluster Controller 先确认整套集群已经就绪，把 Ready 写入 RayCluster；RayJob Controller 再读到这个状态，继续提交程序。watch 和工作队列把两次协调接起来。下面沿这条路径看每一步接收什么、留下什么。

从资源变化到协调请求

监听、缓存与事件通知

第 0 篇中负责回报 Pod 状态的 kubelet，会把本节点观察到的结果写入 API Server。Operator 通过 watch 接收所关注资源的变化通知；这是 Kubernetes API 提供的监听机制。

通知到达时，整个集群是否就绪还没有结论。最后一个 worker 的变化只是一次检查机会；Controller 仍要把 head 和其他 worker 一起检查。

管理这些对象同步和通知的客户端组件叫 informer。它在 Operator 进程内维护资源的本地缓存：例如某个 Pod 的名称、所属集群和已观察到的状态，都可以从这份对象记录中读取。收到对象更新后，informer 更新本地记录，并通知注册的事件处理器，源码中通常称为 handler。

handler 把变化映射到需要检查的对象。本例中，它沿 worker Pod 的 owner reference（拥有关系）找到 RayCluster，将集群的 namespace/name 放进队列。队列记住的是集群标识；补建 Pod、判断就绪等动作由后面的 Reconcile 决定。

Controller 的运行框架在后台从队列取出标识，调用 Reconcile；需要的资源对象则从客户端读取。负责取队列、调用函数的 goroutine 在本文中称为协调协程。goroutine 是 Go 运行时调度的执行单元，可以在同一进程内共享内存，与执行用户计算的 Ray worker 各有职责。

KubeRay 在 main.go 中创建 Manager，配置缓存、监听的 namespace 和客户端。Manager 是 Operator 进程里的框架对象，各个 Controller 通过它注册监听，共享缓存和客户端等基础设施。这些组件在同一管理程序中协作，不需要分别部署成独立服务。

Manager 只接收配置范围内的资源：WatchNamespace 限定命名空间，internal/managercache/cache.go 还为 Pod、Kubernetes Job 等设置标签筛选。Operator 的 RBAC 权限必须允许它读取和监听这些资源。

启动时，框架先等待缓存和事件源完成初始同步，再启动协调协程。因此，后面的读取建立在已经初始化的本地资源视图上；watch 权限或同步失败会阻止这一步完成。

资源关系与事件映射

RayCluster Controller 注册监听的代码位于 SetupWithManager。这段代码是在声明监听规则：哪些资源发生变化时，需要让这个 Controller 再检查一次 RayCluster。其中 predicate 是事件筛选条件，generation 是用于识别期望配置变更的计数。下面保留构建器调用，省略后面的调度器配置和 Controller 选项：

b := ctrl.NewControllerManagedBy(mgr).
    For(&rayv1.RayCluster{}, builder.WithPredicates(predicate.Or(
        predicate.GenerationChangedPredicate{},
        predicate.LabelChangedPredicate{},
        predicate.AnnotationChangedPredicate{},
    ))).
    Owns(&corev1.Pod{}).
    Owns(&corev1.Service{}).
    Owns(&corev1.Secret{}).
    Owns(&corev1.PersistentVolumeClaim{})

For 指定这个 Controller 主要协调 RayCluster。RayCluster 自身发生符合条件的事件时，处理器把它的 namespace 和 name 放进队列。Owns 关注下级资源：Pod 变化时，处理器沿它的 owner reference 找到所属 RayCluster，再把这个集群的标识放进队列。

对照代码中的括号，WithPredicates(...) 是传给 For(RayCluster) 的参数，只筛选 RayCluster 自身这条监听路径。三个条件用 predicate.Or 连接：对更新事件，只要 generation、label 或 annotation 有一项变化，就可以通过这组筛选。

例如，使用者修改 worker 数量要求，RayCluster 的期望配置发生变化，generation 随之变化，可以触发重新检查；Controller 只把观察结果写回 status 时，通常不会改变 generation，也就不会仅凭这一条件触发自身协调。这样可以避免每次写回状态都立即通过同一条监听路径再次入队。

代码中的 Owns(Pod) 没有传入这组筛选条件，因此 worker 的 Ready 条件变化仍然可以触发集群检查。RayJob 对下级 RayCluster 的监听也独立配置。同一次 RayCluster 状态更新，可以被 RayJob Controller 接收，而被 RayCluster Controller 自身的这组条件过滤掉。

本例中，两个 worker 的 owner 都是 demo-job-abcde。无论哪个 worker 更新，送入 RayCluster 队列的都是同一个集群标识。

Secret 和 PersistentVolumeClaim（PVC）也是可监听的下级资源，分别承载敏感配置和持久存储声明；具体是否创建它们取决于集群配置。

对应的框架实现位于 Builder.doWatch：For 使用 EnqueueRequestForObject，Owns 使用 EnqueueRequestForOwner。后者默认读取直接拥有关系中 controller: true 的引用。因此，Pod 先触发 RayCluster 协调，不能越过这一层直接触发 RayJob。

owner handler 匹配资源的 API 组 Group 和类型 Kind，例如 ray.io 与 RayCluster，再构造下面的请求。以下是 getOwnerReconcileRequest 的连续节选：

1
2
3

request := reconcile.Request{NamespacedName: types.NamespacedName{
    Name: ref.Name,
}}

后续代码根据资源作用域补上 namespace。对 RayCluster 来说，请求标识形如 default/demo-job-abcde，只告诉协调函数该检查哪套集群。至于要不要创建 worker，要等读到对象后再判断。

For 和 Owns 把不同来源的变化汇成 RayCluster 请求。RayCluster 状态写回以后，RayJob 自己的监听才产生第二层请求：

待处理请求的合并

假设 compute 组的两个 worker 接连更新，处理器从它们的拥有关系都找到了 default/demo-job-abcde。如果每条通知都对应一次独立协调，Controller 就可能连续检查同一套集群，重复读取相同的资源。队列因此按对象标识合并尚待处理的请求：在取得处理机会之前，同一个 RayCluster 的多次入队可以汇成一个待处理项。

这里可以对照缓存和队列各自保留的内容。缓存保存资源对象，例如各个 worker 已观察到的状态；队列中的 default/demo-job-abcde 只表示“这套集群还需要检查一次”。它不记录每条 Pod 通知，也不携带“创建一个 worker”这样的操作指令。只要下一轮重新读取集群和所属 Pod，就可以一起判断这些变化带来的结果。

取得处理机会以后，情况又有区别。假设这一轮已经读过 Pod，另一个 worker 才更新，新的请求就有必要留下。队列允许为正在处理的对象登记下一轮检查；在本轮结束前，同一队列不会把这个对象再次交给另一个协调协程。期间又到来的相同请求仍可合并。等本轮完成、下一轮开始后，再有变化，还可以继续安排检查。

下一轮可能发现本轮已经完成了所需动作，此时直接结束即可。协调逻辑必须允许重复检查；如果它要调用会产生外部副作用的接口，还需要稳定的请求标识等措施，避免重复提交。第四篇会沿作业提交解释这一点。

一轮协调中的资源与状态

读取当前对象并维护资源

协调协程取得这份请求后，先用其中的标识读取 RayCluster，再检查相关资源。触发请求的 worker 可能在进入 Running 后又将 Ready 条件更新为 True，也可能已经删除；本轮依据当前能读到的对象作判断，无须重放中间每一条通知。

“写 API 成功”和“缓存里已看到新对象”之间存在时间差。Manager 提供的常规类型客户端通常从缓存读，写操作直接发给 API Server；绕过缓存的读取另有路径。这个差异在第三篇分析并发创建时还会用到。

RayCluster 的 Reconcile 开头就进行了读取。下面是连续节选：

instance := &rayv1.RayCluster{}
if err = r.Get(ctx, request.NamespacedName, instance); err == nil {
    return r.rayClusterReconcile(ctx, instance)
}

这段代码用请求里的 namespace/name 读取 RayCluster，把结果放进 instance，读取成功才进入实际维护函数。后面的 NotFound 分支处理对象已删除、旧请求仍留在队列里的情况，此时可以结束本轮；连接或权限等读取错误则要按失败处理。

进入 rayClusterReconcile 后，代码会检查资源是否交给外部 Controller 管理，验证配置，处理删除流程，再按当前状态维护集群资源。不同配置会走不同分支。本例没有删除请求，也不启用自动扩缩容，重点是 head Service、head Pod、worker Pod 和集群状态。

spec 保存使用者的期望，status 保存 Controller 已观察到的集群情况；实际 Pod、Service 等对象也参与判断。只读某一个 status 字段，不能代替完整的资源检查。

假设配置要求两个 worker，本轮只观察到一个，Controller 会结合已有 Pod 和尚未被缓存观察到的创建记录，判断是否需要补建。创建请求发出后，Pod 可能还要等待调度和镜像拉取，本轮只完成当前能做的管理动作；如果数量已满足，就继续检查这些 Pod 的状态。

计算集群状态与 Controller 交接

资源维护之后，还要把观察结果整理成集群状态。在 raycluster_controller.go 中，rayClusterReconcile 执行完各项维护操作，接着调用负责计算状态的 calculateStatus：

1	`newInstance, calculateErr := r.calculateStatus(ctx, instance, reconcileErr)`

这里的 instance 是当前 RayCluster，reconcileErr 记录前面资源维护是否出错。calculateStatus 复制这个对象，列出所属 Pod，再把计算结果放进返回的 newInstance；调用方在计算成功后将新状态写回 API。计算状态和写回状态是这一步里的两项操作。

本例关注集群第一次进入 Ready 的条件。下面是 calculateStatus 中设置 status.state 的连续节选；runtimePods 是刚列出的 Pod，DesiredWorkerReplicas 是期望的 worker 数量：

if reconcileErr == nil && len(runtimePods.Items) == int(newInstance.Status.DesiredWorkerReplicas)+1 { // workers + 1 head
	if utils.CheckAllPodsRunning(ctx, runtimePods) {
		newInstance.Status.State = rayv1.Ready
		newInstance.Status.Reason = ""
	}
}

代入一个 head、两个 worker 的配置，外层条件要求本轮维护没有报错，并且已观察到的 Pod 总数为 3。内层的 CheckAllPodsRunning 再逐个检查这些 Pod，通过后才把状态设为 Ready。仅仅收到了最后一个 worker 的更新通知，还不足以跳过这些检查。

CheckAllPodsRunning 位于 utils/util.go，输入是 Pod 列表，返回值表示是否通过检查。函数如下：

func CheckAllPodsRunning(ctx context.Context, runningPods corev1.PodList) bool {
	log := ctrl.LoggerFrom(ctx)
	// check if there are no pods.
	if len(runningPods.Items) == 0 {
		return false
	}
	for _, pod := range runningPods.Items {
		if pod.Status.Phase != corev1.PodRunning {
			log.Info("CheckAllPodsRunning: Pod is not running.", "podName", pod.Name, "pod Status.Phase", pod.Status.Phase)
			return false
		}
		for _, cond := range pod.Status.Conditions {
			if cond.Type == corev1.PodReady && cond.Status != corev1.ConditionTrue {
				log.Info("CheckAllPodsRunning: Pod is not ready.", "podName", pod.Name, "pod Status.Conditions[PodReady]", cond)
				return false
			}
		}
	}
	return true
}

按代码顺序看：列表为空时返回 false；有 Pod 的 phase 不是 Running 时返回 false；有 Ready 条件且值不是 True 时，也返回 false。因此，若三个 Pod 都已 Running，但最后一个 worker 的 Ready 仍为 False，集群就不会在这一分支被设为 Ready。等它变成 True，后续协调重新读取 Pod，满足上述条件后才会设置集群状态。

集群状态计算成功并写回 API 后，RayJob Controller 通过自己注册的 Owns(RayCluster) 观察到变化，把所属 demo-job 的标识放入自己的队列。轮到它处理时，再读取集群状态、判断提交条件。到这里，worker 的就绪变化才传到了作业管理层：两次协调通过 API 中保存的状态接续，RayCluster Controller 完成本轮后即可返回。

一轮结束后的再次触发

RayCluster 写回 Ready，完成了向 RayJob 的进度交接。但协调函数返回之后，对象可能还需要继续处理：集群会变化，Ray 程序也会结束。下一轮从哪里来，取决于进展能否通过 Kubernetes 事件获知。

事件通知与主动重查

一轮 Reconcile 返回以后，watch 仍然持续工作。例如，worker 的 Ready 条件更新，可以沿前面的事件路径触发新的 RayCluster 协调；集群状态写回后，又可以触发 RayJob 协调。每轮处理都会重新读取当前状态、判断下一步条件。这种由事件触发的检查，本身不需要定时轮询。

随着案例进入作业执行阶段，RayJob Controller 还要跟踪另一类进展：Ray 内部的作业状态。程序执行完成时，承载它的 Pod 可能仍然正常运行，未必产生相应的 Kubernetes 资源更新。只等待 Pod 事件，就可能无法及时发现作业已经结束。因此，RayJob Controller 会查询 Ray Jobs API；如果作业还在运行，就安排稍后再查。

watch 持续接收 Kubernetes 对象的变化，延时重查则给 Controller 另一次查询机会。两者可以同时工作，最后都把同一个 RayJob 标识交回队列。

返回结果与队列安排

前面写回 API 的 status 描述资源已经走到哪一步，供使用者和其他 Controller 读取。这里的 Result 与 error 则返回给调用 Reconcile 的框架：前者表达是否以及何时再查，后者表达本轮是否失败。框架根据它们安排当前对象的下一轮处理。

例如，成功查询 Ray Jobs API、得知作业仍在运行，是正常等待，可以要求稍后再看；查询失败、没能获得作业状态，则需要按错误规则安排重试。一次协调成功，只表示本轮检查和管理动作没有报错，并不要求 Ray 作业已经完成。

需要延时重查时，代码返回 RequeueAfter，把后续安排交给框架，当前函数随后就结束了。等待期间，协调协程可以处理其他对象，不必留在函数里等待 Pod 就绪或 Ray 作业完成。框架登记同一个对象的延时入队要求，到时再给它处理机会；下一轮会从协调入口重新读取资源，而不是从上次函数退出的位置继续执行。

RayJob 的收尾代码把状态写回 API，并返回默认重查间隔。下面两段之间省略了一次指标更新：

if err = r.updateRayJobStatus(ctx, originalRayJobInstance, rayJobInstance); err != nil {
    logger.Info("Failed to update RayJob status", "error", err)
    return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
}
// ... 指标更新省略。
return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, nil

这两处都返回了相同的延时，实际是否采用，还要看 error。框架先处理错误：普通 error 非空时走限速重试，并忽略 RequeueAfter。rate limiter 即限速器，在这里决定重试的等待时间，避免失败请求立即反复执行。TerminalError 则是明确标记为“不由本次错误自动重试”的错误类型，后续资源变化仍可触发新的协调。

在 controller-runtime reconcileHandler 中，成功且设置延时的分支使用下面两行连续代码：

1 2	`c.Queue.Forget(req) c.Queue.AddWithOpts(priorityqueue.AddOpts{After: result.RequeueAfter, Priority: new(priority)}, req)`

Forget(req) 清除这个对象的限速重试记录，AddWithOpts 根据 RequeueAfter 指定的时长安排延时入队，之后由框架再次调用协调函数。清除重试记录不会删除 RayJob，也不会停止监听。

返回结果	框架如何处理
普通 error 非空	记录错误，按 rate limiter 安排重试；忽略返回值中的延时
`TerminalError`	本次错误不触发自动限速重试；后续资源事件仍可能再次触发处理
error 为空，`RequeueAfter > 0`	清除限速记录，安排延时请求
error 为空，`RequeueAfter <= 0` 且 `Requeue: true`	走限速重排路径
error 为空，空 Result	本轮不主动安排重查；后续事件仍可以再次入队

例如，RayJob 已安排延时重查，期间又收到 RayCluster 更新，这个事件可能让它更早得到检查。延时到期后还要等待可用协调协程，所以 RequeueAfter 表示再次入队的安排，不能用作精准定时器。无论哪条路径触发，下一轮都从读取对象开始。

小结：事件、队列与协调的分工

本例中，worker Pod 的变化通过监听进入 Operator，事件处理器按拥有关系把 RayCluster 标识放进队列。RayCluster Controller 读取当前对象和相关资源，维护 Pod、计算状态并写回 API；RayJob Controller 再通过自己的监听和队列读取这个结果，继续获取 Dashboard 地址、创建 submitter Job。两层 Controller 通过保存下来的资源状态协作。

每一轮只完成当前能做的管理动作。返回以后，监听仍然接收资源变化；需要跟踪 Ray 作业等外部进展时，还可以安排延时查询，失败则按错误路径处理。资源状态交给下一层 Controller，返回值交给运行框架，分别承担业务进度传递和后续处理安排。

队列把同一对象尚待处理的请求合并，协调逻辑依据当前能读到的状态判断；本轮进行期间的新变化，还可以留下下一轮检查。对使用者来说，声明已保存、Pod 已就绪、RayJob 已推进到下一阶段，可以出现在不同时间，它们通过上述过程逐步完成。

下一篇把这条路径放到 100 个 RayCluster 同时变化的场景里，沿入队、请求合并和任务分配的实现，分析不同对象怎样并行，以及同一个对象的多轮检查怎样串行衔接。

参考资料

源码阅读

#Ray #分布式计算 #源码阅读 #Kubernetes #KubeRay

第 2 篇｜KubeRay 协调：从 Pod 变化到 Reconcile

https://tanxinyu.work/kuberay-volcano-02-reconcile/

作者

谭新宇

发布于

2026年9月15日

许可协议

第 1 篇｜KubeRay 架构：组件分工与协作上一篇

第 3 篇｜KubeRay 并发：一个 Operator 如何管理 100 个 RayCluster 下一篇