fed: Enable the namespace controller in integration tests

test/e2e_node/services/internal_services.go

@@ -20,6 +20,8 @@ import (
 	"io/ioutil"
 	"os"
 
+	"k8s.io/kubernetes/test/e2e/framework"
+
 	"github.com/golang/glog"
 )
 
@@ -128,7 +130,7 @@ func (es *e2eServices) startApiServer() error {
 // startNamespaceController starts the embedded namespace controller or returns an error.
 func (es *e2eServices) startNamespaceController() error {
 	glog.Info("Starting namespace controller")
-	es.nsController = NewNamespaceController()
+	es.nsController = NewNamespaceController(framework.TestContext.Host)
 	return es.nsController.Start()
 }
 

test/e2e_node/services/namespace_controller.go

@@ -26,7 +26,6 @@ import (
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	informers "k8s.io/kubernetes/pkg/client/informers/informers_generated/externalversions"
 	namespacecontroller "k8s.io/kubernetes/pkg/controller/namespace"
-	"k8s.io/kubernetes/test/e2e/framework"
 )
 
 const (
@@ -40,18 +39,19 @@ const (
 
 // NamespaceController is a server which manages namespace controller.
 type NamespaceController struct {
+	host   string
 	stopCh chan struct{}
 }
 
 // NewNamespaceController creates a new namespace controller.
-func NewNamespaceController() *NamespaceController {
+func NewNamespaceController(host string) *NamespaceController {
-	return &NamespaceController{stopCh: make(chan struct{})}
+	return &NamespaceController{host: host, stopCh: make(chan struct{})}
 }
 
 // Start starts the namespace controller.
 func (n *NamespaceController) Start() error {
 	// Use the default QPS
-	config := restclient.AddUserAgent(&restclient.Config{Host: framework.TestContext.Host}, ncName)
+	config := restclient.AddUserAgent(&restclient.Config{Host: n.host}, ncName)
 	client, err := clientset.NewForConfig(config)
 	if err != nil {
 		return err

test/integration/federation/framework/BUILD

@@ -28,6 +28,7 @@ go_library(
         "//federation/pkg/federation-controller/sync:go_default_library",
         "//pkg/client/clientset_generated/clientset:go_default_library",
         "//pkg/master:go_default_library",
+        "//test/e2e_node/services:go_default_library",
         "//test/integration/framework:go_default_library",
         "//vendor/github.com/pborman/uuid:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",

test/integration/federation/framework/federation.go

@@ -27,14 +27,16 @@ import (
 	clustercontroller "k8s.io/kubernetes/federation/pkg/federation-controller/cluster"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	"k8s.io/kubernetes/pkg/master"
+	"k8s.io/kubernetes/test/e2e_node/services"
 	"k8s.io/kubernetes/test/integration/framework"
 )
 
 type MemberCluster struct {
-	CloseFn framework.CloseFunc
+	CloseFn             framework.CloseFunc
-	Config  *master.Config
+	Config              *master.Config
-	Client  clientset.Interface
+	Client              clientset.Interface
-	Host    string
+	Host                string
+	namespaceController *services.NamespaceController
 }
 
 // FederationFixture manages a federation api server and a set of member clusters
@@ -42,10 +44,11 @@ type FederationFixture struct {
 	APIFixture          *FederationAPIFixture
 	DesiredClusterCount int
 	Clusters            []*MemberCluster
-	ClusterClients      []clientset.Interface
+
-	ClusterController   *clustercontroller.ClusterController
+	ClusterClients    []clientset.Interface
-	fedClient           federationclientset.Interface
+	ClusterController *clustercontroller.ClusterController
-	stopChan            chan struct{}
+	fedClient         federationclientset.Interface
+	stopChan          chan struct{}
 }
 
 func (f *FederationFixture) SetUp(t *testing.T) {
@@ -79,12 +82,18 @@ func (f *FederationFixture) StartCluster(t *testing.T) {
 
 	clusterClient := clientset.NewForConfigOrDie(config.GenericConfig.LoopbackClientConfig)
 	f.ClusterClients = append(f.ClusterClients, clusterClient)
-	f.Clusters = append(f.Clusters, &MemberCluster{
+	memberCluster := &MemberCluster{
-		CloseFn: closeFn,
+		CloseFn:             closeFn,
-		Config:  config,
+		Config:              config,
-		Client:  clusterClient,
+		Client:              clusterClient,
-		Host:    host,
+		Host:                host,
-	})
+		namespaceController: services.NewNamespaceController(host),
+	}
+	f.Clusters = append(f.Clusters, memberCluster)
+	err := memberCluster.namespaceController.Start()
+	if err != nil {
+		t.Fatal(err)
+	}
 
 	clusterId := len(f.ClusterClients)
 
@@ -115,6 +124,10 @@ func (f *FederationFixture) TearDown(t *testing.T) {
 		f.stopChan = nil
 	}
 	for _, cluster := range f.Clusters {
+		// Need to close controllers with active connections to the
+		// cluster api before stopping the api or the connections will
+		// hang until tcp timeout.
+		cluster.namespaceController.Stop()
 		cluster.CloseFn()
 	}
 	f.Clusters = nil