Merge pull request #20264 from Random-Liu/cleanup-config

pkg/kubelet/config/config.go

@@ -208,16 +208,11 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
 		pods = make(map[string]*api.Pod)
 	}
 
-	update := change.(kubetypes.PodUpdate)
-	switch update.Op {
-	case kubetypes.ADD, kubetypes.UPDATE:
-		if update.Op == kubetypes.ADD {
-			glog.V(4).Infof("Adding new pods from source %s : %v", source, update.Pods)
-		} else {
-			glog.V(4).Infof("Updating pods from source %s : %v", source, update.Pods)
-		}
-
-		filtered := filterInvalidPods(update.Pods, source, s.recorder)
+	// updatePodFunc is the local function which updates the pod cache *oldPods* with new pods *newPods*.
+	// After updated, new pod will be stored in the pod cache *pods*.
+	// Notice that *pods* and *oldPods* could be the same cache.
+	updatePodsFunc := func(newPods []*api.Pod, oldPods, pods map[string]*api.Pod) {
+		filtered := filterInvalidPods(newPods, source, s.recorder)
 		for _, ref := range filtered {
 			name := kubecontainer.GetPodFullName(ref)
 			// Annotate the pod with the source before any comparison.
@@ -225,7 +220,8 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
 				ref.Annotations = make(map[string]string)
 			}
 			ref.Annotations[kubetypes.ConfigSourceAnnotationKey] = source
-			if existing, found := pods[name]; found {
+			if existing, found := oldPods[name]; found {
+				pods[name] = existing
 				needUpdate, needReconcile := checkAndUpdatePod(existing, ref)
 				if needUpdate {
 					updatePods = append(updatePods, existing)
@@ -247,6 +243,17 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
 				addPods = append(addPods, ref)
 			}
 		}
+	}
+
+	update := change.(kubetypes.PodUpdate)
+	switch update.Op {
+	case kubetypes.ADD, kubetypes.UPDATE:
+		if update.Op == kubetypes.ADD {
+			glog.V(4).Infof("Adding new pods from source %s : %v", source, update.Pods)
+		} else {
+			glog.V(4).Infof("Updating pods from source %s : %v", source, update.Pods)
+		}
+		updatePodsFunc(update.Pods, pods, pods)
 
 	case kubetypes.REMOVE:
 		glog.V(4).Infof("Removing a pod %v", update)
@@ -267,39 +274,7 @@ func (s *podStorage) merge(source string, change interface{}) (adds, updates, de
 		// Clear the old map entries by just creating a new map
 		oldPods := pods
 		pods = make(map[string]*api.Pod)
-
-		filtered := filterInvalidPods(update.Pods, source, s.recorder)
-		for _, ref := range filtered {
-			name := kubecontainer.GetPodFullName(ref)
-			// Annotate the pod with the source before any comparison.
-			if ref.Annotations == nil {
-				ref.Annotations = make(map[string]string)
-			}
-			ref.Annotations[kubetypes.ConfigSourceAnnotationKey] = source
-			if existing, found := oldPods[name]; found {
-				pods[name] = existing
-				needUpdate, needReconcile := checkAndUpdatePod(existing, ref)
-				if needUpdate {
-					updatePods = append(updatePods, existing)
-				} else if needReconcile {
-					reconcilePods = append(reconcilePods, existing)
-				}
-				continue
-			}
-			recordFirstSeenTime(ref)
-			pods[name] = ref
-			// If a pod is not found in the cache, and it's also not in the
-			// pending phase, it implies that kubelet may have restarted.
-			// Treat this pod as update so that kubelet wouldn't reject the
-			// pod in the admission process.
-			if ref.Status.Phase != api.PodPending {
-				updatePods = append(updatePods, ref)
-			} else {
-				// this is an add
-				addPods = append(addPods, ref)
-			}
-		}
-
+		updatePodsFunc(update.Pods, oldPods, pods)
 		for name, existing := range oldPods {
 			if _, found := pods[name]; !found {
 				// this is a delete