k8s replicaset controller 分析（3）-expectations 机制分析

// pkg/controller/replicaset/replica_set.gofunc (rsc *ReplicaSetController) syncReplicaSet(key string) error {  startTime := time.Now()  defer func() {    klog.V(4).Infof("Finished syncing %v %q (%v)", rsc.Kind, key, time.Since(startTime))  }()
  ...
  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: %v", err))    return nil  }
  ...
  var manageReplicasErr error  if rsNeedsSync && rs.DeletionTimestamp == nil {    manageReplicasErr = rsc.manageReplicas(filteredPods, rs)  }    ...}

// pkg/controller/controller_utils.go// SatisfiedExpectations returns true if the required adds/dels for the given controller have been observed.// Add/del counts are established by the controller at sync time, and updated as controllees are observed by the controller// manager.func (r *ControllerExpectations) SatisfiedExpectations(controllerKey string) bool {  if exp, exists, err := r.GetExpectations(controllerKey); exists {    if exp.Fulfilled() {      klog.V(4).Infof("Controller expectations fulfilled %#v", exp)      return true    } else if exp.isExpired() {      klog.V(4).Infof("Controller expectations expired %#v", exp)      return true    } else {      klog.V(4).Infof("Controller still waiting on expectations %#v", exp)      return false    }  } else if err != nil {    klog.V(2).Infof("Error encountered while checking expectations %#v, forcing sync", err)  } else {    // When a new controller is created, it doesn't have expectations.    // When it doesn't see expected watch events for > TTL, the expectations expire.    //  - In this case it wakes up, creates/deletes controllees, and sets expectations again.    // When it has satisfied expectations and no controllees need to be created/destroyed > TTL, the expectations expire.    //  - In this case it continues without setting expectations till it needs to create/delete controllees.    klog.V(4).Infof("Controller %v either never recorded expectations, or the ttl expired.", controllerKey)  }  // Trigger a sync if we either encountered and error (which shouldn't happen since we're  // getting from local store) or this controller hasn't established expectations.  return true}

// Fulfilled returns true if this expectation has been fulfilled.func (e *ControlleeExpectations) Fulfilled() bool {  // TODO: think about why this line being atomic doesn't matter  return atomic.LoadInt64(&e.add) <= 0 && atomic.LoadInt64(&e.del) <= 0}

func (exp *ControlleeExpectations) isExpired() bool {  return clock.RealClock{}.Since(exp.timestamp) > ExpectationsTimeout}

// pkg/controller/replicaset/replica_set.gofunc (rsc *ReplicaSetController) manageReplicas(filteredPods []*v1.Pod, rs *apps.ReplicaSet) error {  diff := len(filteredPods) - int(*(rs.Spec.Replicas))    ...      if diff > rsc.burstReplicas {      diff = rsc.burstReplicas    }        rsc.expectations.ExpectCreations(rsKey, diff)    klog.V(2).Infof("Too few replicas for %v %s/%s, need %d, creating %d", rsc.Kind, rs.Namespace, rs.Name, *(rs.Spec.Replicas), diff)        successfulCreations, err := slowStartBatch(diff, controller.SlowStartInitialBatchSize, func() error {      ...    })
    if skippedPods := diff - successfulCreations; skippedPods > 0 {      klog.V(2).Infof("Slow-start failure. Skipping creation of %d pods, decrementing expectations for %v %v/%v", skippedPods, rsc.Kind, rs.Namespace, rs.Name)      for i := 0; i < skippedPods; i++ {        // Decrement the expected number of creates because the informer won't observe this pod        rsc.expectations.CreationObserved(rsKey)      }    }  ...

// pkg/controller/controller_utils.gofunc (r *ControllerExpectations) ExpectCreations(controllerKey string, adds int) error {  return r.SetExpectations(controllerKey, adds, 0)}
// SetExpectations registers new expectations for the given controller. Forgets existing expectations.func (r *ControllerExpectations) SetExpectations(controllerKey string, add, del int) error {  exp := &ControlleeExpectations{add: int64(add), del: int64(del), key: controllerKey, timestamp: clock.RealClock{}.Now()}  klog.V(4).Infof("Setting expectations %#v", exp)  return r.Add(exp)}

// pkg/controller/controller_utils.go// CreationObserved atomically decrements the `add` expectation count of the given controller.func (r *ControllerExpectations) CreationObserved(controllerKey string) {  r.LowerExpectations(controllerKey, 1, 0)}
// Decrements the expectation counts of the given controller.func (r *ControllerExpectations) LowerExpectations(controllerKey string, add, del int) {  if exp, exists, err := r.GetExpectations(controllerKey); err == nil && exists {    exp.Add(int64(-add), int64(-del))    // The expectations might've been modified since the update on the previous line.    klog.V(4).Infof("Lowered expectations %#v", exp)  }}

// pkg/controller/replicaset/replica_set.go// When a pod is created, enqueue the replica set that manages it and update its expectations.func (rsc *ReplicaSetController) addPod(obj interface{}) {  pod := obj.(*v1.Pod)
  ...    // If it has a ControllerRef, that's all that matters.  if controllerRef := metav1.GetControllerOf(pod); controllerRef != nil {    rs := rsc.resolveControllerRef(pod.Namespace, controllerRef)    if rs == nil {      return    }    rsKey, err := controller.KeyFunc(rs)    if err != nil {      return    }    klog.V(4).Infof("Pod %s created: %#v.", pod.Name, pod)    rsc.expectations.CreationObserved(rsKey)    rsc.queue.Add(rsKey)    return  }
  ...}