k8s replicaset controller 分析（2）- 核心处理逻辑分析

// syncReplicaSet will sync the ReplicaSet with the given key if it has had its expectations fulfilled,// meaning it did not expect to see any more of its pods created or deleted. This function is not meant to be// invoked concurrently with the same key.func (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))  }()
  namespace, name, err := cache.SplitMetaNamespaceKey(key)  if err != nil {    return err  }  rs, err := rsc.rsLister.ReplicaSets(namespace).Get(name)  if errors.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: %v", 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(rs, selector, filteredPods)  if err != nil {    return err  }
  var manageReplicasErr error  if rsNeedsSync && rs.DeletionTimestamp == nil {    manageReplicasErr = rsc.manageReplicas(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}

// pkg/controller/controller_utils.gofunc (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}
func (exp *ControlleeExpectations) isExpired() bool {  return clock.RealClock{}.Since(exp.timestamp) > ExpectationsTimeout // ExpectationsTimeout = 5 * time.Minute}

// pkg/controller/replicaset/replica_set.gofunc (rsc *ReplicaSetController) manageReplicas(filteredPods []*v1.Pod, rs *apps.ReplicaSet) error {  diff := len(filteredPods) - int(*(rs.Spec.Replicas))  rsKey, err := controller.KeyFunc(rs)  if err != nil {    utilruntime.HandleError(fmt.Errorf("Couldn't get key for %v %#v: %v", rsc.Kind, rs, err))    return nil  }  if diff < 0 {    diff *= -1    if diff > rsc.burstReplicas {      diff = rsc.burstReplicas    }    // TODO: Track UIDs of creates just like deletes. The problem currently    // is we'd need to wait on the result of a create to record the pod's    // UID, which would require locking *across* the create, which will turn    // into a performance bottleneck. We should generate a UID for the pod    // beforehand and store it via ExpectCreations.    rsc.expectations.ExpectCreations(rsKey, diff)    glog.V(2).Infof("Too few replicas for %v %s/%s, need %d, creating %d", rsc.Kind, rs.Namespace, rs.Name, *(rs.Spec.Replicas), diff)    // Batch the pod creates. Batch sizes start at SlowStartInitialBatchSize    // and double with each successful iteration in a kind of "slow start".    // This handles attempts to start large numbers of pods that would    // likely all fail with the same error. For example a project with a    // low quota that attempts to create a large number of pods will be    // prevented from spamming the API service with the pod create requests    // after one of its pods fails.  Conveniently, this also prevents the    // event spam that those failures would generate.    successfulCreations, err := slowStartBatch(diff, controller.SlowStartInitialBatchSize, func() error {      boolPtr := func(b bool) *bool { return &b }      controllerRef := &metav1.OwnerReference{        APIVersion:         rsc.GroupVersion().String(),        Kind:               rsc.Kind,        Name:               rs.Name,        UID:                rs.UID,        BlockOwnerDeletion: boolPtr(true),        Controller:         boolPtr(true),      }      err := rsc.podControl.CreatePodsWithControllerRef(rs.Namespace, &rs.Spec.Template, rs, controllerRef)      if err != nil && errors.IsTimeout(err) {        // Pod is created but its initialization has timed out.        // If the initialization is successful eventually, the        // controller will observe the creation via the informer.        // If the initialization fails, or if the pod keeps        // uninitialized for a long time, the informer will not        // receive any update, and the controller will create a new        // pod when the expectation expires.        return nil      }      return err    })
    // Any skipped pods that we never attempted to start shouldn't be expected.    // The skipped pods will be retried later. The next controller resync will    // retry the slow start process.    if skippedPods := diff - successfulCreations; skippedPods > 0 {      glog.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)      }    }    return err  } else if diff > 0 {    if diff > rsc.burstReplicas {      diff = rsc.burstReplicas    }    glog.V(2).Infof("Too many replicas for %v %s/%s, need %d, deleting %d", rsc.Kind, rs.Namespace, rs.Name, *(rs.Spec.Replicas), diff)
    // Choose which Pods to delete, preferring those in earlier phases of startup.    podsToDelete := getPodsToDelete(filteredPods, diff)
    // Snapshot the UIDs (ns/name) of the pods we're expecting to see    // deleted, so we know to record their expectations exactly once either    // when we see it as an update of the deletion timestamp, or as a delete.    // Note that if the labels on a pod/rs change in a way that the pod gets    // orphaned, the rs will only wake up after the expectations have    // expired even if other pods are deleted.    rsc.expectations.ExpectDeletions(rsKey, getPodKeys(podsToDelete))
    errCh := make(chan error, diff)    var wg sync.WaitGroup    wg.Add(diff)    for _, pod := range podsToDelete {      go func(targetPod *v1.Pod) {        defer wg.Done()        if err := rsc.podControl.DeletePod(rs.Namespace, targetPod.Name, rs); err != nil {          // Decrement the expected number of deletes because the informer won't observe this deletion          podKey := controller.PodKey(targetPod)          glog.V(2).Infof("Failed to delete %v, decrementing expectations for %v %s/%s", podKey, rsc.Kind, rs.Namespace, rs.Name)          rsc.expectations.DeletionObserved(rsKey, podKey)          errCh <- err        }      }(pod)    }    wg.Wait()
    select {    case err := <-errCh:      // all errors have been reported before and they're likely to be the same, so we'll only return the first one we hit.      if err != nil {        return err      }    default:    }  }
  return nil}

diff := len(filteredPods) - int(*(rs.Spec.Replicas))

// pkg/controller/replicaset/replica_set.goconst (  // Realistic value of the burstReplica field for the replica set manager based off  // performance requirements for kubernetes 1.0.  BurstReplicas = 500
  // The number of times we retry updating a ReplicaSet's status.  statusUpdateRetries = 1)

    if diff > rsc.burstReplicas {    diff = rsc.burstReplicas  }

        diff *= -1    if diff > rsc.burstReplicas {      diff = rsc.burstReplicas    }        rsc.expectations.ExpectCreations(rsKey, diff)    glog.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 {      boolPtr := func(b bool) *bool { return &b }      controllerRef := &metav1.OwnerReference{        APIVersion:         rsc.GroupVersion().String(),        Kind:               rsc.Kind,        Name:               rs.Name,        UID:                rs.UID,        BlockOwnerDeletion: boolPtr(true),        Controller:         boolPtr(true),      }      err := rsc.podControl.CreatePodsWithControllerRef(rs.Namespace, &rs.Spec.Template, rs, controllerRef)      if err != nil && errors.IsTimeout(err) {        // Pod is created but its initialization has timed out.        // If the initialization is successful eventually, the        // controller will observe the creation via the informer.        // If the initialization fails, or if the pod keeps        // uninitialized for a long time, the informer will not        // receive any update, and the controller will create a new        // pod when the expectation expires.        return nil      }      return err    })
    if skippedPods := diff - successfulCreations; skippedPods > 0 {      glog.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)      }    }    return err

// pkg/controller/replicaset/replica_set.go// slowStartBatch tries to call the provided function a total of 'count' times,// starting slow to check for errors, then speeding up if calls succeed.//// It groups the calls into batches, starting with a group of initialBatchSize.// Within each batch, it may call the function multiple times concurrently.//// If a whole batch succeeds, the next batch may get exponentially larger.// If there are any failures in a batch, all remaining batches are skipped// after waiting for the current batch to complete.//// It returns the number of successful calls to the function.func slowStartBatch(count int, initialBatchSize int, fn func() error) (int, error) {  remaining := count  successes := 0  for batchSize := integer.IntMin(remaining, initialBatchSize); batchSize > 0; batchSize = integer.IntMin(2*batchSize, remaining) {    errCh := make(chan error, batchSize)    var wg sync.WaitGroup    wg.Add(batchSize)    for i := 0; i < batchSize; i++ {      go func() {        defer wg.Done()        if err := fn(); err != nil {          errCh <- err        }      }()    }    wg.Wait()    curSuccesses := batchSize - len(errCh)    successes += curSuccesses    if len(errCh) > 0 {      return successes, <-errCh    }    remaining -= batchSize  }  return successes, nil}

func (r RealPodControl) CreatePodsWithControllerRef(namespace string, template *v1.PodTemplateSpec, controllerObject runtime.Object, controllerRef *metav1.OwnerReference) error {  if err := validateControllerRef(controllerRef); err != nil {    return err  }  return r.createPods("", namespace, template, controllerObject, controllerRef)}
func (r RealPodControl) createPods(nodeName, namespace string, template *v1.PodTemplateSpec, object runtime.Object, controllerRef *metav1.OwnerReference) error {  pod, err := GetPodFromTemplate(template, object, controllerRef)  if err != nil {    return err  }  if len(nodeName) != 0 {    pod.Spec.NodeName = nodeName  }  if len(labels.Set(pod.Labels)) == 0 {    return fmt.Errorf("unable to create pods, no labels")  }  newPod, err := r.KubeClient.CoreV1().Pods(namespace).Create(pod)  if err != nil {    // only send an event if the namespace isn't terminating    if !apierrors.HasStatusCause(err, v1.NamespaceTerminatingCause) {      r.Recorder.Eventf(object, v1.EventTypeWarning, FailedCreatePodReason, "Error creating: %v", err)    }    return err  }  accessor, err := meta.Accessor(object)  if err != nil {    klog.Errorf("parentObject does not have ObjectMeta, %v", err)    return nil  }  klog.V(4).Infof("Controller %v created pod %v", accessor.GetName(), newPod.Name)  r.Recorder.Eventf(object, v1.EventTypeNormal, SuccessfulCreatePodReason, "Created pod: %v", newPod.Name)
  return nil}

if diff > rsc.burstReplicas {      diff = rsc.burstReplicas    }    glog.V(2).Infof("Too many replicas for %v %s/%s, need %d, deleting %d", rsc.Kind, rs.Namespace, rs.Name, *(rs.Spec.Replicas), diff)
    // Choose which Pods to delete, preferring those in earlier phases of startup.    podsToDelete := getPodsToDelete(filteredPods, diff)
    rsc.expectations.ExpectDeletions(rsKey, getPodKeys(podsToDelete))
    errCh := make(chan error, diff)    var wg sync.WaitGroup    wg.Add(diff)    for _, pod := range podsToDelete {      go func(targetPod *v1.Pod) {        defer wg.Done()        if err := rsc.podControl.DeletePod(rs.Namespace, targetPod.Name, rs); err != nil {          // Decrement the expected number of deletes because the informer won't observe this deletion          podKey := controller.PodKey(targetPod)          glog.V(2).Infof("Failed to delete %v, decrementing expectations for %v %s/%s", podKey, rsc.Kind, rs.Namespace, rs.Name)          rsc.expectations.DeletionObserved(rsKey, podKey)          errCh <- err        }      }(pod)    }    wg.Wait()
    select {    case err := <-errCh:      // all errors have been reported before and they're likely to be the same, so we'll only return the first one we hit.      if err != nil {        return err      }    default:    }

// pkg/controller/replicaset/replica_set.gofunc getPodsToDelete(filteredPods, relatedPods []*v1.Pod, diff int) []*v1.Pod {  // No need to sort pods if we are about to delete all of them.  // diff will always be <= len(filteredPods), so not need to handle > case.  if diff < len(filteredPods) {    podsWithRanks := getPodsRankedByRelatedPodsOnSameNode(filteredPods, relatedPods)    sort.Sort(podsWithRanks)  }  return filteredPods[:diff]}
func getPodsRankedByRelatedPodsOnSameNode(podsToRank, relatedPods []*v1.Pod) controller.ActivePodsWithRanks {  podsOnNode := make(map[string]int)  for _, pod := range relatedPods {    if controller.IsPodActive(pod) {      podsOnNode[pod.Spec.NodeName]++    }  }  ranks := make([]int, len(podsToRank))  for i, pod := range podsToRank {    ranks[i] = podsOnNode[pod.Spec.NodeName]  }  return controller.ActivePodsWithRanks{Pods: podsToRank, Rank: ranks}}

// pkg/controller/controller_utils.gofunc (s ActivePodsWithRanks) Less(i, j int) bool {  // 1. Unassigned < assigned  // If only one of the pods is unassigned, the unassigned one is smaller  if s.Pods[i].Spec.NodeName != s.Pods[j].Spec.NodeName && (len(s.Pods[i].Spec.NodeName) == 0 || len(s.Pods[j].Spec.NodeName) == 0) {    return len(s.Pods[i].Spec.NodeName) == 0  }  // 2. PodPending < PodUnknown < PodRunning  if podPhaseToOrdinal[s.Pods[i].Status.Phase] != podPhaseToOrdinal[s.Pods[j].Status.Phase] {    return podPhaseToOrdinal[s.Pods[i].Status.Phase] < podPhaseToOrdinal[s.Pods[j].Status.Phase]  }  // 3. Not ready < ready  // If only one of the pods is not ready, the not ready one is smaller  if podutil.IsPodReady(s.Pods[i]) != podutil.IsPodReady(s.Pods[j]) {    return !podutil.IsPodReady(s.Pods[i])  }  // 4. Doubled up < not doubled up  // If one of the two pods is on the same node as one or more additional  // ready pods that belong to the same replicaset, whichever pod has more  // colocated ready pods is less  if s.Rank[i] != s.Rank[j] {    return s.Rank[i] > s.Rank[j]  }  // TODO: take availability into account when we push minReadySeconds information from deployment into pods,  //       see https://github.com/kubernetes/kubernetes/issues/22065  // 5. Been ready for empty time < less time < more time  // If both pods are ready, the latest ready one is smaller  if podutil.IsPodReady(s.Pods[i]) && podutil.IsPodReady(s.Pods[j]) {    readyTime1 := podReadyTime(s.Pods[i])    readyTime2 := podReadyTime(s.Pods[j])    if !readyTime1.Equal(readyTime2) {      return afterOrZero(readyTime1, readyTime2)    }  }  // 6. Pods with containers with higher restart counts < lower restart counts  if maxContainerRestarts(s.Pods[i]) != maxContainerRestarts(s.Pods[j]) {    return maxContainerRestarts(s.Pods[i]) > maxContainerRestarts(s.Pods[j])  }  // 7. Empty creation time pods < newer pods < older pods  if !s.Pods[i].CreationTimestamp.Equal(&s.Pods[j].CreationTimestamp) {    return afterOrZero(&s.Pods[i].CreationTimestamp, &s.Pods[j].CreationTimestamp)  }  return false}

// pkg/controller/controller_utils.gofunc (r RealPodControl) DeletePod(namespace string, podID string, object runtime.Object) error {  accessor, err := meta.Accessor(object)  if err != nil {    return fmt.Errorf("object does not have ObjectMeta, %v", err)  }  klog.V(2).Infof("Controller %v deleting pod %v/%v", accessor.GetName(), namespace, podID)  if err := r.KubeClient.CoreV1().Pods(namespace).Delete(podID, nil); err != nil && !apierrors.IsNotFound(err) {    r.Recorder.Eventf(object, v1.EventTypeWarning, FailedDeletePodReason, "Error deleting: %v", err)    return fmt.Errorf("unable to delete pods: %v", err)  }  r.Recorder.Eventf(object, v1.EventTypeNormal, SuccessfulDeletePodReason, "Deleted pod: %v", podID)
  return nil}

func calculateStatus(rs *apps.ReplicaSet, filteredPods []*v1.Pod, manageReplicasErr error) apps.ReplicaSetStatus {  newStatus := rs.Status  // Count the number of pods that have labels matching the labels of the pod  // template of the replica set, the matching pods may have more  // labels than are in the template. Because the label of podTemplateSpec is  // a superset of the selector of the replica set, so the possible  // matching pods must be part of the filteredPods.  fullyLabeledReplicasCount := 0  readyReplicasCount := 0  availableReplicasCount := 0  templateLabel := labels.Set(rs.Spec.Template.Labels).AsSelectorPreValidated()  for _, pod := range filteredPods {    if templateLabel.Matches(labels.Set(pod.Labels)) {      fullyLabeledReplicasCount++    }    if podutil.IsPodReady(pod) {      readyReplicasCount++      if podutil.IsPodAvailable(pod, rs.Spec.MinReadySeconds, metav1.Now()) {        availableReplicasCount++      }    }  }
  failureCond := GetCondition(rs.Status, apps.ReplicaSetReplicaFailure)  if manageReplicasErr != nil && failureCond == nil {    var reason string    if diff := len(filteredPods) - int(*(rs.Spec.Replicas)); diff < 0 {      reason = "FailedCreate"    } else if diff > 0 {      reason = "FailedDelete"    }    cond := NewReplicaSetCondition(apps.ReplicaSetReplicaFailure, v1.ConditionTrue, reason, manageReplicasErr.Error())    SetCondition(&newStatus, cond)  } else if manageReplicasErr == nil && failureCond != nil {    RemoveCondition(&newStatus, apps.ReplicaSetReplicaFailure)  }
  newStatus.Replicas = int32(len(filteredPods))  newStatus.FullyLabeledReplicas = int32(fullyLabeledReplicasCount)  newStatus.ReadyReplicas = int32(readyReplicasCount)  newStatus.AvailableReplicas = int32(availableReplicasCount)  return newStatus}

// pkg/controller/replicaset/replica_set_utils.gofunc updateReplicaSetStatus(c appsclient.ReplicaSetInterface, rs *apps.ReplicaSet, newStatus apps.ReplicaSetStatus) (*apps.ReplicaSet, error) {  // This is the steady state. It happens when the ReplicaSet doesn't have any expectations, since  // we do a periodic relist every 30s. If the generations differ but the replicas are  // the same, a caller might've resized to the same replica count.  if rs.Status.Replicas == newStatus.Replicas &&    rs.Status.FullyLabeledReplicas == newStatus.FullyLabeledReplicas &&    rs.Status.ReadyReplicas == newStatus.ReadyReplicas &&    rs.Status.AvailableReplicas == newStatus.AvailableReplicas &&    rs.Generation == rs.Status.ObservedGeneration &&    reflect.DeepEqual(rs.Status.Conditions, newStatus.Conditions) {    return rs, nil  }
  // Save the generation number we acted on, otherwise we might wrongfully indicate  // that we've seen a spec update when we retry.  // TODO: This can clobber an update if we allow multiple agents to write to the  // same status.  newStatus.ObservedGeneration = rs.Generation
  var getErr, updateErr error  var updatedRS *apps.ReplicaSet  for i, rs := 0, rs; ; i++ {    klog.V(4).Infof(fmt.Sprintf("Updating status for %v: %s/%s, ", rs.Kind, rs.Namespace, rs.Name) +      fmt.Sprintf("replicas %d->%d (need %d), ", rs.Status.Replicas, newStatus.Replicas, *(rs.Spec.Replicas)) +      fmt.Sprintf("fullyLabeledReplicas %d->%d, ", rs.Status.FullyLabeledReplicas, newStatus.FullyLabeledReplicas) +      fmt.Sprintf("readyReplicas %d->%d, ", rs.Status.ReadyReplicas, newStatus.ReadyReplicas) +      fmt.Sprintf("availableReplicas %d->%d, ", rs.Status.AvailableReplicas, newStatus.AvailableReplicas) +      fmt.Sprintf("sequence No: %v->%v", rs.Status.ObservedGeneration, newStatus.ObservedGeneration))
    rs.Status = newStatus    updatedRS, updateErr = c.UpdateStatus(rs)    if updateErr == nil {      return updatedRS, nil    }    // Stop retrying if we exceed statusUpdateRetries - the replicaSet will be requeued with a rate limit.    if i >= statusUpdateRetries {      break    }    // Update the ReplicaSet with the latest resource version for the next poll    if rs, getErr = c.Get(rs.Name, metav1.GetOptions{}); getErr != nil {      // If the GET fails we can't trust status.Replicas anymore. This error      // is bound to be more interesting than the update failure.      return nil, getErr    }  }
  return nil, updateErr}

// staging/src/k8s.io/client-go/kubernetes/typed/apps/v1/replicaset.gofunc (c *replicaSets) UpdateStatus(replicaSet *v1.ReplicaSet) (result *v1.ReplicaSet, err error) {  result = &v1.ReplicaSet{}  err = c.client.Put().    Namespace(c.ns).    Resource("replicasets").    Name(replicaSet.Name).    SubResource("status").    Body(replicaSet).    Do().    Into(result)  return}