Merge pull request kubernetes#34266 from jessfraz/automated-cherry-pick-of-#32914-kubernetes#33163-kubernetes#33227-kubernetes#33359-kubernetes#33605-kubernetes#33967-kubernetes#33977-kubernetes#34158-origin-release-1.4

Automatic merge from submit-queue

Automated cherry pick of kubernetes#32914 kubernetes#33163 kubernetes#33227 kubernetes#33359 kubernetes#33605 kubernetes#33967 kubernetes#33977 kubernetes#34158 origin release 1.4

Cherry pick of kubernetes#32914 kubernetes#33163 kubernetes#33227 kubernetes#33359 kubernetes#33605 kubernetes#33967 kubernetes#33977 kubernetes#34158 on release-1.4.

kubernetes#32914: Limit the number of names per image reported in the node
kubernetes#33163: fix the appending bug
kubernetes#33227: remove cpu limits for dns pod. The current limits are not
kubernetes#33359: Fix goroutine leak in federation service controller
kubernetes#33605: Add periodic ingress reconciliations.
kubernetes#33967: scheduler: cache.delete deletes the pod from node specified
kubernetes#33977: Heal the namespaceless ingresses in federation e2e.
kubernetes#34158: Add missing argument to log message in federated ingress

Author: Kubernetes Submit Queue

cluster/ubuntu/reconfDocker.sh

@@ -59,7 +59,7 @@ function restart_docker {
 
   source /run/flannel/subnet.env
   source /etc/default/docker
-  echo DOCKER_OPTS=\"${DOCKER_OPTS} -H tcp://127.0.0.1:4243 -H unix:///var/run/docker.sock \
+  echo DOCKER_OPTS=\" -H tcp://127.0.0.1:4243 -H unix:///var/run/docker.sock \
        --bip=${FLANNEL_SUBNET} --mtu=${FLANNEL_MTU}\" > /etc/default/docker
   sudo service docker restart
 }

federation/pkg/federation-controller/ingress/ingress_controller.go

@@ -666,7 +666,7 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 					ClusterName: cluster.Name,
 				})
 			} else {
-				glog.V(4).Infof("No annotation %q exists on ingress %q in federation, and index of cluster %q is %d and not zero.  Not queueing create operation for ingress %q until annotation exists", staticIPNameKeyWritable, ingress, cluster.Name, clusterIndex)
+				glog.V(4).Infof("No annotation %q exists on ingress %q in federation, and index of cluster %q is %d and not zero.  Not queueing create operation for ingress %q until annotation exists", staticIPNameKeyWritable, ingress, cluster.Name, clusterIndex, ingress)
 			}
 		} else {
 			clusterIngress := clusterIngressObj.(*extensions_v1beta1.Ingress)
@@ -679,6 +679,16 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 				glog.V(4).Infof(logStr, "Transferring")
 				if !baseIPAnnotationExists && clusterIPNameExists {
 					baseIngress.ObjectMeta.Annotations[staticIPNameKeyWritable] = clusterIPName
+					glog.V(4).Infof("Attempting to update base federated ingress annotations: %v", baseIngress)
+					if updatedFedIngress, err := ic.federatedApiClient.Extensions().Ingresses(baseIngress.Namespace).Update(baseIngress); err != nil {
+						glog.Errorf("Failed to add static IP annotation to federated ingress %q, will try again later: %v", ingress, err)
+						ic.deliverIngress(ingress, ic.ingressReviewDelay, true)
+						return
+					} else {
+						glog.V(4).Infof("Successfully updated federated ingress %q (added IP annotation), after update: %q", ingress, updatedFedIngress)
+						ic.deliverIngress(ingress, ic.smallDelay, false)
+						return
+					}
 				}
 				if !baseLBStatusExists && clusterLBStatusExists {
 					lbstatusObj, lbErr := conversion.NewCloner().DeepCopy(&clusterIngress.Status.LoadBalancer)
@@ -689,16 +699,16 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 						return
 					}
 					baseIngress.Status.LoadBalancer = *lbstatus
-				}
-				glog.V(4).Infof("Attempting to update base federated ingress: %v", baseIngress)
-				if _, err = ic.federatedApiClient.Extensions().Ingresses(baseIngress.Namespace).Update(baseIngress); err != nil {
-					glog.Errorf("Failed to add static IP annotation to federated ingress %q, will try again later: %v", ingress, err)
-					ic.deliverIngress(ingress, ic.ingressReviewDelay, true)
-					return
-				} else {
-					glog.V(4).Infof("Successfully added static IP annotation to federated ingress: %q", ingress)
-					ic.deliverIngress(ingress, ic.smallDelay, false)
-					return
+					glog.V(4).Infof("Attempting to update base federated ingress status: %v", baseIngress)
+					if updatedFedIngress, err := ic.federatedApiClient.Extensions().Ingresses(baseIngress.Namespace).UpdateStatus(baseIngress); err != nil {
+						glog.Errorf("Failed to update federated ingress status of %q (loadbalancer status), will try again later: %v", ingress, err)
+						ic.deliverIngress(ingress, ic.ingressReviewDelay, true)
+						return
+					} else {
+						glog.V(4).Infof("Successfully updated federated ingress status of %q (added loadbalancer status), after update: %q", ingress, updatedFedIngress)
+						ic.deliverIngress(ingress, ic.smallDelay, false)
+						return
+					}
 				}
 			} else {
 				glog.V(4).Infof(logStr, "Not transferring")
@@ -712,10 +722,13 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 				objMeta, err := conversion.NewCloner().DeepCopy(clusterIngress.ObjectMeta)
 				if err != nil {
 					glog.Errorf("Error deep copying ObjectMeta: %v", err)
+					ic.deliverIngress(ingress, ic.ingressReviewDelay, true)
+
 				}
 				desiredIngress.ObjectMeta, ok = objMeta.(v1.ObjectMeta)
 				if !ok {
 					glog.Errorf("Internal error: Failed to cast to v1.ObjectMeta: %v", objMeta)
+					ic.deliverIngress(ingress, ic.ingressReviewDelay, true)
 				}
 				// Merge any annotations and labels on the federated ingress onto the underlying cluster ingress,
 				// overwriting duplicates.
@@ -748,6 +761,7 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 	if len(operations) == 0 {
 		// Everything is in order
 		glog.V(4).Infof("Ingress %q is up-to-date in all clusters - no propagation to clusters required.", ingress)
+		ic.deliverIngress(ingress, ic.ingressReviewDelay, false)
 		return
 	}
 	glog.V(4).Infof("Calling federatedUpdater.Update() - operations: %v", operations)
@@ -760,4 +774,6 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 		ic.deliverIngress(ingress, ic.ingressReviewDelay, true)
 		return
 	}
+	// Schedule another periodic reconciliation, only to account for possible bugs in watch processing.
+	ic.deliverIngress(ingress, ic.ingressReviewDelay, false)
 }

federation/pkg/federation-controller/ingress/ingress_controller_test.go

@@ -99,7 +99,6 @@ func TestIngressController(t *testing.T) {
 			Name:      "test-ingress",
 			Namespace: "mynamespace",
 			SelfLink:  "/api/v1/namespaces/mynamespace/ingress/test-ingress",
-			// TODO: Remove: Annotations: map[string]string{},
 		},
 		Status: extensions_v1beta1.IngressStatus{
 			LoadBalancer: api_v1.LoadBalancerStatus{

federation/pkg/federation-controller/service/endpoint_helper.go

@@ -31,14 +31,34 @@ import (
 // worker runs a worker thread that just dequeues items, processes them, and marks them done.
 // It enforces that the syncHandler is never invoked concurrently with the same key.
 func (sc *ServiceController) clusterEndpointWorker() {
-	fedClient := sc.federationClient
+	// process all pending events in endpointWorkerDoneChan
+ForLoop:
+	for {
+		select {
+		case clusterName := <-sc.endpointWorkerDoneChan:
+			sc.endpointWorkerMap[clusterName] = false
+		default:
+			// non-blocking, comes here if all existing events are processed
+			break ForLoop
+		}
+	}
+
 	for clusterName, cache := range sc.clusterCache.clientMap {
+		workerExist, found := sc.endpointWorkerMap[clusterName]
+		if found && workerExist {
+			continue
+		}
+
+		// create a worker only if the previous worker has finished and gone out of scope
 		go func(cache *clusterCache, clusterName string) {
+			fedClient := sc.federationClient
 			for {
 				func() {
 					key, quit := cache.endpointQueue.Get()
 					// update endpoint cache
 					if quit {
+						// send signal that current worker has finished tasks and is going out of scope
+						sc.endpointWorkerDoneChan <- clusterName
 						return
 					}
 					defer cache.endpointQueue.Done(key)
@@ -49,6 +69,7 @@ func (sc *ServiceController) clusterEndpointWorker() {
 				}()
 			}
 		}(cache, clusterName)
+		sc.endpointWorkerMap[clusterName] = true
 	}
 }
 

federation/pkg/federation-controller/service/service_helper.go

@@ -35,23 +35,44 @@ import (
 // worker runs a worker thread that just dequeues items, processes them, and marks them done.
 // It enforces that the syncHandler is never invoked concurrently with the same key.
 func (sc *ServiceController) clusterServiceWorker() {
-	fedClient := sc.federationClient
+	// process all pending events in serviceWorkerDoneChan
+ForLoop:
+	for {
+		select {
+		case clusterName := <-sc.serviceWorkerDoneChan:
+			sc.serviceWorkerMap[clusterName] = false
+		default:
+			// non-blocking, comes here if all existing events are processed
+			break ForLoop
+		}
+	}
+
 	for clusterName, cache := range sc.clusterCache.clientMap {
+		workerExist, found := sc.serviceWorkerMap[clusterName]
+		if found && workerExist {
+			continue
+		}
+
+		// create a worker only if the previous worker has finished and gone out of scope
 		go func(cache *clusterCache, clusterName string) {
+			fedClient := sc.federationClient
 			for {
 				func() {
 					key, quit := cache.serviceQueue.Get()
-					defer cache.serviceQueue.Done(key)
 					if quit {
+						// send signal that current worker has finished tasks and is going out of scope
+						sc.serviceWorkerDoneChan <- clusterName
 						return
 					}
+					defer cache.serviceQueue.Done(key)
 					err := sc.clusterCache.syncService(key.(string), clusterName, cache, sc.serviceCache, fedClient, sc)
 					if err != nil {
 						glog.Errorf("Failed to sync service: %+v", err)
 					}
 				}()
 			}
 		}(cache, clusterName)
+		sc.serviceWorkerMap[clusterName] = true
 	}
 }
 

federation/pkg/federation-controller/service/servicecontroller.go

@@ -68,6 +68,8 @@ const (
 	UserAgentName = "federation-service-controller"
 	KubeAPIQPS    = 20.0
 	KubeAPIBurst  = 30
+
+	maxNoOfClusters = 100
 )
 
 type cachedService struct {
@@ -119,6 +121,16 @@ type ServiceController struct {
 	// services that need to be synced
 	queue           *workqueue.Type
 	knownClusterSet sets.String
+	// endpoint worker map contains all the clusters registered with an indication that worker exist
+	// key clusterName
+	endpointWorkerMap map[string]bool
+	// channel for worker to signal that it is going out of existence
+	endpointWorkerDoneChan chan string
+	// service worker map contains all the clusters registered with an indication that worker exist
+	// key clusterName
+	serviceWorkerMap map[string]bool
+	// channel for worker to signal that it is going out of existence
+	serviceWorkerDoneChan chan string
 }
 
 // New returns a new service controller to keep DNS provider service resources
@@ -205,6 +217,11 @@ func New(federationClient federation_release_1_4.Interface, dns dnsprovider.Inte
 			},
 		},
 	)
+
+	s.endpointWorkerMap = make(map[string]bool)
+	s.serviceWorkerMap = make(map[string]bool)
+	s.endpointWorkerDoneChan = make(chan string, maxNoOfClusters)
+	s.serviceWorkerDoneChan = make(chan string, maxNoOfClusters)
 	return s
 }
 

pkg/kubelet/kubelet.go

@@ -158,9 +158,6 @@ const (
 	// Period for performing image garbage collection.
 	ImageGCPeriod = 5 * time.Minute
 
-	// maxImagesInStatus is the number of max images we store in image status.
-	maxImagesInNodeStatus = 50
-
 	// Minimum number of dead containers to keep in a pod
 	minDeadContainerInPod = 1
 )

pkg/kubelet/kubelet_node_status.go

@@ -39,6 +39,15 @@ import (
 	"k8s.io/kubernetes/pkg/volume/util/volumehelper"
 )
 
+const (
+	// maxImagesInNodeStatus is the number of max images we store in image status.
+	maxImagesInNodeStatus = 50
+
+	// maxNamesPerImageInNodeStatus is max number of names per image stored in
+	// the node status.
+	maxNamesPerImageInNodeStatus = 5
+)
+
 // registerWithApiServer registers the node with the cluster master. It is safe
 // to call multiple times, but not concurrently (kl.registrationCompleted is
 // not locked).
@@ -501,8 +510,13 @@ func (kl *Kubelet) setNodeStatusImages(node *api.Node) {
 		}
 
 		for _, image := range containerImages {
+			names := append(image.RepoDigests, image.RepoTags...)
+			// Report up to maxNamesPerImageInNodeStatus names per image.
+			if len(names) > maxNamesPerImageInNodeStatus {
+				names = names[0:maxNamesPerImageInNodeStatus]
+			}
 			imagesOnNode = append(imagesOnNode, api.ContainerImage{
-				Names:     append(image.RepoTags, image.RepoDigests...),
+				Names:     names,
 				SizeBytes: image.Size,
 			})
 		}

pkg/kubelet/kubelet_node_status_test.go

@@ -44,6 +44,10 @@ import (
 	"k8s.io/kubernetes/pkg/volume/util/volumehelper"
 )
 
+const (
+	maxImageTagsForTest = 20
+)
+
 // generateTestingImageList generate randomly generated image list and corresponding expectedImageList.
 func generateTestingImageList(count int) ([]kubecontainer.Image, []api.ContainerImage) {
 	// imageList is randomly generated image list
@@ -64,7 +68,7 @@ func generateTestingImageList(count int) ([]kubecontainer.Image, []api.Container
 	var expectedImageList []api.ContainerImage
 	for _, kubeImage := range imageList {
 		apiImage := api.ContainerImage{
-			Names:     kubeImage.RepoTags,
+			Names:     kubeImage.RepoTags[0:maxNamesPerImageInNodeStatus],
 			SizeBytes: kubeImage.Size,
 		}
 
@@ -76,7 +80,9 @@ func generateTestingImageList(count int) ([]kubecontainer.Image, []api.Container
 
 func generateImageTags() []string {
 	var tagList []string
-	count := rand.IntnRange(1, maxImageTagsForTest+1)
+	// Generate > maxNamesPerImageInNodeStatus tags so that the test can verify
+	// that kubelet report up to maxNamesPerImageInNodeStatus tags.
+	count := rand.IntnRange(maxNamesPerImageInNodeStatus+1, maxImageTagsForTest+1)
 	for ; count > 0; count-- {
 		tagList = append(tagList, "gcr.io/google_containers:v"+strconv.Itoa(count))
 	}

pkg/kubelet/kubelet_test.go

@@ -86,8 +86,6 @@ const (
 	testReservationCPU    = "200m"
 	testReservationMemory = "100M"
 
-	maxImageTagsForTest = 3
-
 	// TODO(harry) any global place for these two?
 	// Reasonable size range of all container images. 90%ile of images on dockerhub drops into this range.
 	minImgSize int64 = 23 * 1024 * 1024

plugin/pkg/scheduler/schedulercache/cache.go

@@ -244,12 +244,12 @@ func (cache *schedulerCache) RemovePod(pod *api.Pod) error {
 	cache.mu.Lock()
 	defer cache.mu.Unlock()
 
-	_, ok := cache.podStates[key]
+	cachedstate, ok := cache.podStates[key]
 	switch {
 	// An assumed pod won't have Delete/Remove event. It needs to have Add event
 	// before Remove event, in which case the state would change from Assumed to Added.
 	case ok && !cache.assumedPods[key]:
-		err := cache.removePod(pod)
+		err := cache.removePod(cachedstate.pod)
 		if err != nil {
 			return err
 		}

test/e2e/federated-ingress.go

@@ -284,10 +284,10 @@ func createIngressOrFail(clientset *federation_release_1_4.Clientset, namespace
 		},
 	}
 
-	_, err := clientset.Extensions().Ingresses(namespace).Create(ingress)
+	newIng, err := clientset.Extensions().Ingresses(namespace).Create(ingress)
 	framework.ExpectNoError(err, "Creating ingress %q in namespace %q", ingress.Name, namespace)
 	By(fmt.Sprintf("Successfully created federated ingress %q in namespace %q", FederatedIngressName, namespace))
-	return ingress
+	return newIng
 }
 
 func updateIngressOrFail(clientset *federation_release_1_4.Clientset, namespace string) (newIng *v1beta1.Ingress) {