Merge pull request #25991 from quinton-hoole/fed-service-controller

federation/pkg/dnsprovider/providers/google/clouddns/clouddns.go

@@ -30,6 +30,7 @@ import (
 
 	"k8s.io/kubernetes/federation/pkg/dnsprovider"
 	"k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns/internal"
+	"k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns/internal/stubs"
 	"k8s.io/kubernetes/pkg/cloudprovider/providers/gce"
 )
 
@@ -99,3 +100,17 @@ func CreateInterface(projectID string, tokenSource oauth2.TokenSource) (*Interfa
 	glog.Infof("Successfully got DNS service: %v\n", service)
 	return newInterfaceWithStub(projectID, internal.NewService(service)), nil
 }
+
+// NewFakeInterface returns a fake clouddns interface, useful for unit testing purposes.
+func NewFakeInterface() (dnsprovider.Interface, error) {
+	service := stubs.NewService()
+	interface_ := newInterfaceWithStub("", service)
+	zones := service.ManagedZones_
+	// Add a fake zone to test against.
+	zone := &stubs.ManagedZone{zones, "example.com", []stubs.ResourceRecordSet{}}
+	call := zones.Create(interface_.project(), zone)
+	if _, err := call.Do(); err != nil {
+		return nil, err
+	}
+	return interface_, nil
+}

federation/pkg/dnsprovider/providers/google/clouddns/clouddns_test.go

@@ -23,7 +23,6 @@ import (
 	"testing"
 
 	"k8s.io/kubernetes/federation/pkg/dnsprovider"
-	"k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns/internal/stubs"
 	"k8s.io/kubernetes/federation/pkg/dnsprovider/rrstype"
 )
 
@@ -31,21 +30,7 @@ func newTestInterface() (dnsprovider.Interface, error) {
 	// Use this to test the real cloud service - insert appropriate project-id.  Default token source will be used.  See
 	// https://github.com/golang/oauth2/blob/master/google/default.go for details.
 	// i, err := dnsprovider.GetDnsProvider(ProviderName, strings.NewReader("\n[global]\nproject-id = federation0-cluster00"))
-	return newFakeInterface() // Use this to stub out the entire cloud service
-}
-
-func newFakeInterface() (dnsprovider.Interface, error) {
-	service := stubs.NewService()
-	interface_ := newInterfaceWithStub("", service)
-	zones := service.ManagedZones_
-	// Add a fake zone to test against.
-	zone := &stubs.ManagedZone{zones, "example.com", []stubs.ResourceRecordSet{}}
-	call := zones.Create(interface_.project(), zone)
-	_, err := call.Do()
-	if err != nil {
-		return nil, err
-	}
-	return interface_, nil
+	return NewFakeInterface() // Use this to stub out the entire cloud service
 }
 
 var interface_ dnsprovider.Interface

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

@@ -42,7 +42,7 @@ func (sc *ServiceController) clusterEndpointWorker() {
 						return
 					}
 					defer cache.endpointQueue.Done(key)
-					err := sc.clusterCache.syncEndpoint(key.(string), clusterName, cache, sc.serviceCache, fedClient)
+					err := sc.clusterCache.syncEndpoint(key.(string), clusterName, cache, sc.serviceCache, fedClient, sc)
 					if err != nil {
 						glog.V(2).Infof("Failed to sync endpoint: %+v", err)
 					}
@@ -54,7 +54,7 @@ func (sc *ServiceController) clusterEndpointWorker() {
 
 // Whenever there is change on endpoint, the federation service should be updated
 // key is the namespaced name of endpoint
-func (cc *clusterClientCache) syncEndpoint(key, clusterName string, clusterCache *clusterCache, serviceCache *serviceCache, fedClient federationclientset.Interface) error {
+func (cc *clusterClientCache) syncEndpoint(key, clusterName string, clusterCache *clusterCache, serviceCache *serviceCache, fedClient federationclientset.Interface, serviceController *ServiceController) error {
 	cachedService, ok := serviceCache.get(key)
 	if !ok {
 		// here we filtered all non-federation services
@@ -70,19 +70,19 @@ func (cc *clusterClientCache) syncEndpoint(key, clusterName string, clusterCache
 		endpoint, ok := endpointInterface.(*api.Endpoints)
 		if ok {
 			glog.V(4).Infof("Found endpoint for federation service %s/%s from cluster %s", endpoint.Namespace, endpoint.Name, clusterName)
-			err = cc.processEndpointUpdate(cachedService, endpoint, clusterName)
+			err = cc.processEndpointUpdate(cachedService, endpoint, clusterName, serviceController)
 		} else {
 			_, ok := endpointInterface.(cache.DeletedFinalStateUnknown)
 			if !ok {
 				return fmt.Errorf("Object contained wasn't a service or a deleted key: %+v", endpointInterface)
 			}
 			glog.Infof("Found tombstone for %v", key)
-			err = cc.processEndpointDeletion(cachedService, clusterName)
+			err = cc.processEndpointDeletion(cachedService, clusterName, serviceController)
 		}
 	} else {
 		// service absence in store means watcher caught the deletion, ensure LB info is cleaned
 		glog.Infof("Can not get endpoint %v for cluster %s from endpointStore", key, clusterName)
-		err = cc.processEndpointDeletion(cachedService, clusterName)
+		err = cc.processEndpointDeletion(cachedService, clusterName, serviceController)
 	}
 	if err != nil {
 		glog.Errorf("Failed to sync service: %+v, put back to service queue", err)
@@ -92,7 +92,7 @@ func (cc *clusterClientCache) syncEndpoint(key, clusterName string, clusterCache
 	return nil
 }
 
-func (cc *clusterClientCache) processEndpointDeletion(cachedService *cachedService, clusterName string) error {
+func (cc *clusterClientCache) processEndpointDeletion(cachedService *cachedService, clusterName string, serviceController *ServiceController) error {
 	glog.V(4).Infof("Processing endpoint update for %s/%s, cluster %s", cachedService.lastState.Namespace, cachedService.lastState.Name, clusterName)
 	var err error
 	cachedService.rwlock.Lock()
@@ -103,13 +103,12 @@ func (cc *clusterClientCache) processEndpointDeletion(cachedService *cachedServi
 	if ok {
 		// endpoints lost, clean dns record
 		glog.V(4).Infof("Cached endpoint was not found for %s/%s, cluster %s, building one", cachedService.lastState.Namespace, cachedService.lastState.Name, clusterName)
-		// TODO: need to integrate with dns.go:ensureDNSRecords
 		for i := 0; i < clientRetryCount; i++ {
-			err := ensureDNSRecords(clusterName, cachedService)
-			if err == nil {
+			if err := serviceController.ensureDnsRecords(clusterName, cachedService); err == nil {
 				delete(cachedService.endpointMap, clusterName)
 				return nil
 			}
+			glog.V(4).Infof("Error ensuring DNS Records: %v", err)
 			time.Sleep(cachedService.nextDNSUpdateDelay())
 		}
 	}
@@ -118,7 +117,7 @@ func (cc *clusterClientCache) processEndpointDeletion(cachedService *cachedServi
 
 // Update dns info when endpoint update event received
 // We do not care about the endpoint info, what we need to make sure here is len(endpoints.subsets)>0
-func (cc *clusterClientCache) processEndpointUpdate(cachedService *cachedService, endpoint *api.Endpoints, clusterName string) error {
+func (cc *clusterClientCache) processEndpointUpdate(cachedService *cachedService, endpoint *api.Endpoints, clusterName string, serviceController *ServiceController) error {
 	glog.V(4).Infof("Processing endpoint update for %s/%s, cluster %s", endpoint.Namespace, endpoint.Name, clusterName)
 	cachedService.rwlock.Lock()
 	defer cachedService.rwlock.Unlock()
@@ -132,12 +131,11 @@ func (cc *clusterClientCache) processEndpointUpdate(cachedService *cachedService
 		// first time get endpoints, update dns record
 		glog.V(4).Infof("Cached endpoint was not found for %s/%s, cluster %s, building one", endpoint.Namespace, endpoint.Name, clusterName)
 		cachedService.endpointMap[clusterName] = 1
-		err := ensureDNSRecords(clusterName, cachedService)
-		if err != nil {
-			// TODO: need to integrate with dns.go:ensureDNSRecords
+		if err := serviceController.ensureDnsRecords(clusterName, cachedService); err != nil {
+			glog.V(4).Infof("Error ensuring DNS Records: %v", err)
 			for i := 0; i < clientRetryCount; i++ {
 				time.Sleep(cachedService.nextDNSUpdateDelay())
-				err := ensureDNSRecords(clusterName, cachedService)
+				err := serviceController.ensureDnsRecords(clusterName, cachedService)
 				if err == nil {
 					return nil
 				}

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

@@ -19,9 +19,25 @@ package service
 import (
 	"testing"
 
+	"k8s.io/kubernetes/federation/apis/federation"
+	"k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns" // Only for unit testing purposes.
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/util/sets"
 )
 
+var fakeDns, _ = clouddns.NewFakeInterface() // No need to check for unsupported interfaces, as the fake interface supports everything that's required.
+var fakeDnsZones, _ = fakeDns.Zones()
+
+var fakeServiceController = ServiceController{
+	dns:          fakeDns,
+	dnsZones:     fakeDnsZones,
+	serviceCache: &serviceCache{fedServiceMap: make(map[string]*cachedService)},
+	clusterCache: &clusterClientCache{
+		clientMap: make(map[string]*clusterCache),
+	},
+	knownClusterSet: make(sets.String),
+}
+
 func buildEndpoint(subsets [][]string) *api.Endpoints {
 	endpoint := &api.Endpoints{
 		Subsets: []api.EndpointSubset{
@@ -69,8 +85,9 @@ func TestProcessEndpointUpdate(t *testing.T) {
 			1,
 		},
 	}
+
 	for _, test := range tests {
-		cc.processEndpointUpdate(test.cachedService, test.endpoint, test.clusterName)
+		cc.processEndpointUpdate(test.cachedService, test.endpoint, test.clusterName, &fakeServiceController)
 		if test.expectResult != test.cachedService.endpointMap[test.clusterName] {
 			t.Errorf("Test failed for %s, expected %v, saw %v", test.name, test.expectResult, test.cachedService.endpointMap[test.clusterName])
 		}
@@ -78,8 +95,18 @@ func TestProcessEndpointUpdate(t *testing.T) {
 }
 
 func TestProcessEndpointDeletion(t *testing.T) {
+	clusterName := "foo"
 	cc := clusterClientCache{
-		clientMap: make(map[string]*clusterCache),
+		clientMap: map[string]*clusterCache{
+			clusterName: {
+				cluster: &federation.Cluster{
+					Status: federation.ClusterStatus{
+						Zones:  []string{"foozone"},
+						Region: "fooregion",
+					},
+				},
+			},
+		},
 	}
 	tests := []struct {
 		name          string
@@ -95,7 +122,7 @@ func TestProcessEndpointDeletion(t *testing.T) {
 				endpointMap: make(map[string]int),
 			},
 			buildEndpoint([][]string{{"ip1", ""}}),
-			"foo",
+			clusterName,
 			0,
 		},
 		{
@@ -103,16 +130,17 @@ func TestProcessEndpointDeletion(t *testing.T) {
 			&cachedService{
 				lastState: &api.Service{},
 				endpointMap: map[string]int{
-					"foo": 1,
+					clusterName: 1,
 				},
 			},
 			buildEndpoint([][]string{{"ip1", ""}}),
-			"foo",
+			clusterName,
 			0,
 		},
 	}
+	fakeServiceController.clusterCache = &cc
 	for _, test := range tests {
-		cc.processEndpointDeletion(test.cachedService, test.clusterName)
+		cc.processEndpointDeletion(test.cachedService, test.clusterName, &fakeServiceController)
 		if test.expectResult != test.cachedService.endpointMap[test.clusterName] {
 			t.Errorf("Test failed for %s, expected %v, saw %v", test.name, test.expectResult, test.cachedService.endpointMap[test.clusterName])
 		}

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

@@ -78,14 +78,10 @@ type cachedService struct {
 	// key clusterName
 	// value is a flag that if there is ready address, 1 means there is ready address, 0 means no ready address
 	endpointMap map[string]int
-	// cluster service map hold serivice status info from kubernetes clusters
-	// key clusterName
-
+	// serviceStatusMap map holds service status info from kubernetes clusters, keyed on clusterName
 	serviceStatusMap map[string]api.LoadBalancerStatus
-
 	// Ensures only one goroutine can operate on this service at any given time.
 	rwlock sync.Mutex
-
 	// Controls error back-off for procceeding federation service to k8s clusters
 	lastRetryDelay time.Duration
 	// Controls error back-off for updating federation service back to federation apiserver
@@ -104,9 +100,11 @@ type serviceCache struct {
 type ServiceController struct {
 	dns              dnsprovider.Interface
 	federationClient federationclientset.Interface
-	zones            []dnsprovider.Zone
-	serviceCache     *serviceCache
-	clusterCache     *clusterClientCache
+	federationName   string
+	// each federation should be configured with a single zone (e.g. "mycompany.com")
+	dnsZones     dnsprovider.Zones
+	serviceCache *serviceCache
+	clusterCache *clusterClientCache
 	// A store of services, populated by the serviceController
 	serviceStore cache.StoreToServiceLister
 	// Watches changes to all services
@@ -690,7 +688,7 @@ func (s *ServiceController) lockedUpdateDNSRecords(service *cachedService, clust
 	for key := range s.clusterCache.clientMap {
 		for _, clusterName := range clusterNames {
 			if key == clusterName {
-				ensureDNSRecords(clusterName, service)
+				s.ensureDnsRecords(clusterName, service)
 			}
 		}
 	}

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

@@ -17,22 +17,38 @@ limitations under the License.
 package service
 
 import (
+	"sync"
 	"testing"
 
 	"k8s.io/kubernetes/federation/apis/federation"
+	"k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns" // Only for unit testing purposes.
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/util/sets"
 )
 
 func TestGetClusterConditionPredicate(t *testing.T) {
+	fakedns, _ := clouddns.NewFakeInterface() // No need to check for unsupported interfaces, as the fake interface supports everything that's required.
+	serviceController := ServiceController{
+		dns:          fakedns,
+		serviceCache: &serviceCache{fedServiceMap: make(map[string]*cachedService)},
+		clusterCache: &clusterClientCache{
+			rwlock:    sync.Mutex{},
+			clientMap: make(map[string]*clusterCache),
+		},
+		knownClusterSet: make(sets.String),
+	}
+
 	tests := []struct {
-		cluster      federation.Cluster
-		expectAccept bool
-		name         string
+		cluster           federation.Cluster
+		expectAccept      bool
+		name              string
+		serviceController *ServiceController
 	}{
 		{
-			cluster:      federation.Cluster{},
-			expectAccept: false,
-			name:         "empty",
+			cluster:           federation.Cluster{},
+			expectAccept:      false,
+			name:              "empty",
+			serviceController: &serviceController,
 		},
 		{
 			cluster: federation.Cluster{
@@ -42,8 +58,9 @@ func TestGetClusterConditionPredicate(t *testing.T) {
 					},
 				},
 			},
-			expectAccept: true,
-			name:         "basic",
+			expectAccept:      true,
+			name:              "basic",
+			serviceController: &serviceController,
 		},
 		{
 			cluster: federation.Cluster{
@@ -53,8 +70,9 @@ func TestGetClusterConditionPredicate(t *testing.T) {
 					},
 				},
 			},
-			expectAccept: false,
-			name:         "notready",
+			expectAccept:      false,
+			name:              "notready",
+			serviceController: &serviceController,
 		},
 	}
 	pred := getClusterConditionPredicate()