Merge pull request #21400 from derekwaynecarr/namespace_deletion_discovery

cmd/kube-controller-manager/app/controllermanager.go

@@ -40,6 +40,7 @@ import (
 	"k8s.io/kubernetes/pkg/client/leaderelection"
 	"k8s.io/kubernetes/pkg/client/record"
 	"k8s.io/kubernetes/pkg/client/restclient"
+	"k8s.io/kubernetes/pkg/client/typed/dynamic"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
 	"k8s.io/kubernetes/pkg/cloudprovider"
@@ -266,7 +267,14 @@ func StartControllers(s *options.CMServer, kubeClient *client.Client, kubeconfig
 		glog.Fatalf("Failed to get supported resources from server: %v", err)
 	}
 
-	namespaceController := namespacecontroller.NewNamespaceController(clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "namespace-controller")), versions, s.NamespaceSyncPeriod.Duration)
+	// Find the list of namespaced resources via discovery that the namespace controller must manage
+	namespaceKubeClient := clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "namespace-controller"))
+	namespaceClientPool := dynamic.NewClientPool(restclient.AddUserAgent(kubeconfig, "namespace-controller"), dynamic.LegacyAPIPathResolverFunc)
+	groupVersionResources, err := namespacecontroller.ServerPreferredNamespacedGroupVersionResources(namespaceKubeClient.Discovery())
+	if err != nil {
+		glog.Fatalf("Failed to get supported resources from server: %v", err)
+	}
+	namespaceController := namespacecontroller.NewNamespaceController(namespaceKubeClient, namespaceClientPool, groupVersionResources, s.NamespaceSyncPeriod.Duration, api.FinalizerKubernetes)
 	go namespaceController.Run(s.ConcurrentNamespaceSyncs, wait.NeverStop)
 
 	groupVersion := "extensions/v1beta1"

contrib/mesos/pkg/controllermanager/controllermanager.go

@@ -32,6 +32,7 @@ import (
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/client/restclient"
+	"k8s.io/kubernetes/pkg/client/typed/dynamic"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
 	clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api"
@@ -214,7 +215,14 @@ func (s *CMServer) Run(_ []string) error {
 		glog.Fatalf("Failed to get supported resources from server: %v", err)
 	}
 
-	namespaceController := namespacecontroller.NewNamespaceController(clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "namespace-controller")), versions, s.NamespaceSyncPeriod.Duration)
+	// Find the list of namespaced resources via discovery that the namespace controller must manage
+	namespaceKubeClient := clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "namespace-controller"))
+	namespaceClientPool := dynamic.NewClientPool(restclient.AddUserAgent(kubeconfig, "namespace-controller"), dynamic.LegacyAPIPathResolverFunc)
+	groupVersionResources, err := namespacecontroller.ServerPreferredNamespacedGroupVersionResources(namespaceKubeClient.Discovery())
+	if err != nil {
+		glog.Fatalf("Failed to get supported resources from server: %v", err)
+	}
+	namespaceController := namespacecontroller.NewNamespaceController(namespaceKubeClient, namespaceClientPool, groupVersionResources, s.NamespaceSyncPeriod.Duration, api.FinalizerKubernetes)
 	go namespaceController.Run(s.ConcurrentNamespaceSyncs, wait.NeverStop)
 
 	groupVersion := "extensions/v1beta1"

pkg/client/typed/dynamic/client_pool.go

+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package dynamic
+
+import (
+	"sync"
+
+	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/client/restclient"
+)
+
+// ClientPool manages a pool of dynamic clients.
+type ClientPool interface {
+	// ClientForGroupVersion returns a client configured for the specified groupVersion.
+	ClientForGroupVersion(groupVersion unversioned.GroupVersion) (*Client, error)
+}
+
+// APIPathResolverFunc knows how to convert a groupVersion to its API path.
+type APIPathResolverFunc func(groupVersion unversioned.GroupVersion) string
+
+// LegacyAPIPathResolverFunc can resolve paths properly with the legacy API.
+func LegacyAPIPathResolverFunc(groupVersion unversioned.GroupVersion) string {
+	if len(groupVersion.Group) == 0 {
+		return "/api"
+	}
+	return "/apis"
+}
+
+// clientPoolImpl implements Factory
+type clientPoolImpl struct {
+	lock                sync.RWMutex
+	config              *restclient.Config
+	clients             map[unversioned.GroupVersion]*Client
+	apiPathResolverFunc APIPathResolverFunc
+}
+
+// NewClientPool returns a ClientPool from the specified config
+func NewClientPool(config *restclient.Config, apiPathResolverFunc APIPathResolverFunc) ClientPool {
+	return &clientPoolImpl{
+		config:              config,
+		clients:             map[unversioned.GroupVersion]*Client{},
+		apiPathResolverFunc: apiPathResolverFunc,
+	}
+}
+
+// ClientForGroupVersion returns a client for the specified groupVersion, creates one if none exists
+func (c *clientPoolImpl) ClientForGroupVersion(groupVersion unversioned.GroupVersion) (*Client, error) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	// do we have a client already configured?
+	if existingClient, found := c.clients[groupVersion]; found {
+		return existingClient, nil
+	}
+
+	// avoid changing the original config
+	confCopy := *c.config
+	conf := &confCopy
+
+	// we need to set the api path based on group version, if no group, default to legacy path
+	conf.APIPath = c.apiPathResolverFunc(groupVersion)
+
+	// we need to make a client
+	conf.GroupVersion = &groupVersion
+	dynamicClient, err := NewClient(conf)
+	if err != nil {
+		return nil, err
+	}
+	c.clients[groupVersion] = dynamicClient
+	return dynamicClient, nil
+}

pkg/controller/namespace/namespace_controller.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	"k8s.io/kubernetes/pkg/client/cache"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
+	"k8s.io/kubernetes/pkg/client/typed/dynamic"
 	"k8s.io/kubernetes/pkg/controller"
 	"k8s.io/kubernetes/pkg/controller/framework"
 	"k8s.io/kubernetes/pkg/runtime"
@@ -38,23 +39,37 @@ import (
 type NamespaceController struct {
 	// client that purges namespace content, must have list/delete privileges on all content
 	kubeClient clientset.Interface
+	// clientPool manages a pool of dynamic clients
+	clientPool dynamic.ClientPool
 	// store that holds the namespaces
 	store cache.Store
 	// controller that observes the namespaces
 	controller *framework.Controller
 	// namespaces that have been queued up for processing by workers
 	queue *workqueue.Type
-	// list of versions to process
-	versions *unversioned.APIVersions
+	// list of preferred group versions and their corresponding resource set for namespace deletion
+	groupVersionResources []unversioned.GroupVersionResource
+	// opCache is a cache to remember if a particular operation is not supported to aid dynamic client.
+	opCache operationNotSupportedCache
+	// finalizerToken is the finalizer token managed by this controller
+	finalizerToken api.FinalizerName
 }
 
 // NewNamespaceController creates a new NamespaceController
-func NewNamespaceController(kubeClient clientset.Interface, versions *unversioned.APIVersions, resyncPeriod time.Duration) *NamespaceController {
+func NewNamespaceController(
+	kubeClient clientset.Interface,
+	clientPool dynamic.ClientPool,
+	groupVersionResources []unversioned.GroupVersionResource,
+	resyncPeriod time.Duration,
+	finalizerToken api.FinalizerName) *NamespaceController {
 	// create the controller so we can inject the enqueue function
 	namespaceController := &NamespaceController{
 		kubeClient: kubeClient,
-		versions:   versions,
+		clientPool: clientPool,
 		queue:      workqueue.New(),
+		groupVersionResources: groupVersionResources,
+		opCache:               operationNotSupportedCache{},
+		finalizerToken:        finalizerToken,
 	}
 
 	// configure the backing store/controller
@@ -144,7 +159,7 @@ func (nm *NamespaceController) syncNamespaceFromKey(key string) (err error) {
 		return err
 	}
 	namespace := obj.(*api.Namespace)
-	return syncNamespace(nm.kubeClient, nm.versions, namespace)
+	return syncNamespace(nm.kubeClient, nm.clientPool, nm.opCache, nm.groupVersionResources, namespace, nm.finalizerToken)
 }
 
 // Run starts observing the system with the specified number of workers.

test/e2e/namespace.go

@@ -23,6 +23,8 @@ import (
 	"time"
 
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/errors"
+	"k8s.io/kubernetes/pkg/util/intstr"
 	"k8s.io/kubernetes/pkg/util/wait"
 
 	. "github.com/onsi/ginkgo"
@@ -74,6 +76,109 @@ func extinguish(f *Framework, totalNS int, maxAllowedAfterDel int, maxSeconds in
 		}))
 }
 
+func ensurePodsAreRemovedWhenNamespaceIsDeleted(f *Framework) {
+	var err error
+
+	By("Creating a test namespace")
+	namespace, err := f.CreateNamespace("nsdeletetest", nil)
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Waiting for a default service account to be provisioned in namespace")
+	err = waitForDefaultServiceAccountInNamespace(f.Client, namespace.Name)
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Creating a pod in the namespace")
+	pod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{
+			Name: "test-pod",
+		},
+		Spec: api.PodSpec{
+			Containers: []api.Container{
+				{
+					Name:  "nginx",
+					Image: "gcr.io/google_containers/pause:2.0",
+				},
+			},
+		},
+	}
+	pod, err = f.Client.Pods(namespace.Name).Create(pod)
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Waiting for the pod to have running status")
+	expectNoError(waitForPodRunningInNamespace(f.Client, pod.Name, pod.Namespace))
+
+	By("Deleting the namespace")
+	err = f.Client.Namespaces().Delete(namespace.Name)
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Waiting for the namespace to be removed.")
+	maxWaitSeconds := int64(60) + *pod.Spec.TerminationGracePeriodSeconds
+	expectNoError(wait.Poll(1*time.Second, time.Duration(maxWaitSeconds)*time.Second,
+		func() (bool, error) {
+			_, err = f.Client.Namespaces().Get(namespace.Name)
+			if err != nil && errors.IsNotFound(err) {
+				return true, nil
+			}
+			return false, nil
+		}))
+
+	By("Verifying there is no pod in the namespace")
+	_, err = f.Client.Pods(namespace.Name).Get(pod.Name)
+	Expect(err).To(HaveOccurred())
+}
+
+func ensureServicesAreRemovedWhenNamespaceIsDeleted(f *Framework) {
+	var err error
+
+	By("Creating a test namespace")
+	namespace, err := f.CreateNamespace("nsdeletetest", nil)
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Waiting for a default service account to be provisioned in namespace")
+	err = waitForDefaultServiceAccountInNamespace(f.Client, namespace.Name)
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Creating a service in the namespace")
+	serviceName := "test-service"
+	labels := map[string]string{
+		"foo": "bar",
+		"baz": "blah",
+	}
+	service := &api.Service{
+		ObjectMeta: api.ObjectMeta{
+			Name: serviceName,
+		},
+		Spec: api.ServiceSpec{
+			Selector: labels,
+			Ports: []api.ServicePort{{
+				Port:       80,
+				TargetPort: intstr.FromInt(80),
+			}},
+		},
+	}
+	service, err = f.Client.Services(namespace.Name).Create(service)
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Deleting the namespace")
+	err = f.Client.Namespaces().Delete(namespace.Name)
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Waiting for the namespace to be removed.")
+	maxWaitSeconds := int64(60)
+	expectNoError(wait.Poll(1*time.Second, time.Duration(maxWaitSeconds)*time.Second,
+		func() (bool, error) {
+			_, err = f.Client.Namespaces().Get(namespace.Name)
+			if err != nil && errors.IsNotFound(err) {
+				return true, nil
+			}
+			return false, nil
+		}))
+
+	By("Verifying there is no service in the namespace")
+	_, err = f.Client.Services(namespace.Name).Get(service.Name)
+	Expect(err).To(HaveOccurred())
+}
+
 // This test must run [Serial] due to the impact of running other parallel
 // tests can have on its performance.  Each test that follows the common
 // test framework follows this pattern:
@@ -106,10 +211,17 @@ var _ = Describe("Namespaces [Serial]", func() {
 
 	f := NewDefaultFramework("namespaces")
 
+	It("should ensure that all pods are removed when a namespace is deleted.",
+		func() { ensurePodsAreRemovedWhenNamespaceIsDeleted(f) })
+
+	It("should ensure that all services are removed when a namespace is deleted.",
+		func() { ensureServicesAreRemovedWhenNamespaceIsDeleted(f) })
+
 	It("should delete fast enough (90 percent of 100 namespaces in 150 seconds)",
 		func() { extinguish(f, 100, 10, 150) })
 
 	// On hold until etcd3; see #7372
 	It("should always delete fast (ALL of 100 namespaces in 150 seconds) [Feature:ComprehensiveNamespaceDraining]",
 		func() { extinguish(f, 100, 0, 150) })
+
 })