Merge pull request #241 from pravisankar/fix-watch-get-resources

Fix SDN get and watch resource workflow

Author: danwinship

plugins/osdn/common.go

@@ -42,12 +42,10 @@ type OsdnController struct {
 	localSubnet     *osapi.HostSubnet
 	HostName        string
 	subnetAllocator *netutils.SubnetAllocator
-	sig             chan struct{}
 	podNetworkReady chan struct{}
 	VNIDMap         map[string]uint
 	netIDManager    *netutils.NetIDAllocator
 	adminNamespaces []string
-	services        map[string]*kapi.Service
 }
 
 // Called by plug factory functions to initialize the generic plugin instance
@@ -84,10 +82,8 @@ func (oc *OsdnController) BaseInit(registry *Registry, pluginHooks PluginHooks,
 	oc.localIP = selfIP
 	oc.HostName = hostname
 	oc.VNIDMap = make(map[string]uint)
-	oc.sig = make(chan struct{})
 	oc.podNetworkReady = make(chan struct{})
 	oc.adminNamespaces = make([]string, 0)
-	oc.services = make(map[string]*kapi.Service)
 
 	return nil
 }
@@ -119,7 +115,7 @@ func (oc *OsdnController) validateNetworkConfig(clusterNetwork, serviceNetwork *
 	}
 
 	// Ensure each host subnet is within the cluster network
-	subnets, _, err := oc.Registry.GetSubnets()
+	subnets, err := oc.Registry.GetSubnets()
 	if err != nil {
 		return fmt.Errorf("Error in initializing/fetching subnets: %v", err)
 	}
@@ -135,7 +131,7 @@ func (oc *OsdnController) validateNetworkConfig(clusterNetwork, serviceNetwork *
 	}
 
 	// Ensure each service is within the services network
-	services, _, err := oc.Registry.GetServices()
+	services, err := oc.Registry.GetServices()
 	if err != nil {
 		return err
 	}
@@ -242,57 +238,6 @@ func (oc *OsdnController) WaitForPodNetworkReady() error {
 	return fmt.Errorf("SDN pod network is not ready(timeout: 2 mins)")
 }
 
-func (oc *OsdnController) Stop() {
-	close(oc.sig)
-}
-
-// Wait for ready signal from Watch interface for the given resource
-// Closes the ready channel as we don't need it anymore after this point
-func waitForWatchReadiness(ready chan bool, resourceName string) {
-	timeout := time.Minute
-	select {
-	case <-ready:
-		close(ready)
-	case <-time.After(timeout):
-		log.Fatalf("Watch for resource %s is not ready(timeout: %v)", resourceName, timeout)
-	}
-	return
-}
-
-type watchWatcher func(oc *OsdnController, ready chan<- bool, start <-chan string)
-type watchGetter func(registry *Registry) (interface{}, string, error)
-
-// watchAndGetResource will fetch current items in etcd and watch for any new
-// changes for the given resource.
-// Supported resources: nodes, subnets, namespaces, services, netnamespaces, and pods.
-//
-// To avoid any potential race conditions during this process, these steps are followed:
-// 1. Initiator(master/node): Watch for a resource as an async op, lets say WatchProcess
-// 2. WatchProcess: When ready for watching, send ready signal to initiator
-// 3. Initiator: Wait for watch resource to be ready
-//    This is needed as step-1 is an asynchronous operation
-// 4. WatchProcess: Collect new changes in the queue but wait for initiator
-//    to indicate which version to start from
-// 5. Initiator: Get existing items with their latest version for the resource
-// 6. Initiator: Send version from step-5 to WatchProcess
-// 7. WatchProcess: Ignore any items with version <= start version got from initiator on step-6
-// 8. WatchProcess: Handle new changes
-func (oc *OsdnController) watchAndGetResource(resourceName string, watcher watchWatcher, getter watchGetter) (interface{}, error) {
-	ready := make(chan bool)
-	start := make(chan string)
-
-	go watcher(oc, ready, start)
-	waitForWatchReadiness(ready, strings.ToLower(resourceName))
-	getOutput, version, err := getter(oc.Registry)
-	if err != nil {
-		return nil, err
-	}
-
-	start <- version
-
-	return getOutput, nil
-}
-
 type FirewallRule struct {
 	table string
 	chain string