JobFlow Controller 深度解析

type jobflowcontroller struct {
    kubeClient kubernetes.Interface     // Kubernetes 原生客户端
    vcClient   vcclientset.Interface    // Volcano CRD 客户端

    // Informers - 监听三种 CRD 资源
    jobFlowInformer     flowinformer.JobFlowInformer
    jobTemplateInformer flowinformer.JobTemplateInformer
    jobInformer         batchinformer.JobInformer

    // Listers - 本地缓存的资源查询接口
    jobFlowLister   flowlister.JobFlowLister
    jobTemplateLister flowlister.JobTemplateLister
    jobLister        batchlister.JobLister

    // Event recorder - 事件记录器
    recorder record.EventRecorder

    // Work Queue - 异步处理队列
    queue          workqueue.TypedRateLimitingInterface[apis.FlowRequest]
    enqueueJobFlow func(req apis.FlowRequest)
    syncHandler    func(req *apis.FlowRequest) error

    maxRequeueNum int  // 最大重入队列次数（-1 表示无限制）
}

type FlowRequest struct {
    Namespace       string                // JobFlow 所在 Namespace
    JobFlowName     string                // JobFlow 名称
    JobTemplateName string                // JobTemplate 名称（当前未使用）
    Action          flowv1alpha1.Action   // 操作类型：SyncJobFlow / SyncJobTemplate
    Event           flowv1alpha1.Event    // 事件类型：OutOfSync
}

apiVersion: flow.volcano.sh/v1alpha1
kind: JobFlow
metadata:
  name: ml-pipeline
  namespace: default
spec:
  # 工作流中的所有 Flow（每个 Flow 对应一个 JobTemplate）
  flows:
    - name: data-preprocess        # 对应同名 JobTemplate
      # 无依赖，首先执行
    - name: model-training
      dependsOn:
        targets:
          - data-preprocess        # 依赖 data-preprocess 完成
    - name: model-evaluation
      dependsOn:
        targets:
          - model-training         # 依赖 model-training 完成
  # Job 保留策略：retain（保留）或 delete（完成后删除）
  jobRetainPolicy: retain
status:
  state:
    phase: Running                 # 当前状态
  pendingJobs: []
  runningJobs: ["ml-pipeline-model-training"]
  completedJobs: ["ml-pipeline-data-preprocess"]
  failedJobs: []
  terminatedJobs: []
  jobStatusList: [...]             # 每个 Job 的详细状态
  conditions: {...}                # 每个 Job 的 Condition

type Flow struct {
    Name      string     `json:"name"`                // Flow 名称，同时也是 JobTemplate 名称
    DependsOn *DependsOn `json:"dependsOn,omitempty"` // 依赖声明
    Patch     *Patch     `json:"patch,omitempty"`     // Job Spec 覆写（预留能力）
}

type DependsOn struct {
    Targets []string `json:"targets,omitempty"` // 依赖的 Flow 名称列表
    Probe   *Probe   `json:"probe,omitempty"`   // 健康检查探针（预留能力）
}

type JobTemplate struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`
    Spec   v1alpha1.JobSpec  `json:"spec,omitempty"`   // 直接嵌入 Volcano JobSpec
    Status JobTemplateStatus `json:"status,omitempty"`
}

type JobFlowStatus struct {
    PendingJobs    []string               // Pending 状态的 Job 名称列表
    RunningJobs    []string               // Running 状态的 Job 名称列表
    FailedJobs     []string               // Failed 状态的 Job 名称列表
    CompletedJobs  []string               // Completed 状态的 Job 名称列表
    TerminatedJobs []string               // Terminated 状态的 Job 名称列表
    UnKnowJobs     []string               // 未知状态的 Job 名称列表
    JobStatusList  []JobStatus            // 每个 Job 的详细状态
    Conditions     map[string]Condition   // Job 名称 -> Condition 映射
    State          State                  // JobFlow 整体状态
}

func (p *pendingState) Execute(action Action) error {
    return SyncJobFlow(p.jobFlow, func(status *JobFlowStatus, allJobList int) {
        if (len(status.RunningJobs) > 0 || len(status.CompletedJobs) > 0) &&
            len(status.FailedJobs) <= 0 {
            // 条件1：有 Job 开始运行/已完成，且无失败 -> Running
            status.State.Phase = Running
        } else if len(status.FailedJobs) > 0 || len(status.TerminatedJobs) > 0 {
            // 条件2：有 Job 失败或被终止 -> Failed
            UpdateJobFlowFailed(p.jobFlow.Namespace) // 递增 Prometheus 计数器
            status.State.Phase = Failed
        } else {
            // 条件3：保持 Pending
            status.State.Phase = Pending
        }
    })
}

func (p *runningState) Execute(action Action) error {
    return SyncJobFlow(p.jobFlow, func(status *JobFlowStatus, allJobList int) {
        if len(status.CompletedJobs) == allJobList {
            // 所有 Job 完成 -> Succeed
            UpdateJobFlowSucceed(p.jobFlow.Namespace) // 递增 Prometheus 计数器
            status.State.Phase = Succeed
        } else if len(status.FailedJobs) > 0 {
            // 存在失败 -> Failed
            status.State.Phase = Failed
        }
        // 否则保持 Running（不修改 Phase）
    })
}

func getJobName(jobFlowName string, jobTemplateName string) string {
    return jobFlowName + "-" + jobTemplateName
}

func (jf *jobflowcontroller) judge(jobFlow *JobFlow, flow Flow) (bool, error) {
    for _, targetName := range flow.DependsOn.Targets {
        targetJobName := getJobName(jobFlow.Name, targetName)
        job, err := jf.jobLister.Jobs(jobFlow.Namespace).Get(targetJobName)
        if err != nil {
            if errors.IsNotFound(err) {
                return false, nil  // 依赖 Job 尚未创建，不满足
            }
            return false, err
        }
        if job.Status.State.Phase != v1alpha1.Completed {
            return false, nil      // 依赖 Job 未完成，不满足
        }
    }
    return true, nil               // 所有依赖均已 Completed
}

// OwnerReference 示例
{
    APIVersion:         "flow.volcano.sh/v1alpha1",
    Kind:               "JobFlow",
    Name:               "ml-pipeline",
    Controller:         true,
    BlockOwnerDeletion: true,
}

func (jf *jobflowcontroller) getAllJobsCreatedByJobFlow(jobFlow *JobFlow) ([]*Job, error) {
    selector := labels.NewSelector()
    r, _ := labels.NewRequirement(
        CreatedByJobFlow,
        selection.In,
        []string{GenerateObjectString(jobFlow.Namespace, jobFlow.Name)},
    )
    selector = selector.Add(*r)
    return jf.jobLister.Jobs(jobFlow.Namespace).List(selector)
}

func (jf *jobflowcontroller) syncJobFlow(jobFlow *JobFlow, updateStateFn UpdateJobFlowStatusFn) error {
    // 阶段一：Job 保留策略检查
    if jobFlow.Spec.JobRetainPolicy == Delete &&
        jobFlow.Status.State.Phase == Succeed {
        return jf.deleteAllJobsCreatedByJobFlow(jobFlow)
    }

    // 阶段二：按 DAG 依赖顺序部署 Job
    if err := jf.deployJob(jobFlow); err != nil {
        return err
    }

    // 阶段三：收集所有 Job 状态
    jobFlowStatus, err := jf.getAllJobStatus(jobFlow)
    if err != nil {
        return err
    }

    // 阶段四：状态机转换（由各 State 的回调函数决定）
    jobFlow.Status = *jobFlowStatus
    updateStateFn(&jobFlow.Status, len(jobFlow.Spec.Flows))

    // 阶段五：更新 JobFlow Status 到 API Server
    _, err = jf.vcClient.FlowV1alpha1().JobFlows(jobFlow.Namespace).
        UpdateStatus(context.Background(), jobFlow, metav1.UpdateOptions{})
    return err
}

statusListJobMap := map[JobPhase][]string{
    Pending:     make([]string, 0),
    Running:     make([]string, 0),
    Completing:  make([]string, 0),
    Completed:   make([]string, 0),
    Terminating: make([]string, 0),
    Terminated:  make([]string, 0),
    Failed:      make([]string, 0),
}

conditions[job.Name] = Condition{
    Phase:           job.Status.State.Phase,
    CreateTimestamp: job.CreationTimestamp,
    RunningDuration: job.Status.RunningDuration,
    TaskStatusCount: job.Status.TaskStatusCount,
}

func getRunningHistories(jobStatusList []JobStatus, job *Job) []JobRunningHistory {
    // 1. 查找已有历史记录
    for _, jobStatus := range jobStatusList {
        if jobStatus.Name == job.Name && jobStatus.RunningHistories != nil {
            histories := jobStatus.RunningHistories
            // 2. 检测状态变化
            if lastState != currentState {
                // 关闭上一条历史记录（设置 EndTimestamp）
                histories[last].EndTimestamp = now
                // 开启新的历史记录
                histories = append(histories, JobRunningHistory{
                    StartTimestamp: now,
                    State:         currentState,
                })
            }
            return histories
        }
    }
    // 3. 首次记录
    if job.Status.State.Phase != "" {
        return []JobRunningHistory{{
            StartTimestamp: now,
            State:         currentState,
        }}
    }
}

[
  {"state": "Pending",   "startTimestamp": "T1", "endTimestamp": "T2"},
  {"state": "Running",   "startTimestamp": "T2", "endTimestamp": "T3"},
  {"state": "Completed", "startTimestamp": "T3", "endTimestamp": ""}
]

func (jf *jobflowcontroller) deleteAllJobsCreatedByJobFlow(jobFlow *JobFlow) error {
    // 通过 Label Selector 查询所有关联 Job
    jobList, err := jf.getAllJobsCreatedByJobFlow(jobFlow)
    if err != nil {
        return err
    }
    // 逐一删除
    for _, job := range jobList {
        err := jf.vcClient.BatchV1alpha1().Jobs(jobFlow.Namespace).
            Delete(context.Background(), job.Name, metav1.DeleteOptions{})
        if err != nil {
            return err  // 单个删除失败即中断
        }
    }
    return nil
}

func (jf *jobflowcontroller) updateJobFlow(oldObj, newObj interface{}) {
    // ...
    // 仅当 Phase == Succeed 且 JobRetainPolicy == Delete 时才入队
    if newJobFlow.Status.State.Phase != Succeed ||
        newJobFlow.Spec.JobRetainPolicy != Delete {
        return
    }
    // 入队触发 syncJobFlow()
    jf.enqueueJobFlow(req)
}

func (jf *jobflowcontroller) addJobFlow(obj interface{}) {
    jobFlow := obj.(*JobFlow)
    req := apis.FlowRequest{
        Namespace:   jobFlow.Namespace,
        JobFlowName: jobFlow.Name,
        Action:      SyncJobFlowAction,
        Event:       OutOfSyncEvent,
    }
    jf.enqueueJobFlow(req)
}

func (jf *jobflowcontroller) updateJob(oldObj, newObj interface{}) {
    // 1. 过滤非 JobFlow 创建的 Job
    if !isControlledBy(newJob, helpers.JobFlowKind) {
        return
    }
    // 2. 排除重复通知
    if newJob.ResourceVersion == oldJob.ResourceVersion {
        return
    }
    // 3. 通过 OwnerReference 反查 JobFlow 名称
    jobFlowName := getJobFlowNameByJob(newJob)
    if jobFlowName == "" {
        return
    }
    // 4. 将关联的 JobFlow 入队
    jf.enqueueJobFlow(FlowRequest{...})
}

func (jf *jobflowcontroller) handleJobFlowErr(err error, req FlowRequest) {
    if err == nil {
        jf.queue.Forget(req)  // 成功：清除重试计数
        return
    }
    if jf.maxRequeueNum == -1 || jf.queue.NumRequeues(req) < jf.maxRequeueNum {
        jf.queue.AddRateLimited(req)  // 未超限：按指数退避重入队列
        return
    }
    // 超过重试上限：记录 Warning 事件，丢弃请求
    jf.recordEventsForJobFlow(namespace, name, EventTypeWarning, action, message)
    jf.queue.Forget(req)
}

func (jf *jobflowcontroller) Run(stopCh <-chan struct{}) {
    defer jf.queue.ShutDown()

    // 1. 启动 Informer 并等待缓存同步
    jf.vcInformerFactory.Start(stopCh)
    for informerType, ok := range jf.vcInformerFactory.WaitForCacheSync(stopCh) {
        if !ok {
            klog.Errorf("caches failed to sync: %v", informerType)
            return
        }
    }

    // 2. 启动单个 Worker goroutine（每秒一次轮询）
    go wait.Until(jf.worker, time.Second, stopCh)

    // 3. 阻塞直到 stopCh 关闭
    <-stopCh
}

# 数据预处理 JobTemplate
apiVersion: flow.volcano.sh/v1alpha1
kind: JobTemplate
metadata:
  name: data-preprocess
  namespace: default
spec:
  minAvailable: 1
  schedulerName: volcano
  tasks:
    - replicas: 1
      name: preprocess
      template:
        spec:
          containers:
            - name: preprocess
              image: myregistry/data-preprocess:latest
              command: ["python", "preprocess.py"]
              resources:
                requests:
                  cpu: "2"
                  memory: "4Gi"
          restartPolicy: OnFailure
---
# 模型训练 JobTemplate
apiVersion: flow.volcano.sh/v1alpha1
kind: JobTemplate
metadata:
  name: model-training
  namespace: default
spec:
  minAvailable: 2
  schedulerName: volcano
  tasks:
    - replicas: 2
      name: trainer
      template:
        spec:
          containers:
            - name: trainer
              image: myregistry/model-training:latest
              command: ["python", "train.py"]
              resources:
                requests:
                  cpu: "4"
                  memory: "8Gi"
                  nvidia.com/gpu: "1"
          restartPolicy: OnFailure
---
# 模型评估 JobTemplate
apiVersion: flow.volcano.sh/v1alpha1
kind: JobTemplate
metadata:
  name: model-evaluation
  namespace: default
spec:
  minAvailable: 1
  schedulerName: volcano
  tasks:
    - replicas: 1
      name: evaluator
      template:
        spec:
          containers:
            - name: evaluator
              image: myregistry/model-eval:latest
              command: ["python", "evaluate.py"]
              resources:
                requests:
                  cpu: "2"
                  memory: "4Gi"
          restartPolicy: OnFailure

apiVersion: flow.volcano.sh/v1alpha1
kind: JobFlow
metadata:
  name: ml-pipeline
  namespace: default
spec:
  jobRetainPolicy: retain
  flows:
    # 第一阶段：数据预处理（无依赖，立即执行）
    - name: data-preprocess

    # 第二阶段：模型训练（依赖数据预处理完成）
    - name: model-training
      dependsOn:
        targets:
          - data-preprocess

    # 第三阶段：模型评估（依赖模型训练完成）
    - name: model-evaluation
      dependsOn:
        targets:
          - model-training

apiVersion: flow.volcano.sh/v1alpha1
kind: JobFlow
metadata:
  name: parallel-pipeline
  namespace: default
spec:
  jobRetainPolicy: delete   # 完成后自动清理 Job
  flows:
    - name: data-download           # 无依赖

    - name: feature-engineering     # 依赖 data-download
      dependsOn:
        targets: [data-download]

    - name: model-training-v1       # 依赖 data-download（与 feature-engineering 并行）
      dependsOn:
        targets: [data-download]

    - name: model-training-v2       # 依赖 feature-engineering
      dependsOn:
        targets: [feature-engineering]

    - name: model-ensemble          # 依赖两个训练任务都完成
      dependsOn:
        targets:
          - model-training-v1
          - model-training-v2