Garbage Collector Controller 深度解析

// 源码: pkg/controllers/garbagecollector/garbagecollector.go

type gccontroller struct {
    vcClient    vcclientset.Interface                          // Volcano API 客户端，用于直接访问 API Server
    jobInformer batchinformers.JobInformer                     // Job Informer，监听 Job 事件
    vcInformerFactory vcinformer.SharedInformerFactory          // Informer 工厂，管理 Informer 生命周期
    jobLister   batchlisters.JobLister                         // Job Lister，从本地缓存读取 Job
    jobSynced   func() bool                                    // 缓存同步状态检查函数
    queue       workqueue.TypedRateLimitingInterface[string]   // 带速率限制的工作队列
    workers     uint32                                         // Worker goroutine 数量
}

// 源码: pkg/controllers/garbagecollector/garbagecollector.go

func init() {
    framework.RegisterController(&gccontroller{})
}

func (gc *gccontroller) Name() string {
    return "gc-controller"
}

func (gc *gccontroller) Initialize(opt *framework.ControllerOption) error {
    gc.vcClient = opt.VolcanoClient

    factory := opt.VCSharedInformerFactory
    jobInformer := factory.Batch().V1alpha1().Jobs()

    gc.vcInformerFactory = factory
    gc.jobInformer = jobInformer
    gc.jobLister = jobInformer.Lister()
    gc.jobSynced = jobInformer.Informer().HasSynced
    gc.queue = workqueue.NewTypedRateLimitingQueue(
        workqueue.DefaultTypedControllerRateLimiter[string](),
    )
    gc.workers = opt.WorkerThreadsForGC

    jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    gc.addJob,
        UpdateFunc: gc.updateJob,
        // 注意: 没有 DeleteFunc，GC Controller 不关心删除事件
    })

    return nil
}

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

    klog.Infof("Starting garbage collector")
    defer klog.Infof("Shutting down garbage collector")

    // 1. 启动 Informer
    gc.vcInformerFactory.Start(stopCh)

    // 2. 等待缓存同步
    for informerType, ok := range gc.vcInformerFactory.WaitForCacheSync(stopCh) {
        if !ok {
            klog.Errorf("caches failed to sync: %v", informerType)
            return
        }
    }

    // 3. 启动 N 个 Worker goroutine
    for i := 0; i < int(gc.workers); i++ {
        go wait.Until(gc.worker, time.Second, stopCh)
    }

    // 4. 阻塞直到收到停止信号
    <-stopCh
}

func (gc *gccontroller) addJob(obj interface{}) {
    job := obj.(*v1alpha1.Job)
    klog.V(4).Infof("Adding job %s/%s", job.Namespace, job.Name)

    if job.DeletionTimestamp == nil && needsCleanup(job) {
        gc.enqueue(job)
    }
}

func (gc *gccontroller) updateJob(old, cur interface{}) {
    job := cur.(*v1alpha1.Job)
    klog.V(4).Infof("Updating job %s/%s", job.Namespace, job.Name)

    if job.DeletionTimestamp == nil && needsCleanup(job) {
        gc.enqueue(job)
    }
}

// 立即入队
func (gc *gccontroller) enqueue(job *v1alpha1.Job) {
    key, err := cache.MetaNamespaceKeyFunc(job)
    if err != nil {
        klog.Errorf("couldn't get key for object %#v: %v", job, err)
        return
    }
    gc.queue.Add(key)
}

// 延迟入队 - 用于 TTL 尚未到期的场景
func (gc *gccontroller) enqueueAfter(job *v1alpha1.Job, after time.Duration) {
    key, err := cache.MetaNamespaceKeyFunc(job)
    if err != nil {
        klog.Errorf("couldn't get key for object %#v: %v", job, err)
        return
    }
    gc.queue.AddAfter(key, after)
}

func (gc *gccontroller) worker() {
    for gc.processNextWorkItem() {
    }
}

func (gc *gccontroller) processNextWorkItem() bool {
    key, quit := gc.queue.Get()  // 阻塞获取下一个待处理的 key
    if quit {
        return false             // 队列已关闭，退出 Worker
    }
    defer gc.queue.Done(key)     // 标记 key 处理完成

    err := gc.processJob(key)    // 处理 Job
    gc.handleErr(err, key)       // 处理错误

    return true                  // 继续处理下一个
}

func (gc *gccontroller) processJob(key string) error {
    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
        return err
    }

    // ============ 第一阶段: 缓存检查 ============
    // 从本地 Informer 缓存获取 Job，避免给 API Server 施加压力
    job, err := gc.jobLister.Jobs(namespace).Get(name)
    if apierrors.IsNotFound(err) {
        return nil  // Job 已不存在，无需处理
    }
    if err != nil {
        return err
    }

    // 第一次 TTL 检查 (基于缓存数据)
    if expired, err := gc.processTTL(job); err != nil {
        return err
    } else if !expired {
        return nil  // TTL 未过期，已通过 enqueueAfter 延迟重新入队
    }

    // ============ 第二阶段: API Server 确认 ============
    // TTL 看起来已过期，但缓存数据可能是过时的
    // 从 API Server 获取最新副本进行最终确认
    fresh, err := gc.vcClient.BatchV1alpha1().Jobs(namespace).Get(
        context.TODO(), name, metav1.GetOptions{},
    )
    if apierrors.IsNotFound(err) {
        return nil  // Job 在此期间已被删除
    }
    if err != nil {
        return err
    }

    // 第二次 TTL 检查 (基于最新数据)
    if expired, err := gc.processTTL(fresh); err != nil {
        return err
    } else if !expired {
        return nil  // TTL 被修改，实际上未过期
    }

    // ============ 第三阶段: 安全删除 ============
    policy := metav1.DeletePropagationForeground
    options := metav1.DeleteOptions{
        PropagationPolicy: &policy,
        Preconditions:     &metav1.Preconditions{UID: &fresh.UID},
    }
    err = gc.vcClient.BatchV1alpha1().Jobs(fresh.Namespace).Delete(
        context.TODO(), fresh.Name, options,
    )
    if apierrors.IsNotFound(err) {
        return nil  // 已被删除，视为成功
    }
    return err
}

func (gc *gccontroller) processTTL(job *v1alpha1.Job) (expired bool, err error) {
    // 前置检查: 正在删除中或不需要清理的 Job 直接跳过
    if job.DeletionTimestamp != nil || !needsCleanup(job) {
        return false, nil
    }

    now := time.Now()
    t, err := timeLeft(job, &now)
    if err != nil {
        return false, err
    }

    // TTL 已过期 (剩余时间 <= 0)
    if *t <= 0 {
        return true, nil
    }

    // TTL 未过期，延迟入队，在 TTL 到期时重新检查
    gc.enqueueAfter(job, *t)
    return false, nil
}

func needsCleanup(j *v1alpha1.Job) bool {
    return j.Spec.TTLSecondsAfterFinished != nil && isJobFinished(j)
}

func isJobFinished(job *v1alpha1.Job) bool {
    return job.Status.State.Phase == v1alpha1.Completed ||
        job.Status.State.Phase == v1alpha1.Failed ||
        job.Status.State.Phase == v1alpha1.Terminated
}

func jobFinishTime(finishedJob *v1alpha1.Job) (metav1.Time, error) {
    if finishedJob.Status.State.LastTransitionTime.IsZero() {
        return metav1.Time{}, fmt.Errorf(
            "unable to find the time when the Job %s/%s finished",
            finishedJob.Namespace, finishedJob.Name,
        )
    }
    return finishedJob.Status.State.LastTransitionTime, nil
}

func getFinishAndExpireTime(j *v1alpha1.Job) (*time.Time, *time.Time, error) {
    if !needsCleanup(j) {
        return nil, nil, fmt.Errorf("job %s/%s should not be cleaned up", j.Namespace, j.Name)
    }
    finishAt, err := jobFinishTime(j)
    if err != nil {
        return nil, nil, err
    }
    finishAtUTC := finishAt.UTC()
    expireAtUTC := finishAtUTC.Add(
        time.Duration(*j.Spec.TTLSecondsAfterFinished) * time.Second,
    )
    return &finishAtUTC, &expireAtUTC, nil
}

func timeLeft(j *v1alpha1.Job, since *time.Time) (*time.Duration, error) {
    finishAt, expireAt, err := getFinishAndExpireTime(j)
    if err != nil {
        return nil, err
    }
    // 时钟偏移检测
    if finishAt.UTC().After(since.UTC()) {
        klog.Warningf("Warning: Found Job %s/%s finished in the future. "+
            "This is likely due to time skew in the cluster. "+
            "Job cleanup will be deferred.", j.Namespace, j.Name)
    }
    remaining := expireAt.UTC().Sub(since.UTC())
    return &remaining, nil
}

policy := metav1.DeletePropagationForeground
options := metav1.DeleteOptions{
    PropagationPolicy: &policy,
    Preconditions:     &metav1.Preconditions{UID: &fresh.UID},
}

Preconditions: &metav1.Preconditions{UID: &fresh.UID}

err = gc.vcClient.BatchV1alpha1().Jobs(fresh.Namespace).Delete(
    context.TODO(), fresh.Name, options,
)
if apierrors.IsNotFound(err) {
    return nil  // 已被删除，视为成功
}
return err

func (gc *gccontroller) handleErr(err error, key string) {
    if err == nil {
        gc.queue.Forget(key)  // 处理成功，清除重试计数器
        return
    }

    klog.Errorf("error cleaning up Job %v, will retry: %v", key, err)
    gc.queue.AddRateLimited(key)  // 处理失败，指数退避重试
}

// 源码: cmd/controller-manager/app/options/options.go

const (
    defaultGCWorkers = 1
)

// ServerOption 中的字段
type ServerOption struct {
    // ...
    WorkerThreadsForGC uint32
    // ...
}

// 注册命令行参数
fs.Uint32Var(&s.WorkerThreadsForGC, "worker-threads-for-gc",
    defaultGCWorkers,
    "The number of threads for recycling jobs. "+
    "The larger the number, the faster the job recycling, "+
    "but requires more CPU load.")

// 源码: staging/src/volcano.sh/apis/pkg/apis/batch/v1alpha1/job.go

type JobSpec struct {
    // ...

    // ttlSecondsAfterFinished limits the lifetime of a Job that has finished
    // execution (either Completed or Failed). If this field is set,
    // ttlSecondsAfterFinished after the Job finishes, it is eligible to be
    // deleted immediately after it finishes.
    // +kubebuilder:validation:Minimum=0
    // +optional
    TTLSecondsAfterFinished *int32 `json:"ttlSecondsAfterFinished,omitempty"`

    // ...
}

apiVersion: batch.volcano.sh/v1alpha1
kind: Job
metadata:
  name: training-job
  namespace: default
spec:
  ttlSecondsAfterFinished: 3600  # 完成后 1 小时自动清理
  minAvailable: 1
  schedulerName: volcano
  queue: default
  tasks:
    - replicas: 1
      name: trainer
      template:
        spec:
          containers:
            - name: tensorflow
              image: tensorflow/tensorflow:2.12.0-gpu
              command: ["python", "train.py"]
              resources:
                limits:
                  nvidia.com/gpu: 1
          restartPolicy: Never

apiVersion: batch.volcano.sh/v1alpha1
kind: Job
metadata:
  name: data-preprocessing
  namespace: default
spec:
  ttlSecondsAfterFinished: 0  # 完成后立即清理
  minAvailable: 3
  schedulerName: volcano
  tasks:
    - replicas: 3
      name: preprocessor
      template:
        spec:
          containers:
            - name: spark
              image: apache/spark:3.4.0
              command: ["spark-submit", "--class", "Main", "app.jar"]
          restartPolicy: Never

apiVersion: batch.volcano.sh/v1alpha1
kind: Job
metadata:
  name: critical-job
  namespace: production
spec:
  # 不设置 ttlSecondsAfterFinished，GC Controller 不会自动清理
  # 需要管理员手动删除或通过外部脚本清理
  minAvailable: 2
  schedulerName: volcano
  tasks:
    - replicas: 2
      name: worker
      template:
        spec:
          containers:
            - name: app
              image: my-app:latest
              command: ["./run.sh"]
          restartPolicy: OnFailure