Merge pull request #587 from floreks/concurrent-api-calls

Make single API call to get events and filter them on backend

src/app/backend/events.go

@@ -17,7 +17,7 @@ package main
 import (
 	"log"
 
-	api "k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/fields"
@@ -142,54 +142,18 @@ func GetReplicationControllerPodsEvents(client *client.Client, namespace, replic
 		return nil, err
 	}
 
-	events, err := GetPodsEvents(client, pods)
-
-	if err != nil {
-		return nil, err
-	}
-
-	return events, nil
-}
-
-// GetPodsEvents gets events associated to given list of pods
-// TODO(floreks): refactor this to make single API call instead of N api calls
-func GetPodsEvents(client *client.Client, pods *api.PodList) ([]api.Event, error) {
-	events := make([]api.Event, 0, 0)
-
-	for _, pod := range pods.Items {
-		list, err := GetPodEvents(client, pod)
-
-		if err != nil {
-			return nil, err
-		}
-
-		for _, event := range list.Items {
-			events = append(events, event)
-		}
-
-	}
-
-	return events, nil
-}
-
-// GetPodEvents gets events associated to given pod
-func GetPodEvents(client client.Interface, pod api.Pod) (*api.EventList, error) {
-	fieldSelector, err := fields.ParseSelector("involvedObject.name=" + pod.Name)
-
-	if err != nil {
-		return nil, err
-	}
-
-	list, err := client.Events(pod.Namespace).List(api.ListOptions{
+	eventList, err := client.Events(namespace).List(api.ListOptions{
 		LabelSelector: labels.Everything(),
-		FieldSelector: fieldSelector,
+		FieldSelector: fields.Everything(),
 	})
 
 	if err != nil {
 		return nil, err
 	}
 
-	return list, nil
+	events := filterEventsByPodsUID(eventList.Items, pods.Items)
+
+	return events, nil
 }
 
 // AppendEvents appends events from source slice to target events representation.

src/app/backend/eventscommon.go

@@ -16,8 +16,7 @@ package main
 
 import (
 	"k8s.io/kubernetes/pkg/api"
-	client "k8s.io/kubernetes/pkg/client/unversioned"
-	"log"
+	"k8s.io/kubernetes/pkg/types"
 	"strings"
 )
 
@@ -30,48 +29,73 @@ import (
 var FailedReasonPartials = []string{"failed", "err", "exceeded", "invalid", "unhealthy",
 	"mismatch", "insufficient", "conflict", "outof", "nil"}
 
-// GetPodsEventWarnings returns warning pod events based on given list of pods.
+// GetPodsEventWarnings returns warning pod events by filtering out events targeting only given pods
 // TODO(floreks) : Import and use Set instead of custom function to get rid of duplicates
-func GetPodsEventWarnings(client client.Interface, pods []api.Pod) (result []Event, err error) {
+func GetPodsEventWarnings(eventList *api.EventList, pods []api.Pod) []Event {
+	result := make([]Event, 0)
+	if eventList == nil {
+		return result
+	}
+
+	// Filter out only warning events
+	events := getWarningEvents(eventList)
+	failedPods := make([]api.Pod, 0)
+
+	// Filter out only 'failed' pods
 	for _, pod := range pods {
 		if !isRunningOrSucceeded(pod) {
-			log.Printf("Getting warning events from pod: %s", pod.Name)
-			events, err := GetPodEvents(client, pod)
-
-			if err != nil {
-				return nil, err
+			failedPods = append(failedPods, pod)
 		}
-
-			result = getPodsEventWarnings(events)
 	}
+
+	// Filter events by failed pods UID
+	events = filterEventsByPodsUID(events, failedPods)
+	events = removeDuplicates(events)
+
+	for _, event := range events {
+		result = append(result, Event{
+			Message: event.Message,
+			Reason:  event.Reason,
+			Type:    event.Type,
+		})
 	}
 
-	return removeDuplicates(result), nil
+	return result
 }
 
-// Returns list of Pod Event model objects based on kubernetes API event list object
-// Event list object is filtered to get only warning events.
-func getPodsEventWarnings(eventList *api.EventList) []Event {
-	result := make([]Event, 0)
+// Returns filtered list of event objects.
+// Events list is filtered to get only events targeting pods on the list.
+func filterEventsByPodsUID(events []api.Event, pods []api.Pod) []api.Event {
+	result := make([]api.Event, 0)
+	podEventMap := make(map[types.UID]bool, 0)
 
-	var events []api.Event
-	if !isTypeFilled(eventList.Items) {
-		eventList.Items = fillEventsType(eventList.Items)
+	if len(pods) == 0 || len(events) == 0 {
+		return result
 	}
 
-	events = filterEventsByType(eventList.Items, api.EventTypeWarning)
+	for _, pod := range pods {
+		podEventMap[pod.UID] = true
+	}
 
 	for _, event := range events {
-		result = append(result, Event{
-			Message: event.Message,
-			Reason:  event.Reason,
-			Type:    event.Type,
-		})
+		if _, exists := podEventMap[event.InvolvedObject.UID]; exists {
+			result = append(result, event)
+		}
 	}
 
 	return result
 }
 
+// Returns filtered list of event objects.
+// Event list object is filtered to get only warning events.
+func getWarningEvents(eventList *api.EventList) []api.Event {
+	if !isTypeFilled(eventList.Items) {
+		eventList.Items = fillEventsType(eventList.Items)
+	}
+
+	return filterEventsByType(eventList.Items, api.EventTypeWarning)
+}
+
 // Filters kubernetes API event objects based on event type.
 // Empty string will return all events.
 func filterEventsByType(events []api.Event, eventType string) []api.Event {
@@ -131,9 +155,9 @@ func fillEventsType(events []api.Event) []api.Event {
 }
 
 // Removes duplicate strings from the slice
-func removeDuplicates(slice []Event) []Event {
+func removeDuplicates(slice []api.Event) []api.Event {
 	visited := make(map[string]bool, 0)
-	result := make([]Event, 0)
+	result := make([]api.Event, 0)
 
 	for _, elem := range slice {
 		if !visited[elem.Reason] {

src/app/backend/replicationcontrollerlist.go

@@ -25,7 +25,7 @@ import (
 )
 
 // GetPodsEventWarningsFunc is a callback function used to get the pod status errors.
-type GetPodsEventWarningsFunc func(pods []api.Pod) ([]Event, error)
+type GetPodsEventWarningsFunc func(pods []api.Pod) []Event
 
 // GetNodeFunc is a callback function used to get nodes by names.
 type GetNodeFunc func(nodeName string) (*api.Node, error)
@@ -94,17 +94,20 @@ func GetReplicationControllerList(client *client.Client) (*ReplicationController
 		return nil, err
 	}
 
-	// Anonymous callback function to get pods warnings.
-	// Function fulfils GetPodsEventWarningsFunc type contract.
-	// Based on list of api pods returns list of pod related warning events
-	getPodsEventWarningsFn := func(pods []api.Pod) ([]Event, error) {
-		errors, err := GetPodsEventWarnings(client, pods)
+	eventsList, err := client.Events(api.NamespaceAll).List(api.ListOptions{
+		LabelSelector: labels.Everything(),
+		FieldSelector: fields.Everything(),
+	})
 
 	if err != nil {
 		return nil, err
 	}
 
-		return errors, nil
+	// Anonymous callback function to get pods warnings.
+	// 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)
 	}
 
 	// Anonymous callback function to get nodes by their names.
@@ -155,11 +158,7 @@ func getReplicationControllerList(replicationControllers []api.ReplicationContro
 			}
 		}
 		podInfo := getReplicationControllerPodInfo(&replicationController, matchingPods)
-		podErrors, err := getPodsEventWarningsFn(matchingPods)
-
-		if err != nil {
-			return nil, err
-		}
+		podErrors := getPodsEventWarningsFn(matchingPods)
 
 		podInfo.Warnings = podErrors
 

src/test/backend/eventscommon_test.go

@@ -19,24 +19,41 @@ import (
 	"testing"
 
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/client/unversioned/testclient"
 )
 
 func TestGetPodsEventWarningsApi(t *testing.T) {
 	cases := []struct {
 		pods      []api.Pod
-		expectedActions []string
+		eventList *api.EventList
+		expected  []Event
 	}{
-		{nil, []string{}},
+		{nil, nil, []Event{}},
 		{
 			[]api.Pod{
 				{
+					ObjectMeta: api.ObjectMeta{
+						Name: "FailedPod",
+					},
 					Status: api.PodStatus{
 						Phase: api.PodFailed,
 					},
 				},
 			},
-			[]string{"get"},
+			&api.EventList{Items: []api.Event{
+				{
+					Type:    api.EventTypeWarning,
+					Message: "Test Message",
+					InvolvedObject: api.ObjectReference{
+						Name: "FailedPod",
+					},
+				},
+			}},
+			[]Event{
+				{
+					Message: "Test Message",
+					Type:    api.EventTypeWarning,
+				},
+			},
 		},
 		{
 			[]api.Pod{
@@ -46,31 +63,27 @@ func TestGetPodsEventWarningsApi(t *testing.T) {
 					},
 				},
 			},
-			[]string{},
+			&api.EventList{},
+			[]Event{},
 		},
 	}
 
 	for _, c := range cases {
-		eventList := &api.EventList{}
-		fakeClient := testclient.NewSimpleFake(eventList)
+		actual := GetPodsEventWarnings(c.eventList, c.pods)
 
-		GetPodsEventWarnings(fakeClient, c.pods)
-
-		actions := fakeClient.Actions()
-		if len(actions) != len(c.expectedActions) {
-			t.Errorf("Unexpected actions: %v, expected %d actions got %d", actions,
-				len(c.expectedActions), len(actions))
-			continue
+		if len(actual) != len(c.expected) || !reflect.DeepEqual(actual, c.expected) {
+			t.Errorf("GetPodsEventWarnings(%#v, %#v) == \n%#v\nexpected \n%#v\n",
+				c.eventList, c.pods, actual, c.expected)
 		}
 	}
 }
 
-func TestGetPodsEventWarnings(t *testing.T) {
+func TestGetWarningEvents(t *testing.T) {
 	cases := []struct {
 		events   *api.EventList
-		expected []Event
+		expected []api.Event
 	}{
-		{&api.EventList{Items: nil}, []Event{}},
+		{&api.EventList{Items: []api.Event{}}, []api.Event{}},
 		{
 			&api.EventList{
 				Items: []api.Event{
@@ -81,7 +94,7 @@ func TestGetPodsEventWarnings(t *testing.T) {
 					},
 				},
 			},
-			[]Event{
+			[]api.Event{
 				{
 					Message: "msg",
 					Reason:  "reason",
@@ -98,7 +111,7 @@ func TestGetPodsEventWarnings(t *testing.T) {
 					},
 				},
 			},
-			[]Event{
+			[]api.Event{
 				{
 					Message: "msg",
 					Reason:  "failed",
@@ -115,14 +128,14 @@ func TestGetPodsEventWarnings(t *testing.T) {
 					},
 				},
 			},
-			[]Event{},
+			[]api.Event{},
 		},
 	}
 
 	for _, c := range cases {
-		actual := getPodsEventWarnings(c.events)
+		actual := getWarningEvents(c.events)
 		if !reflect.DeepEqual(actual, c.expected) {
-			t.Errorf("getPodsEventErrors(%#v) == \n%#v\nexpected \n%#v\n",
+			t.Errorf("getWarningEvents(%#v) == \n%#v\nexpected \n%#v\n",
 				c.events, actual, c.expected)
 		}
 	}
@@ -173,38 +186,38 @@ func TestFilterEventsByType(t *testing.T) {
 
 func TestRemoveDuplicates(t *testing.T) {
 	cases := []struct {
-		slice    []Event
-		expected []Event
+		slice    []api.Event
+		expected []api.Event
 	}{
-		{nil, []Event{}},
+		{nil, []api.Event{}},
 		{
-			[]Event{
+			[]api.Event{
 				{Reason: "test"},
 				{Reason: "test2"},
 				{Reason: "test"},
 			},
-			[]Event{
+			[]api.Event{
 				{Reason: "test"},
 				{Reason: "test2"},
 			},
 		},
 		{
-			[]Event{
+			[]api.Event{
 				{Reason: "test"},
 				{Reason: "test"},
 				{Reason: "test"},
 			},
-			[]Event{
+			[]api.Event{
 				{Reason: "test"},
 			},
 		},
 		{
-			[]Event{
+			[]api.Event{
 				{Reason: "test"},
 				{Reason: "test2"},
 				{Reason: "test3"},
 			},
-			[]Event{
+			[]api.Event{
 				{Reason: "test"},
 				{Reason: "test2"},
 				{Reason: "test3"},
@@ -353,3 +366,47 @@ func TestIsFailedReason(t *testing.T) {
 		}
 	}
 }
+
+func TestFilterEventsByPodsUID(t *testing.T) {
+	cases := []struct {
+		events   []api.Event
+		pods     []api.Pod
+		expected []api.Event
+	}{
+		{nil, nil, []api.Event{}},
+		{
+			[]api.Event{
+				{InvolvedObject: api.ObjectReference{UID: "TestPod"}},
+				{InvolvedObject: api.ObjectReference{UID: "TestObject"}},
+			},
+			[]api.Pod{
+				{ObjectMeta: api.ObjectMeta{UID: "TestPod"}},
+			},
+			[]api.Event{
+				{InvolvedObject: api.ObjectReference{UID: "TestPod"}},
+			},
+		},
+		{
+			// To check whether multiple events targeting same object are correctly filtered
+			[]api.Event{
+				{InvolvedObject: api.ObjectReference{UID: "TestPod"}},
+				{InvolvedObject: api.ObjectReference{UID: "TestPod"}},
+			},
+			[]api.Pod{
+				{ObjectMeta: api.ObjectMeta{UID: "TestPod"}},
+			},
+			[]api.Event{
+				{InvolvedObject: api.ObjectReference{UID: "TestPod"}},
+				{InvolvedObject: api.ObjectReference{UID: "TestPod"}},
+			},
+		},
+	}
+
+	for _, c := range cases {
+		actual := filterEventsByPodsUID(c.events, c.pods)
+		if !reflect.DeepEqual(actual, c.expected) {
+			t.Errorf("filterEventsByPodsName(%#v, %#v) == \n%#v\nexpected \n%#v\n",
+				c.events, c.pods, actual, c.expected)
+		}
+	}
+}

src/test/backend/replicationcontrollerlist_test.go

@@ -83,8 +83,8 @@ func TestGetMatchingServices(t *testing.T) {
 }
 
 func TestGetReplicationControllerList(t *testing.T) {
-	getPodsErrorFnMock := func(pods []api.Pod) ([]Event, error) {
-		return []Event{}, nil
+	getPodsErrorFnMock := func(pods []api.Pod) []Event {
+		return []Event{}
 	}
 
 	cases := []struct {