Created gatherer for gathering openshift-machine-api events (!=Normal).

docs/gathered-data.md

@@ -684,6 +684,17 @@ API Reference:
   * 4.9+
 
 
+## OpenshiftMachineAPIEvents
+
+collects warning ("abnormal") events
+from "openshift-machine-api" namespace
+
+* Location of events in archive: events/
+* Id in config: clusterconfig/openshift_machine_api_events
+* Since versions:
+	 * 4.12+
+
+
 ## OpenshiftSDNControllerLogs
 
 collects logs from sdn-controller pod in openshift-sdn namespace with following substrings:

docs/insights-archive-sample/events/openshift-machine-api.json

@@ -0,0 +1,18 @@
+{
+  "items":[
+    {
+      "namespace":"openshift-machine-api",
+      "lastTimestamp":"2022-08-24T11:40:44+02:00",
+      "reason":"FailedUpdate",
+      "message":"cluster-lrqft-worker-us-east-2a-jsmg4: reconciler failed to Update machine: requeue in: 20s",
+      "type":"Warning"
+    },
+    {
+      "namespace":"openshift-machine-api",
+      "lastTimestamp":"2022-08-24T11:41:05+02:00",
+      "reason":"FailedUpdate",
+      "message":"cluster-lrqft-worker-us-east-2c-rlfpx: reconciler failed to Update machine: requeue in: 20s",
+      "type":"Warning"
+    }
+  ]
+}

pkg/gatherers/clusterconfig/clusterconfig_gatherer.go

@@ -84,6 +84,7 @@ var gatheringFunctions = map[string]gathererFuncPtr{
 	"support_secret":                    (*Gatherer).GatherSupportSecret,
 	"active_alerts":                     (*Gatherer).GatherActiveAlerts,
 	"ceph_cluster":                      (*Gatherer).GatherCephCluster,
+	"openshift_machine_api_events":      (*Gatherer).GatherOpenshiftMachineAPIEvents,
 }
 
 func New(

pkg/gatherers/clusterconfig/events_filtering.go

@@ -0,0 +1,77 @@
+package clusterconfig
+
+import (
+	"sort"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+)
+
+// getEventsForInterval() returns events that occoured since last interval
+func getEventsForInterval(interval time.Duration, events *v1.EventList) v1.EventList {
+	oldestEventTime := time.Now().Add(-interval)
+	var filteredEvents v1.EventList
+	for i := range events.Items {
+		if isEventNew(&events.Items[i], oldestEventTime) {
+			filteredEvents.Items = append(filteredEvents.Items, events.Items[i])
+		}
+	}
+	return filteredEvents
+}
+
+// isEventNew() returns true if event occoured after given time, otherwise returns false
+func isEventNew(event *v1.Event, oldestEventTime time.Time) bool {
+	if event.LastTimestamp.Time.After(oldestEventTime) {
+		return true
+		// if LastTimestamp is zero then try to check the event series
+	} else if event.LastTimestamp.IsZero() {
+		if event.Series != nil {
+			if event.Series.LastObservedTime.Time.After(oldestEventTime) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// filterAbnormalEvents returns events that have Type different from "Normal"
+func filterAbnormalEvents(events *v1.EventList) v1.EventList {
+	var filteredEvents v1.EventList
+	for i := range events.Items {
+		if isEventAbnormal(&events.Items[i]) {
+			filteredEvents.Items = append(filteredEvents.Items, events.Items[i])
+		}
+	}
+	return filteredEvents
+}
+
+func isEventAbnormal(event *v1.Event) bool {
+	return event.Type != "Normal"
+}
+
+// eventListToCompactedEventList() converts EventList into CompactedEventList
+func eventListToCompactedEventList(events *v1.EventList) CompactedEventList {
+	var compactedEvents CompactedEventList
+	for i := range events.Items {
+		event := events.Items[i]
+		compactedEvent := CompactedEvent{
+			Namespace:     event.Namespace,
+			LastTimestamp: event.LastTimestamp.Time,
+			Reason:        event.Reason,
+			Message:       event.Message,
+			Type:          event.Type,
+		}
+		if event.LastTimestamp.Time.IsZero() {
+			if event.Series != nil {
+				compactedEvent.LastTimestamp = event.Series.LastObservedTime.Time
+			}
+		}
+		compactedEvents.Items = append(compactedEvents.Items, compactedEvent)
+	}
+
+	sort.Slice(compactedEvents.Items, func(i, j int) bool {
+		return compactedEvents.Items[i].LastTimestamp.Before(compactedEvents.Items[j].LastTimestamp)
+	})
+
+	return compactedEvents
+}

pkg/gatherers/clusterconfig/events_filtering_test.go

@@ -0,0 +1,170 @@
+package clusterconfig
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func Test_getEventsForInterval(t *testing.T) {
+	timeNow := time.Now()
+	test := struct {
+		events   v1.EventList
+		expected v1.EventList
+	}{
+		events: v1.EventList{
+			Items: []v1.Event{
+				{
+					ObjectMeta:    metav1.ObjectMeta{Name: "oldEvent1"},
+					LastTimestamp: metav1.Time{},
+				},
+				{
+					ObjectMeta:    metav1.ObjectMeta{Name: "newEvent1"},
+					LastTimestamp: metav1.NewTime(timeNow),
+				},
+				{
+					ObjectMeta:    metav1.ObjectMeta{Name: "oldEvent2"},
+					LastTimestamp: metav1.Time{},
+				},
+				{
+					ObjectMeta:    metav1.ObjectMeta{Name: "newEvent2"},
+					LastTimestamp: metav1.NewTime(timeNow),
+				},
+				{
+					ObjectMeta:    metav1.ObjectMeta{Name: "newEvent3"},
+					LastTimestamp: metav1.NewTime(timeNow),
+				},
+			},
+		},
+		expected: v1.EventList{
+			Items: []v1.Event{
+				{
+					ObjectMeta:    metav1.ObjectMeta{Name: "newEvent1"},
+					LastTimestamp: metav1.NewTime(timeNow),
+				},
+				{
+					ObjectMeta:    metav1.ObjectMeta{Name: "newEvent2"},
+					LastTimestamp: metav1.NewTime(timeNow),
+				},
+				{
+					ObjectMeta:    metav1.ObjectMeta{Name: "newEvent3"},
+					LastTimestamp: metav1.NewTime(timeNow),
+				},
+			},
+		},
+	}
+
+	filteredEvents := getEventsForInterval(1*time.Minute, &test.events)
+	if !reflect.DeepEqual(filteredEvents, test.expected) {
+		t.Errorf("filterEvents() = %v, want %v", filteredEvents, test.expected)
+	}
+}
+
+func Test_filterAbnormalEvents(t *testing.T) {
+	test := struct {
+		events   v1.EventList
+		expected v1.EventList
+	}{
+		events: v1.EventList{
+			Items: []v1.Event{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "normalEvent1"},
+					Type:       "Normal",
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "warningEvent1"},
+					Type:       "Warning",
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "normalEvent2"},
+					Type:       "Normal",
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "warningEvent2"},
+					Type:       "Warning",
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "warningEvent3"},
+					Type:       "Warning",
+				},
+			},
+		},
+		expected: v1.EventList{
+			Items: []v1.Event{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "warningEvent1"},
+					Type:       "Warning",
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "warningEvent2"},
+					Type:       "Warning",
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "warningEvent3"},
+					Type:       "Warning",
+				},
+			},
+		},
+	}
+
+	filteredEvents := filterAbnormalEvents(&test.events)
+	if !reflect.DeepEqual(filteredEvents, test.expected) {
+		t.Errorf("filterEvents() = %v, want %v", filteredEvents, test.expected)
+	}
+}
+
+func Test_isEventNew(t *testing.T) {
+	tests := []struct {
+		event    v1.Event
+		expected bool
+	}{
+		{
+			event: v1.Event{
+				ObjectMeta:    metav1.ObjectMeta{Name: "newEvent"},
+				LastTimestamp: metav1.Now(),
+				Type:          "Normal",
+			},
+			expected: true,
+		},
+		{
+			event: v1.Event{
+				ObjectMeta:    metav1.ObjectMeta{Name: "oldEvent"},
+				LastTimestamp: metav1.NewTime(time.Now().Add(-6 * time.Minute)),
+				Type:          "Normal",
+			},
+			expected: false,
+		},
+	}
+
+	for _, test := range tests {
+		if isEventNew(&test.event, time.Now().Add(-5*time.Minute)) != test.expected {
+			t.Errorf("isEventNew() = %v, got %v", !test.expected, test.expected)
+		}
+	}
+}
+
+func Test_eventListToCompactedEventList(t *testing.T) {
+	timeNow := time.Now()
+	event := v1.Event{
+		ObjectMeta:    metav1.ObjectMeta{Name: "event", Namespace: "test namespace"},
+		LastTimestamp: metav1.NewTime(timeNow),
+		Type:          "Normal",
+		Reason:        "test reason",
+		Message:       "test message",
+	}
+	compactedEvent := CompactedEvent{
+		Namespace:     "test namespace",
+		LastTimestamp: timeNow,
+		Reason:        "test reason",
+		Message:       "test message",
+		Type:          "Normal",
+	}
+	compactedEventList := eventListToCompactedEventList(&v1.EventList{Items: []v1.Event{event}})
+
+	if !reflect.DeepEqual(compactedEvent, compactedEventList.Items[0]) {
+		t.Errorf("eventListToCompactedEventList() = %v, want %v", compactedEventList.Items[0], compactedEvent)
+	}
+}

pkg/gatherers/clusterconfig/openshift_machine_api_events.go

@@ -0,0 +1,49 @@
+package clusterconfig
+
+import (
+	"context"
+	"time"
+
+	"github.com/openshift/insights-operator/pkg/record"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+)
+
+// GatherOpenshiftMachineAPIEvents collects warning ("abnormal") events
+// from "openshift-machine-api" namespace
+//
+// * Location of events in archive: events/
+// * Id in config: clusterconfig/openshift_machine_api_events
+// * Since versions:
+//   * 4.12+
+func (g *Gatherer) GatherOpenshiftMachineAPIEvents(ctx context.Context) ([]record.Record, []error) {
+	gatherKubeClient, err := kubernetes.NewForConfig(g.gatherProtoKubeConfig)
+	if err != nil {
+		return nil, []error{err}
+	}
+	records, err := gatherOpenshiftMachineAPIEvents(ctx, gatherKubeClient.CoreV1(), g.interval)
+	if err != nil {
+		return nil, []error{err}
+	}
+	return records, nil
+}
+
+func gatherOpenshiftMachineAPIEvents(ctx context.Context,
+	coreClient corev1client.CoreV1Interface,
+	interval time.Duration) ([]record.Record, error) {
+	events, err := coreClient.Events("openshift-machine-api").List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+	// filter the event list to only recent events with type different than "Normal"
+	filteredEvents := getEventsForInterval(interval, events)
+	filteredEvents = filterAbnormalEvents(&filteredEvents)
+
+	if len(filteredEvents.Items) == 0 {
+		return nil, nil
+	}
+	compactedEvents := eventListToCompactedEventList(&filteredEvents)
+
+	return []record.Record{{Name: "events/openshift-machine-api", Item: record.JSONMarshaller{Object: &compactedEvents}}}, nil
+}