Merge pull request #667 from bryk/async-lists

Move RC list to resource channels framework

src/app/backend/replicationcontrollerlist.go

@@ -15,13 +15,12 @@
 package main
 
 import (
+	"errors"
 	"log"
 
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
-	"k8s.io/kubernetes/pkg/fields"
-	"k8s.io/kubernetes/pkg/labels"
 )
 
 // GetPodsEventWarningsFunc is a callback function used to get the pod status errors.
@@ -71,35 +70,44 @@ type ReplicationController struct {
 func GetReplicationControllerList(client *client.Client) (*ReplicationControllerList, error) {
 	log.Printf("Getting list of all replication controllers in the cluster")
 
-	listEverything := api.ListOptions{
-		LabelSelector: labels.Everything(),
-		FieldSelector: fields.Everything(),
+	channels := &ResourceChannels{
+		ReplicationControllerList: getReplicationControllerListChannel(client, 1),
+		ServiceList:               getServiceListChannel(client, 1),
+		PodList:                   getPodListChannel(client, 1),
+		EventList:                 getEventListChannel(client, 1),
+		NodeList:                  getNodeListChannel(client, 1),
 	}
 
-	replicationControllers, err := client.ReplicationControllers(api.NamespaceAll).List(listEverything)
+	return GetReplicationControllerListFromChannels(channels)
+}
 
-	if err != nil {
+// GetReplicationControllerList returns a list of all Replication Controllers in the cluster
+// reading required resource list once from the channels.
+func GetReplicationControllerListFromChannels(channels *ResourceChannels) (
+	*ReplicationControllerList, error) {
+
+	replicationControllers := <-channels.ReplicationControllerList.List
+	if err := <-channels.ReplicationControllerList.Error; err != nil {
 		return nil, err
 	}
 
-	services, err := client.Services(api.NamespaceAll).List(listEverything)
-
-	if err != nil {
+	services := <-channels.ServiceList.List
+	if err := <-channels.ServiceList.Error; err != nil {
 		return nil, err
 	}
 
-	pods, err := client.Pods(api.NamespaceAll).List(listEverything)
-
-	if err != nil {
+	pods := <-channels.PodList.List
+	if err := <-channels.PodList.Error; err != nil {
 		return nil, err
 	}
 
-	eventsList, err := client.Events(api.NamespaceAll).List(api.ListOptions{
-		LabelSelector: labels.Everything(),
-		FieldSelector: fields.Everything(),
-	})
+	events := <-channels.EventList.List
+	if err := <-channels.EventList.Error; err != nil {
+		return nil, err
+	}
 
-	if err != nil {
+	nodes := <-channels.NodeList.List
+	if err := <-channels.NodeList.Error; err != nil {
 		return nil, err
 	}
 
@@ -107,12 +115,17 @@ func GetReplicationControllerList(client *client.Client) (*ReplicationController
 	// Function fulfils GetPodsEventWarningsFunc type contract.
 	// Based on list of api pods returns list of pod related warning events
 	getPodsEventWarningsFn := func(pods []api.Pod) []Event {
-		return GetPodsEventWarnings(eventsList, pods)
+		return GetPodsEventWarnings(events, pods)
 	}
 
 	// Anonymous callback function to get nodes by their names.
 	getNodeFn := func(nodeName string) (*api.Node, error) {
-		return client.Nodes().Get(nodeName)
+		for _, node := range nodes.Items {
+			if node.ObjectMeta.Name == nodeName {
+				return &node, nil
+			}
+		}
+		return nil, errors.New("Cannot find node " + nodeName)
 	}
 
 	result, err := getReplicationControllerList(replicationControllers.Items, services.Items,

src/app/backend/resourcechannels.go

+package main
+
+import (
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/apis/extensions"
+	client "k8s.io/kubernetes/pkg/client/unversioned"
+	"k8s.io/kubernetes/pkg/fields"
+	"k8s.io/kubernetes/pkg/labels"
+)
+
+// ResourceChannels struct holds channels to resource lists. Each list channel is paired with
+// an error channel which *must* be read sequentially: first read the list channel and then
+// the error channel.
+//
+// This struct can be used when there are multiple clients that want to process, e.g., a
+// list of pods. With this helper, the list can be read only once from the backend and
+// distributed asynchronously to clients that need it.
+//
+// When a channel is nil, it means that no resource list is available for getting.
+//
+// Each channel pair can be read up to N times. N is specified upon creation of the channels.
+type ResourceChannels struct {
+	// List and error channels to Replication Controllers.
+	ReplicationControllerList ReplicationControllerListChannel
+
+	// List and error channels to Replica Sets.
+	ReplicaSetList ReplicaSetListChannel
+
+	// List and error channels to Services.
+	ServiceList ServiceListChannel
+
+	// List and error channels to Pods.
+	PodList PodListChannel
+
+	// List and error channels to Events.
+	EventList EventListChannel
+
+	// List and error channels to Nodes.
+	NodeList NodeListChannel
+}
+
+// List and error channels to Services.
+type ServiceListChannel struct {
+	List  chan *api.ServiceList
+	Error chan error
+}
+
+var listEverything api.ListOptions = api.ListOptions{
+	LabelSelector: labels.Everything(),
+	FieldSelector: fields.Everything(),
+}
+
+// Returns a pair of channels to a Service list and errors that both must be read
+// numReads times.
+func getServiceListChannel(client *client.Client, numReads int) ServiceListChannel {
+	channel := ServiceListChannel{
+		List:  make(chan *api.ServiceList, numReads),
+		Error: make(chan error, numReads),
+	}
+	go func() {
+		services, err := client.Services(api.NamespaceAll).List(listEverything)
+
+		for i := 0; i < numReads; i++ {
+			channel.List <- services
+			channel.Error <- err
+		}
+	}()
+
+	return channel
+}
+
+// List and error channels to Nodes.
+type NodeListChannel struct {
+	List  chan *api.NodeList
+	Error chan error
+}
+
+// Returns a pair of channels to a Node list and errors that both must be read
+// numReads times.
+func getNodeListChannel(client *client.Client, numReads int) NodeListChannel {
+	channel := NodeListChannel{
+		List:  make(chan *api.NodeList, numReads),
+		Error: make(chan error, numReads),
+	}
+
+	go func() {
+		nodes, err := client.Nodes().List(listEverything)
+
+		for i := 0; i < numReads; i++ {
+			channel.List <- nodes
+			channel.Error <- err
+		}
+	}()
+
+	return channel
+}
+
+// List and error channels to Nodes.
+type EventListChannel struct {
+	List  chan *api.EventList
+	Error chan error
+}
+
+// Returns a pair of channels to an Event list and errors that both must be read
+// numReads times.
+func getEventListChannel(client *client.Client, numReads int) EventListChannel {
+	channel := EventListChannel{
+		List:  make(chan *api.EventList, numReads),
+		Error: make(chan error, numReads),
+	}
+
+	go func() {
+		events, err := client.Events(api.NamespaceAll).List(listEverything)
+
+		for i := 0; i < numReads; i++ {
+			channel.List <- events
+			channel.Error <- err
+		}
+	}()
+
+	return channel
+}
+
+// List and error channels to Nodes.
+type PodListChannel struct {
+	List  chan *api.PodList
+	Error chan error
+}
+
+// Returns a pair of channels to a Pod list and errors that both must be read
+// numReads times.
+func getPodListChannel(client *client.Client, numReads int) PodListChannel {
+	channel := PodListChannel{
+		List:  make(chan *api.PodList, numReads),
+		Error: make(chan error, numReads),
+	}
+
+	go func() {
+		pods, err := client.Pods(api.NamespaceAll).List(listEverything)
+
+		for i := 0; i < numReads; i++ {
+			channel.List <- pods
+			channel.Error <- err
+		}
+	}()
+
+	return channel
+}
+
+// List and error channels to Nodes.
+type ReplicationControllerListChannel struct {
+	List  chan *api.ReplicationControllerList
+	Error chan error
+}
+
+// Returns a pair of channels to a Replication Controller list and errors that both must be read
+// numReads times.
+func getReplicationControllerListChannel(client *client.Client, numReads int) ReplicationControllerListChannel {
+	channel := ReplicationControllerListChannel{
+		List:  make(chan *api.ReplicationControllerList, numReads),
+		Error: make(chan error, numReads),
+	}
+
+	go func() {
+		rcs, err := client.ReplicationControllers(api.NamespaceAll).List(listEverything)
+		for i := 0; i < numReads; i++ {
+			channel.List <- rcs
+			channel.Error <- err
+		}
+	}()
+
+	return channel
+}
+
+// List and error channels to Nodes.
+type ReplicaSetListChannel struct {
+	List  chan *extensions.ReplicaSetList
+	Error chan error
+}
+
+// Returns a pair of channels to a ReplicaSet list and errors that both must be read
+// numReads times.
+func getReplicaSetListChannel(client *client.Client, numReads int) ReplicaSetListChannel {
+	channel := ReplicaSetListChannel{
+		List:  make(chan *extensions.ReplicaSetList, numReads),
+		Error: make(chan error, numReads),
+	}
+
+	go func() {
+		rcs, err := client.ReplicaSets(api.NamespaceAll).List(listEverything)
+		for i := 0; i < numReads; i++ {
+			channel.List <- rcs
+			channel.Error <- err
+		}
+	}()
+
+	return channel
+}