Make endpoint controller use framework

cmd/integration/integration.go

@@ -208,7 +208,7 @@ func startComponents(firstManifestURL, secondManifestURL, apiVersion string) (st
 
 	endpoints := service.NewEndpointController(cl)
 	// ensure the service endpoints are sync'd several times within the window that the integration tests wait
-	go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*4)
+	go endpoints.Run(3, util.NeverStop)
 
 	controllerManager := replicationControllerPkg.NewReplicationManager(cl)
 
@@ -285,7 +285,7 @@ func endpointsSet(c *client.Client, serviceNamespace, serviceID string, endpoint
 	return func() (bool, error) {
 		endpoints, err := c.Endpoints(serviceNamespace).Get(serviceID)
 		if err != nil {
-			glog.Infof("Error on creating endpoints: %v", err)
+			glog.Infof("Error getting endpoints: %v", err)
 			return false, nil
 		}
 		count := 0

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

@@ -50,6 +50,7 @@ type CMServer struct {
 	ClientConfig            client.Config
 	CloudProvider           string
 	CloudConfigFile         string
+	ConcurrentEndpointSyncs int
 	MinionRegexp            string
 	NodeSyncPeriod          time.Duration
 	ResourceQuotaSyncPeriod time.Duration
@@ -79,6 +80,7 @@ func NewCMServer() *CMServer {
 	s := CMServer{
 		Port:                    ports.ControllerManagerPort,
 		Address:                 util.IP(net.ParseIP("127.0.0.1")),
+		ConcurrentEndpointSyncs: 5,
 		NodeSyncPeriod:          10 * time.Second,
 		ResourceQuotaSyncPeriod: 10 * time.Second,
 		NamespaceSyncPeriod:     5 * time.Minute,
@@ -101,6 +103,7 @@ func (s *CMServer) AddFlags(fs *pflag.FlagSet) {
 	client.BindClientConfigFlags(fs, &s.ClientConfig)
 	fs.StringVar(&s.CloudProvider, "cloud_provider", s.CloudProvider, "The provider for cloud services.  Empty string for no provider.")
 	fs.StringVar(&s.CloudConfigFile, "cloud_config", s.CloudConfigFile, "The path to the cloud provider configuration file.  Empty string for no configuration file.")
+	fs.IntVar(&s.ConcurrentEndpointSyncs, "concurrent_endpoint_syncs", s.ConcurrentEndpointSyncs, "The number of endpoint syncing operations that will be done concurrently. Larger number = faster endpoint updating, but more CPU (and network) load")
 	fs.StringVar(&s.MinionRegexp, "minion_regexp", s.MinionRegexp, "If non empty, and --cloud_provider is specified, a regular expression for matching minion VMs.")
 	fs.DurationVar(&s.NodeSyncPeriod, "node_sync_period", s.NodeSyncPeriod, ""+
 		"The period for syncing nodes from cloudprovider. Longer periods will result in "+
@@ -171,7 +174,7 @@ func (s *CMServer) Run(_ []string) error {
 	}()
 
 	endpoints := service.NewEndpointController(kubeClient)
-	go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*10)
+	go endpoints.Run(s.ConcurrentEndpointSyncs, util.NeverStop)
 
 	controllerManager := replicationControllerPkg.NewReplicationManager(kubeClient)
 	controllerManager.Run(replicationControllerPkg.DefaultSyncPeriod)

cmd/kubernetes/kubernetes.go

@@ -139,7 +139,7 @@ func runControllerManager(machineList []string, cl *client.Client, nodeMilliCPU,
 	}
 
 	endpoints := service.NewEndpointController(cl)
-	go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*10)
+	go endpoints.Run(5, util.NeverStop)
 
 	controllerManager := controller.NewReplicationManager(cl)
 	controllerManager.Run(controller.DefaultSyncPeriod)

pkg/client/cache/store.go

@@ -18,6 +18,8 @@ package cache
 
 import (
 	"fmt"
+	"strings"
+
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
 )
 
@@ -67,6 +69,9 @@ type ExplicitKey string
 // keys for API objects which implement meta.Interface.
 // The key uses the format <namespace>/<name> unless <namespace> is empty, then
 // it's just <name>.
+//
+// TODO: replace key-as-string with a key-as-struct so that this
+// packing/unpacking won't be necessary.
 func MetaNamespaceKeyFunc(obj interface{}) (string, error) {
 	if key, ok := obj.(ExplicitKey); ok {
 		return string(key), nil
@@ -81,6 +86,25 @@ func MetaNamespaceKeyFunc(obj interface{}) (string, error) {
 	return meta.Name(), nil
 }
 
+// SplitMetaNamespaceKey returns the namespace and name that
+// MetaNamespaceKeyFunc encoded into key.
+//
+// TODO: replace key-as-string with a key-as-struct so that this
+// packing/unpacking won't be necessary.
+func SplitMetaNamespaceKey(key string) (namespace, name string, err error) {
+	parts := strings.Split(key, "/")
+	switch len(parts) {
+	case 1:
+		// name only, no namespace
+		return "", parts[0], nil
+	case 2:
+		// name and namespace
+		return parts[0], parts[1], nil
+	}
+
+	return "", "", fmt.Errorf("unexpected key format: %q", key)
+}
+
 // cache responsibilities are limited to:
 //	1. Computing keys for objects via keyFunc
 //  2. Invoking methods of a ThreadSafeStorage interface

pkg/util/set.go

@@ -91,6 +91,24 @@ func (s StringSet) Difference(s2 StringSet) StringSet {
 	return result
 }
 
+// Union returns a new set which includes items in either s1 or s2.
+// vof objects that are not in s2
+// For example:
+// s1 = {1, 2}
+// s2 = {3, 4}
+// s1.Union(s2) = {1, 2, 3, 4}
+// s2.Union(s1) = {1, 2, 3, 4}
+func (s1 StringSet) Union(s2 StringSet) StringSet {
+	result := NewStringSet()
+	for key := range s1 {
+		result.Insert(key)
+	}
+	for key := range s2 {
+		result.Insert(key)
+	}
+	return result
+}
+
 // IsSuperset returns true iff s1 is a superset of s2.
 func (s1 StringSet) IsSuperset(s2 StringSet) bool {
 	for item := range s2 {