kubernetes · k8s-github-robot · Jun 14, 2016 · Jun 12, 2016 · Jun 13, 2016 · Jun 13, 2016
diff --git a/federation/cluster/common.sh b/federation/cluster/common.sh
@@ -94,6 +94,7 @@ function create-federation-api-objects {
     export FEDERATION_API_NODEPORT=32111
     export FEDERATION_NAMESPACE
     export FEDERATION_NAME="${FEDERATION_NAME:-federation}"
+    export DNS_ZONE_NAME="${DNS_ZONE_NAME:-example.com}"
 
     template="go run ${KUBE_ROOT}/federation/cluster/template.go"
 

diff --git a/federation/cmd/federation-controller-manager/app/controllermanager.go b/federation/cmd/federation-controller-manager/app/controllermanager.go
@@ -121,7 +121,7 @@ func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) err
 		glog.Fatalf("Cloud provider could not be initialized: %v", err)
 	}
 	scClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, servicecontroller.UserAgentName))
-	servicecontroller := servicecontroller.New(scClientset, dns)
+	servicecontroller := servicecontroller.New(scClientset, dns, s.FederationName, s.ZoneName)
 	if err := servicecontroller.Run(s.ConcurrentServiceSyncs, wait.NeverStop); err != nil {
 		glog.Errorf("Failed to start service controller: %v", err)
 	}

diff --git a/federation/cmd/federation-controller-manager/app/options/options.go b/federation/cmd/federation-controller-manager/app/options/options.go
@@ -35,6 +35,10 @@ type ControllerManagerConfiguration struct {
 	Port int `json:"port"`
 	// address is the IP address to serve on (set to 0.0.0.0 for all interfaces).
 	Address string `json:"address"`
+	// federation name.
+	FederationName string `json:"federationName"`
+	// zone name, like example.com.
+	ZoneName string `json:"zoneName"`
 	// dnsProvider is the provider for dns services.
 	DnsProvider string `json:"dnsProvider"`
 	// dnsConfigFile is the path to the dns provider configuration file.
@@ -90,6 +94,8 @@ func NewCMServer() *CMServer {
 func (s *CMServer) AddFlags(fs *pflag.FlagSet) {
 	fs.IntVar(&s.Port, "port", s.Port, "The port that the controller-manager's http service runs on")
 	fs.Var(componentconfig.IPVar{Val: &s.Address}, "address", "The IP address to serve on (set to 0.0.0.0 for all interfaces)")
+	fs.StringVar(&s.FederationName, "federation-name", s.FederationName, "Federation name.")
+	fs.StringVar(&s.ZoneName, "zone-name", s.ZoneName, "Zone name, like example.com.")
 	fs.IntVar(&s.ConcurrentServiceSyncs, "concurrent-service-syncs", s.ConcurrentServiceSyncs, "The number of service syncing operations that will be done concurrently. Larger number = faster endpoint updating, but more CPU (and network) load")
 	fs.DurationVar(&s.ClusterMonitorPeriod.Duration, "cluster-monitor-period", s.ClusterMonitorPeriod.Duration, "The period for syncing ClusterStatus in ClusterController.")
 	fs.BoolVar(&s.EnableProfiling, "profiling", true, "Enable profiling via web interface host:port/debug/pprof/")

diff --git a/federation/manifests/federation-controller-manager-deployment.yaml b/federation/manifests/federation-controller-manager-deployment.yaml
@@ -21,6 +21,8 @@ spec:
         - --master=https://{{.FEDERATION_APISERVER_DEPLOYMENT_NAME}}:443
         - --dns-provider={{.FEDERATION_DNS_PROVIDER}}
         - --dns-provider-config={{.FEDERATION_DNS_PROVIDER_CONFIG}}
+        - --federation-name={{.FEDERATION_NAME}}
+        - --zone-name={{.DNS_ZONE_NAME}}
         ports:
         - containerPort: 443
           name: https

diff --git a/federation/pkg/dnsprovider/providers/google/clouddns/rrsets.go b/federation/pkg/dnsprovider/providers/google/clouddns/rrsets.go
@@ -36,7 +36,7 @@ func (rrsets ResourceRecordSets) List() ([]dnsprovider.ResourceRecordSet, error)
 	}
 	list := make([]dnsprovider.ResourceRecordSet, len(response.Rrsets()))
 	for i, rrset := range response.Rrsets() {
-		list[i] = &ResourceRecordSet{rrset, &rrsets}
+		list[i] = ResourceRecordSet{rrset, &rrsets}
 	}
 	return list, nil
 }

diff --git a/federation/pkg/federation-controller/service/dns.go b/federation/pkg/federation-controller/service/dns.go
@@ -85,8 +85,7 @@ func (s *ServiceController) getClusterZoneNames(clusterName string) (zones []str
 
 // getFederationDNSZoneName returns the name of the managed DNS Zone configured for this federation
 func (s *ServiceController) getFederationDNSZoneName() (string, error) {
-	return "example.com", nil // TODO: quinton: Get this from the federation configuration.
-	// Note: For unit testing this must match the domain populated in the test/stub dnsprovider.
+	return s.zoneName, nil
 }
 
 // getDnsZone is a hack around the fact that dnsprovider does not yet support a Get() method, only a List() method.  TODO:  Fix that.
@@ -147,7 +146,7 @@ func getResolvedEndpoints(endpoints []string) ([]string, error) {
 /* ensureDnsRrsets ensures (idempotently, and with minimum mutations) that all of the DNS resource record sets for dnsName are consistent with endpoints.
    if endpoints is nil or empty, a CNAME record to uplevelCname is ensured.
 */
-func (s *ServiceController) ensureDnsRrsets(dnsZoneName, dnsName string, endpoints []string, uplevelCname string) error {
+func (s *ServiceController) ensureDnsRrsets(dnsZoneName, dnsName string, endpoints []string, uplevelCname string, endpointReachable bool) error {
 	dnsZone, err := getDnsZone(dnsZoneName, s.dnsZones)
 	if err != nil {
 		return err
@@ -161,37 +160,44 @@ func (s *ServiceController) ensureDnsRrsets(dnsZoneName, dnsName string, endpoin
 		return err
 	}
 	if rrset == nil {
-		// It doesn't exist yet, so create it, if we indeed have healthy endpoints
-		if len(endpoints) < 1 {
-			// There are no endpoint addresses at this level, so CNAME to uplevel, if provided
-			if uplevelCname != "" {
-				newRrset := rrsets.New(dnsName, []string{uplevelCname}, minDnsTtl, rrstype.CNAME)
+		if endpointReachable {
+			// It doesn't exist yet, so create it, if we indeed have healthy endpoints
+			if len(endpoints) < 1 {
+				// There are no endpoint addresses at this level, so CNAME to uplevel, if provided
+				if uplevelCname != "" {
+					newRrset := rrsets.New(dnsName, []string{uplevelCname}, minDnsTtl, rrstype.CNAME)
+					rrset, err = rrsets.Add(newRrset)
+					if err != nil {
+						return err
+					}
+				}
+				// else we want no record, and we have no record, so we're all good.
+			} else {
+				// We have valid endpoint addresses, so just add them as A records.
+				// But first resolve DNS names, as some cloud providers (like AWS) expose
+				// load balancers behind DNS names, not IP addresses.
+				resolvedEndpoints, err := getResolvedEndpoints(endpoints)
+				if err != nil {
+					return err // TODO: We could potentially add the ones we did get back, even if some of them failed to resolve.
+				}
+				newRrset := rrsets.New(dnsName, resolvedEndpoints, minDnsTtl, rrstype.A)
 				rrset, err = rrsets.Add(newRrset)
 				if err != nil {
 					return err
 				}
 			}
-			// else we want no record, and we have no record, so we're all good.
-		} else {
-			// We have valid endpoint addresses, so just add them as A records.
-			// But first resolve DNS names, as some cloud providers (like AWS) expose
-			// load balancers behind DNS names, not IP addresses.
-			resolvedEndpoints, err := getResolvedEndpoints(endpoints)
-			if err != nil {
-				return err // TODO: We could potentially add the ones we did get back, even if some of them failed to resolve.
-			}
-			newRrset := rrsets.New(dnsName, resolvedEndpoints, minDnsTtl, rrstype.A)
-			rrset, err = rrsets.Add(newRrset)
-			if err != nil {
-				return err
-			}
 		}
 	} else {
 		// the rrset already exists, so make it right.
 		if len(endpoints) < 1 {
 			// Need an appropriate CNAME record.  Check that we have it.
 			newRrset := rrsets.New(dnsName, []string{uplevelCname}, minDnsTtl, rrstype.CNAME)
 			if rrset == newRrset {
+				if !endpointReachable {
+					if err = rrsets.Remove(rrset); err != nil {
+						return err
+					}
+				}
 				// The existing rrset is equal to the required one - our work is done here
 				return nil
 			} else {
@@ -200,7 +206,7 @@ func (s *ServiceController) ensureDnsRrsets(dnsZoneName, dnsName string, endpoin
 				if err = rrsets.Remove(rrset); err != nil {
 					return err
 				}
-				if uplevelCname != "" {
+				if uplevelCname != "" && endpointReachable {
 					if _, err = rrsets.Add(newRrset); err != nil {
 						return err
 					}
@@ -215,6 +221,11 @@ func (s *ServiceController) ensureDnsRrsets(dnsZoneName, dnsName string, endpoin
 			}
 			newRrset := rrsets.New(dnsName, resolvedEndpoints, minDnsTtl, rrstype.A)
 			if rrset == newRrset {
+				if !endpointReachable {
+					if err = rrsets.Remove(rrset); err != nil {
+						return err
+					}
+				}
 				// The existing rrset is equal to the required one - our work is done here
 				// TODO: We could be more thorough about checking for equivalence to avoid unnecessary updates, but in the
 				//       worst case we'll just replace what's there with an equivalent, if not exactly identical record set.
@@ -225,8 +236,10 @@ func (s *ServiceController) ensureDnsRrsets(dnsZoneName, dnsName string, endpoin
 				if err = rrsets.Remove(rrset); err != nil {
 					return err
 				}
-				if _, err = rrsets.Add(newRrset); err != nil {
-					return err
+				if endpointReachable {
+					if _, err = rrsets.Add(newRrset); err != nil {
+						return err
+					}
 				}
 			}
 		}
@@ -254,12 +267,12 @@ func (s *ServiceController) ensureDnsRecords(clusterName string, cachedService *
 	// the state of the service when we last successfully sync'd it's DNS records.
 	// So this time around we only need to patch that (add new records, remove deleted records, and update changed records.
 	//
-	if s.dns == nil {
-		return nil
-	}
 	if s == nil {
 		return fmt.Errorf("nil ServiceController passed to ServiceController.ensureDnsRecords(clusterName: %s, cachedService: %v)", clusterName, cachedService)
 	}
+	if s.dns == nil {
+		return nil
+	}
 	if cachedService == nil {
 		return fmt.Errorf("nil cachedService passed to ServiceController.ensureDnsRecords(clusterName: %s, cachedService: %v)", clusterName, cachedService)
 	}
@@ -280,7 +293,7 @@ func (s *ServiceController) ensureDnsRecords(clusterName string, cachedService *
 	if err != nil {
 		return err
 	}
-
+	_, endpointReachable := cachedService.endpointMap[clusterName]
 	commonPrefix := serviceName + "." + namespaceName + "." + s.federationName + ".svc"
 	// dnsNames is the path up the DNS search tree, starting at the leaf
 	dnsNames := []string{
@@ -293,7 +306,7 @@ func (s *ServiceController) ensureDnsRecords(clusterName string, cachedService *
 	endpoints := [][]string{zoneEndpoints, regionEndpoints, globalEndpoints}
 
 	for i, endpoint := range endpoints {
-		if err = s.ensureDnsRrsets(dnsZoneName, dnsNames[i], endpoint, dnsNames[i+1]); err != nil {
+		if err = s.ensureDnsRrsets(dnsZoneName, dnsNames[i], endpoint, dnsNames[i+1], endpointReachable); err != nil {
 			return err
 		}
 	}

diff --git a/federation/pkg/federation-controller/service/endpoint_helper.go b/federation/pkg/federation-controller/service/endpoint_helper.go
@@ -93,7 +93,7 @@ func (cc *clusterClientCache) syncEndpoint(key, clusterName string, clusterCache
 }
 
 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)
+	glog.V(4).Infof("Processing endpoint deletion for %s/%s, cluster %s", cachedService.lastState.Namespace, cachedService.lastState.Name, clusterName)
 	var err error
 	cachedService.rwlock.Lock()
 	defer cachedService.rwlock.Unlock()
@@ -102,10 +102,10 @@ func (cc *clusterClientCache) processEndpointDeletion(cachedService *cachedServi
 	// need to query dns info from dnsprovider and make sure of if deletion is needed
 	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)
+		glog.V(4).Infof("Cached endpoint was found for %s/%s, cluster %s, removing", cachedService.lastState.Namespace, cachedService.lastState.Name, clusterName)
+		delete(cachedService.endpointMap, clusterName)
 		for i := 0; i < clientRetryCount; i++ {
 			if err := serviceController.ensureDnsRecords(clusterName, cachedService); err == nil {
-				delete(cachedService.endpointMap, clusterName)
 				return nil
 			}
 			glog.V(4).Infof("Error ensuring DNS Records: %v", err)
@@ -120,27 +120,54 @@ func (cc *clusterClientCache) processEndpointDeletion(cachedService *cachedServi
 func (cc *clusterClientCache) processEndpointUpdate(cachedService *cachedService, endpoint *v1.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()
+	var reachable bool
 	defer cachedService.rwlock.Unlock()
-	for _, subset := range endpoint.Subsets {
-		if len(subset.Addresses) > 0 {
-			cachedService.endpointMap[clusterName] = 1
-		}
-	}
 	_, ok := cachedService.endpointMap[clusterName]
 	if !ok {
-		// 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
-		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 := serviceController.ensureDnsRecords(clusterName, cachedService)
-				if err == nil {
-					return nil
+		for _, subset := range endpoint.Subsets {
+			if len(subset.Addresses) > 0 {
+				reachable = true
+				break
+			}
+		}
+		if reachable {
+			// first time get endpoints, update dns record
+			glog.V(4).Infof("Reachable endpoint was found for %s/%s, cluster %s, building endpointMap", endpoint.Namespace, endpoint.Name, clusterName)
+			cachedService.endpointMap[clusterName] = 1
+			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 := serviceController.ensureDnsRecords(clusterName, cachedService)
+					if err == nil {
+						return nil
+					}
+				}
+				return err
+			}
+		}
+	} else {
+		for _, subset := range endpoint.Subsets {
+			if len(subset.Addresses) > 0 {
+				reachable = true
+				break
+			}
+		}
+		if !reachable {
+			// first time get endpoints, update dns record
+			glog.V(4).Infof("Reachable endpoint was lost for %s/%s, cluster %s, deleting endpointMap", endpoint.Namespace, endpoint.Name, clusterName)
+			delete(cachedService.endpointMap, clusterName)
+			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 := serviceController.ensureDnsRecords(clusterName, cachedService)
+					if err == nil {
+						return nil
+					}
 				}
+				return err
 			}
-			return err
 		}
 	}
 	return nil

diff --git a/federation/pkg/federation-controller/service/endpoint_helper_test.go b/federation/pkg/federation-controller/service/endpoint_helper_test.go
@@ -29,9 +29,11 @@ var fakeDns, _ = clouddns.NewFakeInterface() // No need to check for unsupported
 var fakeDnsZones, _ = fakeDns.Zones()
 
 var fakeServiceController = ServiceController{
-	dns:          fakeDns,
-	dnsZones:     fakeDnsZones,
-	serviceCache: &serviceCache{fedServiceMap: make(map[string]*cachedService)},
+	dns:            fakeDns,
+	dnsZones:       fakeDnsZones,
+	federationName: "fed1",
+	zoneName:       "example.com",
+	serviceCache:   &serviceCache{fedServiceMap: make(map[string]*cachedService)},
 	clusterCache: &clusterClientCache{
 		clientMap: make(map[string]*clusterCache),
 	},
@@ -52,8 +54,18 @@ func buildEndpoint(subsets [][]string) *v1.Endpoints {
 }
 
 func TestProcessEndpointUpdate(t *testing.T) {
+	clusterName := "foo"
 	cc := clusterClientCache{
-		clientMap: make(map[string]*clusterCache),
+		clientMap: map[string]*clusterCache{
+			clusterName: {
+				cluster: &v1alpha1.Cluster{
+					Status: v1alpha1.ClusterStatus{
+						Zones:  []string{"foozone"},
+						Region: "fooregion",
+					},
+				},
+			},
+		},
 	}
 	tests := []struct {
 		name          string
@@ -69,7 +81,7 @@ func TestProcessEndpointUpdate(t *testing.T) {
 				endpointMap: make(map[string]int),
 			},
 			buildEndpoint([][]string{{"ip1", ""}}),
-			"foo",
+			clusterName,
 			1,
 		},
 		{
@@ -81,11 +93,11 @@ func TestProcessEndpointUpdate(t *testing.T) {
 				},
 			},
 			buildEndpoint([][]string{{"ip1", ""}}),
-			"foo",
+			clusterName,
 			1,
 		},
 	}
-
+	fakeServiceController.clusterCache = &cc
 	for _, test := range tests {
 		cc.processEndpointUpdate(test.cachedService, test.endpoint, test.clusterName, &fakeServiceController)
 		if test.expectResult != test.cachedService.endpointMap[test.clusterName] {