連載: Kubernetesでカスタムコントローラを作ろう！ ～第6回 ReplicaSetコントローラからコントローラの実装を学ぶ～

// Run begins watching and syncing.
func (rsc *ReplicaSetController) Run(ctx context.Context, workers int) {
    defer utilruntime.HandleCrash()

    // Start events processing pipeline.
    rsc.eventBroadcaster.StartStructuredLogging(0)
    rsc.eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: rsc.kubeClient.CoreV1().Events("")})
    defer rsc.eventBroadcaster.Shutdown()

    defer rsc.queue.ShutDown()

    controllerName := strings.ToLower(rsc.Kind)
    klog.Infof("Starting %v controller", controllerName)
    defer klog.Infof("Shutting down %v controller", controllerName)

    if !cache.WaitForNamedCacheSync(rsc.Kind, ctx.Done(), rsc.podListerSynced, rsc.rsListerSynced) {
        return
    }

    for i := 0; i < workers; i++ {
        go wait.UntilWithContext(ctx, rsc.worker, time.Second)
    }

    <-ctx.Done()
}

そして各ワーカーは以下のprocessNextWorkItem関数を無限ループで呼び出します。 この中で呼ばれているsyncHandlerがReconcile関数の実態です。 syncHandlerにはキューから取り出したkeyを渡していますが、一体これはなんなのでしょうか。 次にそちらを確認していきましょう。

func (rsc *ReplicaSetController) processNextWorkItem(ctx context.Context) bool {
    key, quit := rsc.queue.Get()
    if quit {
        return false
    }
    defer rsc.queue.Done(key)

    err := rsc.syncHandler(ctx, key.(string))
    if err == nil {
        rsc.queue.Forget(key)
        return true
    }

    utilruntime.HandleError(fmt.Errorf("sync %q failed with %v", key, err))
    rsc.queue.AddRateLimited(key)

    return true
}

このキューにはReconcileしたいReplicaSetが ネームスペース名/ReplicaSet名 という形でエンキューされています。 そのためReplicaSetのイベントが発生した場合はもちろんそうですが、ReplicaSetに紐づくPodにイベントがあった場合でも、その親であるReplicaSetがキューへエンキューされます。 このキューはReplicaSetControllerのaddPod, updatePod, deletePodやaddRS, updateRS, deleteRS関数の中でエンキューされています。 以上から、ReplicaSetコントローラのReconcileはReplicaSetにしか興味がないことがわかります。 これは基本的に他のコントローラも同じで、Reconcileは何か１種類のリソースに対してのみ責任を持つことがほとんどです。

syncHandlerはstructであるReplicaSetContrllerのフィールドであり、実態はsyncReplicaSetが代入されています。 ではメインであるsyncReplicaSetの実装を確認していきます。

先ほど説明した通り、syncReplicaSetは引数であるkeyでネームスペース名/ReplicaSet名を受け取ります(658行目)。 そしてそれは分解され、ネームスペース名とReplicaSet名が取り出されます(664行目)。 取り出されたReplicaSetのみを対象にこのReconcile処理が実施され、それ以外のReplicaSetはそのReconcile処理では対象としません。

func (rsc *ReplicaSetController) syncReplicaSet(ctx context.Context, key string) error {
    startTime := time.Now()
    defer func() {
        klog.V(4).Infof("Finished syncing %v %q (%v)", rsc.Kind, key, time.Since(startTime))
    }()

    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
        return err
    }
    ...(続く)

...(続き)
rs, err := rsc.rsLister.ReplicaSets(namespace).Get(name)
if apierrors.IsNotFound(err) {
    klog.V(4).Infof("%v %v has been deleted", rsc.Kind, key)
    rsc.expectations.DeleteExpectations(key)
    return nil
}
if err != nil {
    return err
}
...(続く)

...(続き)
rsNeedsSync := rsc.expectations.SatisfiedExpectations(key)
selector, err := metav1.LabelSelectorAsSelector(rs.Spec.Selector)
if err != nil {
    utilruntime.HandleError(fmt.Errorf("error converting pod selector to selector for rs %v/%v: %v", namespace, name, err))
    return nil
}

// list all pods to include the pods that don't match the rs`s selector
// anymore but has the stale controller ref.
// TODO: Do the List and Filter in a single pass, or use an index.
allPods, err := rsc.podLister.Pods(rs.Namespace).List(labels.Everything())
if err != nil {
    return err
}
// Ignore inactive pods.
filteredPods := controller.FilterActivePods(allPods)

// NOTE: filteredPods are pointing to objects from cache - if you need to
// modify them, you need to copy it first.
filteredPods, err = rsc.claimPods(ctx, rs, selector, filteredPods)
if err != nil {
    return err
}
...(続く)

...(続き)
var manageReplicasErr error
if rsNeedsSync && rs.DeletionTimestamp == nil {
    manageReplicasErr = rsc.manageReplicas(ctx, filteredPods, rs)
}
...(続く)

    ...(続き)
    rs = rs.DeepCopy()
    newStatus := calculateStatus(rs, filteredPods, manageReplicasErr)

    // Always updates status as pods come up or die.
    updatedRS, err := updateReplicaSetStatus(rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace), rs, newStatus)
    if err != nil {
        // Multiple things could lead to this update failing. Requeuing the replica set ensures
        // Returning an error causes a requeue without forcing a hotloop
        return err
    }
    // Resync the ReplicaSet after MinReadySeconds as a last line of defense to guard against clock-skew.
    if manageReplicasErr == nil && updatedRS.Spec.MinReadySeconds > 0 &&
        updatedRS.Status.ReadyReplicas == *(updatedRS.Spec.Replicas) &&
        updatedRS.Status.AvailableReplicas != *(updatedRS.Spec.Replicas) {
        rsc.queue.AddAfter(key, time.Duration(updatedRS.Spec.MinReadySeconds)*time.Second)
    }
    return manageReplicasErr
}