From d727895826eb7b2550449a6b263b82d2ccb094ec Mon Sep 17 00:00:00 2001 From: Venkat Kunta Date: Mon, 12 Oct 2026 14:04:47 -0500 Subject: [PATCH] @W-24431985: Watch Events with an unstructured informer so events.k8s.io/v1 fields are not dropped Some API servers return Events on /api/v1 with only the events.k8s.io/v1 field names (regarding, note, deprecatedCount, deprecatedFirstTimestamp, deprecatedLastTimestamp, deprecatedSource). The typed corev1.Event informer decodes them with encoding/json, which silently drops those fields, so Sloop stored the Event with "involvedObject":{}, no message, count 0 and null timestamps. The event count table skips count 0, so the Event heatmap was empty and the resource Events panel showed "No events found". - Watch Events with a dynamic informer (GVR v1/events), built from the same rest config as the typed client (MakeDynamicClient). - normalizeEvent copies regarding/note/deprecated* (and series/eventTime as a fallback) into the missing core/v1 fields. Core/v1 Events are unchanged. - Use meta.Accessor when logging excluded objects; the reflect lookup of ObjectMeta panicked for unstructured objects (Events and CRDs). --- pkg/sloop/ingress/eventnormalize.go | 77 +++++++++ pkg/sloop/ingress/eventnormalize_test.go | 199 +++++++++++++++++++++++ pkg/sloop/ingress/kubeclient.go | 54 ++++-- pkg/sloop/ingress/kubewatcher.go | 39 ++++- pkg/sloop/ingress/kubewatcher_test.go | 9 +- pkg/sloop/server/server.go | 7 +- 6 files changed, 360 insertions(+), 25 deletions(-) create mode 100644 pkg/sloop/ingress/eventnormalize.go create mode 100644 pkg/sloop/ingress/eventnormalize_test.go diff --git a/pkg/sloop/ingress/eventnormalize.go b/pkg/sloop/ingress/eventnormalize.go new file mode 100644 index 00000000..7ad87a72 --- /dev/null +++ b/pkg/sloop/ingress/eventnormalize.go @@ -0,0 +1,77 @@ +/* + * Copyright (c) 2026, salesforce.com, inc. + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + * For full license text, see LICENSE.txt file in the repo root or https://opensource.org/licenses/BSD-3-Clause + */ + +package ingress + +import ( + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" +) + +// Sloop reads Events with the core/v1 field names (involvedObject, message, count, firstTimestamp, +// lastTimestamp, source). Some API servers (for example a proxy that stores events.k8s.io/v1 Events) +// return Events on /api/v1 with only the events.k8s.io/v1 field names: +// +// {"regarding":{"kind":"Pod",...},"note":"Pulling image","deprecatedCount":3, +// "deprecatedFirstTimestamp":"...","deprecatedLastTimestamp":"...","deprecatedSource":{...}} +// +// Decoded as corev1.Event, all of these fields are silently dropped, so the event is stored with an empty +// involvedObject and a count of 0. normalizeEvent copies the events.k8s.io/v1 values into the missing +// core/v1 fields. Events that already have an involvedObject are returned unchanged. +// +// The input comes from the informer cache and must not be modified, so a copy is returned. +func normalizeEvent(event *unstructured.Unstructured) *unstructured.Unstructured { + if isEmptyField(event.Object, "regarding") || !isEmptyField(event.Object, "involvedObject") { + return event + } + + out := event.DeepCopy() + copyIfMissing(out.Object, "involvedObject", "regarding") + copyIfMissing(out.Object, "message", "note") + copyIfMissing(out.Object, "source", "deprecatedSource") + copyIfMissing(out.Object, "reportingComponent", "reportingController") + copyIfMissing(out.Object, "count", "deprecatedCount") + copyIfMissing(out.Object, "firstTimestamp", "deprecatedFirstTimestamp") + copyIfMissing(out.Object, "lastTimestamp", "deprecatedLastTimestamp") + + // Events created through events.k8s.io/v1 may only have eventTime and series + copyIfMissing(out.Object, "count", "series", "count") + copyIfMissing(out.Object, "lastTimestamp", "series", "lastObservedTime") + copyIfMissing(out.Object, "firstTimestamp", "eventTime") + copyIfMissing(out.Object, "lastTimestamp", "eventTime") + if isEmptyField(out.Object, "count") { + out.Object["count"] = int64(1) + } + return out +} + +// copyIfMissing sets obj[dst] to the value at the path src when obj[dst] is empty and the src value is not +func copyIfMissing(obj map[string]interface{}, dst string, src ...string) { + if !isEmptyField(obj, dst) { + return + } + val, found, err := unstructured.NestedFieldNoCopy(obj, src...) + if err != nil || !found || isEmpty(val) { + return + } + obj[dst] = val +} + +func isEmptyField(obj map[string]interface{}, field string) bool { + return isEmpty(obj[field]) +} + +func isEmpty(val interface{}) bool { + switch v := val.(type) { + case nil: + return true + case string: + return v == "" + case map[string]interface{}: + return len(v) == 0 + } + return false +} diff --git a/pkg/sloop/ingress/eventnormalize_test.go b/pkg/sloop/ingress/eventnormalize_test.go new file mode 100644 index 00000000..c8322a37 --- /dev/null +++ b/pkg/sloop/ingress/eventnormalize_test.go @@ -0,0 +1,199 @@ +/* + * Copyright (c) 2026, salesforce.com, inc. + * All rights reserved. + * SPDX-License-Identifier: BSD-3-Clause + * For full license text, see LICENSE.txt file in the repo root or https://opensource.org/licenses/BSD-3-Clause + */ + +package ingress + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "github.com/salesforce/sloop/pkg/sloop/kubeextractor" + "github.com/salesforce/sloop/pkg/sloop/store/typed" + "github.com/stretchr/testify/assert" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + kubernetesFake "k8s.io/client-go/kubernetes/fake" +) + +// An Event served on /api/v1 with only the events.k8s.io/v1 field names +const eventsV1Payload = `{ + "apiVersion": "v1", + "kind": "Event", + "metadata": {"name": "my-app-5d8f7c6b9-x2x7q.18dcf5b287d4a557", "namespace": "default", "uid": "72d4c858-0000-0000-0000-000000000000", "resourceVersion": "5940321610", "creationTimestamp": "2026-10-09T20:14:58Z"}, + "regarding": {"kind": "Pod", "namespace": "default", "name": "my-app-5d8f7c6b9-x2x7q", "uid": "2358ba5b-0000-0000-0000-000000000000", "apiVersion": "v1", "fieldPath": "spec.containers{app}"}, + "note": "Pulling image \"busybox\"", + "reason": "Pulling", + "type": "Normal", + "deprecatedSource": {"component": "kubelet", "host": "node-1"}, + "deprecatedFirstTimestamp": "2026-10-09T20:14:58Z", + "deprecatedLastTimestamp": "2026-10-09T20:16:10Z", + "deprecatedCount": 3, + "reportingController": "kubelet", + "reportingInstance": "node-1" +}` + +const coreV1Payload = `{ + "apiVersion": "v1", + "kind": "Event", + "metadata": {"name": "my-app-5d8f7c6b9-x2x7q.18dcf5b287d4a557", "namespace": "default"}, + "involvedObject": {"kind": "Pod", "namespace": "default", "name": "my-app-5d8f7c6b9-x2x7q"}, + "message": "Pulling image \"busybox\"", + "reason": "Pulling", + "type": "Normal", + "source": {"component": "kubelet", "host": "node-1"}, + "firstTimestamp": "2026-10-09T20:14:58Z", + "lastTimestamp": "2026-10-09T20:16:10Z", + "count": 3 +}` + +func toUnstructured(t *testing.T, payload string) *unstructured.Unstructured { + obj := &unstructured.Unstructured{} + assert.NoError(t, obj.UnmarshalJSON([]byte(payload))) + return obj +} + +func toJson(t *testing.T, obj interface{}) string { + bytes, err := json.Marshal(obj) + assert.NoError(t, err) + return string(bytes) +} + +func assertSloopCanReadEvent(t *testing.T, payload string) { + eventInfo, err := kubeextractor.ExtractEventInfo(payload) + assert.NoError(t, err) + assert.Equal(t, 3, eventInfo.Count) + assert.Equal(t, time.Date(2026, 10, 9, 20, 14, 58, 0, time.UTC), eventInfo.FirstTimestamp) + assert.Equal(t, time.Date(2026, 10, 9, 20, 16, 10, 0, time.UTC), eventInfo.LastTimestamp) + assert.Equal(t, "Pulling", eventInfo.Reason) + + involvedObject, err := kubeextractor.ExtractInvolvedObject(payload) + assert.NoError(t, err) + assert.Equal(t, "Pod", involvedObject.Kind) + assert.Equal(t, "default", involvedObject.Namespace) + assert.Equal(t, "my-app-5d8f7c6b9-x2x7q", involvedObject.Name) +} + +// Shows the bug: decoding the events.k8s.io/v1 shape into corev1.Event (what the typed informer does) drops the fields +func Test_typedEventDecodeDropsEventsV1Fields(t *testing.T) { + event := &corev1.Event{} + assert.NoError(t, json.Unmarshal([]byte(eventsV1Payload), event)) + payload := toJson(t, event) + + assert.Contains(t, payload, `"involvedObject":{}`) + assert.NotContains(t, payload, "Pulling image") + eventInfo, err := kubeextractor.ExtractEventInfo(payload) + assert.NoError(t, err) + assert.Equal(t, 0, eventInfo.Count) + assert.True(t, eventInfo.LastTimestamp.IsZero()) +} + +func Test_normalizeEvent_EventsV1(t *testing.T) { + event := toUnstructured(t, eventsV1Payload) + payload := toJson(t, normalizeEvent(event)) + + assertSloopCanReadEvent(t, payload) + result := map[string]interface{}{} + assert.NoError(t, json.Unmarshal([]byte(payload), &result)) + assert.Equal(t, "Pulling image \"busybox\"", result["message"]) + assert.Equal(t, map[string]interface{}{"component": "kubelet", "host": "node-1"}, result["source"]) + assert.Equal(t, "kubelet", result["reportingComponent"]) +} + +func Test_normalizeEvent_DoesNotModifyInput(t *testing.T) { + event := toUnstructured(t, eventsV1Payload) + before := event.DeepCopy() + _ = normalizeEvent(event) + assert.Equal(t, before, event) +} + +func Test_normalizeEvent_CoreV1Unchanged(t *testing.T) { + event := toUnstructured(t, coreV1Payload) + normalized := normalizeEvent(event) + + assert.Same(t, event, normalized) + assertSloopCanReadEvent(t, toJson(t, normalized)) +} + +func Test_normalizeEvent_SeriesAndEventTime(t *testing.T) { + event := toUnstructured(t, `{ + "apiVersion": "v1", + "kind": "Event", + "metadata": {"name": "my-app.1", "namespace": "default"}, + "regarding": {"kind": "Pod", "namespace": "default", "name": "my-app"}, + "reason": "BackOff", + "eventTime": "2026-10-09T20:14:58.123456Z", + "series": {"count": 7, "lastObservedTime": "2026-10-09T20:20:00.000000Z"} +}`) + eventInfo, err := kubeextractor.ExtractEventInfo(toJson(t, normalizeEvent(event))) + assert.NoError(t, err) + assert.Equal(t, 7, eventInfo.Count) + assert.Equal(t, time.Date(2026, 10, 9, 20, 14, 58, 123456000, time.UTC), eventInfo.FirstTimestamp) + assert.Equal(t, time.Date(2026, 10, 9, 20, 20, 0, 0, time.UTC), eventInfo.LastTimestamp) +} + +func Test_normalizeEvent_SingleEventTime(t *testing.T) { + event := toUnstructured(t, `{ + "apiVersion": "v1", + "kind": "Event", + "metadata": {"name": "my-app.1", "namespace": "default"}, + "regarding": {"kind": "Pod", "namespace": "default", "name": "my-app"}, + "reason": "Scheduled", + "eventTime": "2026-10-09T20:14:58.000000Z" +}`) + eventInfo, err := kubeextractor.ExtractEventInfo(toJson(t, normalizeEvent(event))) + assert.NoError(t, err) + assert.Equal(t, 1, eventInfo.Count) + assert.Equal(t, eventInfo.FirstTimestamp, eventInfo.LastTimestamp) + assert.False(t, eventInfo.LastTimestamp.IsZero()) +} + +func Test_processUpdate_EventsV1Event(t *testing.T) { + outChan := make(chan typed.KubeWatchResult, 5) + kw := &kubeWatcherImpl{protection: &sync.Mutex{}, outchan: outChan} + + kw.processUpdate("Event", toUnstructured(t, eventsV1Payload), &typed.KubeWatchResult{Kind: "Event"}, true) + result := <-outChan + assert.Equal(t, "Event", result.Kind) + assertSloopCanReadEvent(t, result.Payload) + + verifyChannelEmpty(t, outChan) +} + +// Excluding an unstructured object used to panic while logging its name +func Test_processUpdate_ExcludedUnstructured(t *testing.T) { + outChan := make(chan typed.KubeWatchResult, 5) + exclusionRules := map[string][]any{"Event": {map[string]any{"==": []any{map[string]any{"var": "reason"}, "Pulling"}}}} + kw := &kubeWatcherImpl{protection: &sync.Mutex{}, outchan: outChan, exclusionRules: exclusionRules} + + kw.processUpdate("Event", toUnstructured(t, eventsV1Payload), &typed.KubeWatchResult{Kind: "Event"}, true) + verifyChannelEmpty(t, outChan) +} + +// End to end: an events.k8s.io/v1 shaped Event from the API server reaches the out channel in core/v1 shape +func Test_bigPictureEventsV1Event(t *testing.T) { + dynamicClient := newTestDynamicClient() + outChan := make(chan typed.KubeWatchResult, 5) + kw, err := NewKubeWatcherSource(kubernetesFake.NewSimpleClientset(), dynamicClient, outChan, 30*time.Minute, false, 10*time.Second, "url", "", true, map[string][]any{}) + assert.NoError(t, err) + + _, err = dynamicClient.Resource(eventsGvr).Namespace("default").Create(context.TODO(), toUnstructured(t, eventsV1Payload), metav1.CreateOptions{}) + assert.NoError(t, err) + + select { + case result := <-outChan: + assert.Equal(t, "Event", result.Kind) + assertSloopCanReadEvent(t, result.Payload) + case <-time.After(10 * time.Second): + assert.Fail(t, "timed out waiting for the Event") + } + + kw.Stop() +} diff --git a/pkg/sloop/ingress/kubeclient.go b/pkg/sloop/ingress/kubeclient.go index 7eee066b..b72f96bd 100644 --- a/pkg/sloop/ingress/kubeclient.go +++ b/pkg/sloop/ingress/kubeclient.go @@ -9,6 +9,7 @@ package ingress import ( "github.com/golang/glog" + "k8s.io/client-go/dynamic" "k8s.io/client-go/kubernetes" _ "k8s.io/client-go/plugin/pkg/client/auth/azure" _ "k8s.io/client-go/plugin/pkg/client/auth/gcp" @@ -49,20 +50,9 @@ func GetKubernetesContext(masterURL string, kubeContextPreference string, privil // and returns a K8s client func MakeKubernetesClient(masterURL string, kubeContext string, privilegedAccess bool) (kubernetes.Interface, error) { glog.Infof("Creating k8sclient with user-defined config masterURL=%v, kubeContext=%v.", masterURL, kubeContext) - var config *rest.Config - var err error - if privilegedAccess { - clientConfig := getConfig(masterURL, kubeContext) - config, err = ClientConfig(clientConfig) - glog.Infof("Building k8sclient with context=%v, masterURL=%v, configFile=%v.", kubeContext, config.Host, clientConfig.ConfigAccess().GetLoadingPrecedence()) - } else { - glog.Infof("Creating Config using BuildConfigFromFlags") - config, err = BuildConfigFromFlags(masterURL, "") - if err != nil { - glog.Errorf("Cannot create config using BuildConfigFromFlags") - return nil, err - } - glog.Infof("Building k8sclient with context=%v, masterURL=%v.", kubeContext, config.Host) + config, err := makeRestConfig(masterURL, kubeContext, privilegedAccess) + if err != nil { + return nil, err } clientset, err := kubernetes.NewForConfig(config) @@ -75,6 +65,42 @@ func MakeKubernetesClient(masterURL string, kubeContext string, privilegedAccess return clientset, nil } +// MakeDynamicClient returns a dynamic client built from the same config as MakeKubernetesClient +func MakeDynamicClient(masterURL string, kubeContext string, privilegedAccess bool) (dynamic.Interface, error) { + config, err := makeRestConfig(masterURL, kubeContext, privilegedAccess) + if err != nil { + return nil, err + } + + dynamicClient, err := dynamic.NewForConfig(config) + if err != nil { + glog.Errorf("Cannot Initialize Kubernetes Dynamic Client API: %v", err) + return nil, err + } + return dynamicClient, nil +} + +func makeRestConfig(masterURL string, kubeContext string, privilegedAccess bool) (*rest.Config, error) { + if privilegedAccess { + clientConfig := getConfig(masterURL, kubeContext) + config, err := ClientConfig(clientConfig) + if err != nil { + return nil, err + } + glog.Infof("Building k8sclient with context=%v, masterURL=%v, configFile=%v.", kubeContext, config.Host, clientConfig.ConfigAccess().GetLoadingPrecedence()) + return config, nil + } + + glog.Infof("Creating Config using BuildConfigFromFlags") + config, err := BuildConfigFromFlags(masterURL, "") + if err != nil { + glog.Errorf("Cannot create config using BuildConfigFromFlags") + return nil, err + } + glog.Infof("Building k8sclient with context=%v, masterURL=%v.", kubeContext, config.Host) + return config, nil +} + func getConfig(masterURL string, kubeContext string) clientcmd.ClientConfig { loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() return clientcmd.NewNonInteractiveDeferredLoadingClientConfig( diff --git a/pkg/sloop/ingress/kubewatcher.go b/pkg/sloop/ingress/kubewatcher.go index 06663ba3..7c557be5 100644 --- a/pkg/sloop/ingress/kubewatcher.go +++ b/pkg/sloop/ingress/kubewatcher.go @@ -12,7 +12,6 @@ import ( "context" "encoding/json" "fmt" - "reflect" "strings" "sync" "sync/atomic" @@ -28,7 +27,9 @@ import ( "github.com/salesforce/sloop/pkg/sloop/kubeextractor" "github.com/salesforce/sloop/pkg/sloop/store/typed" "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" + "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/dynamic" "k8s.io/client-go/dynamic/dynamicinformer" @@ -59,8 +60,9 @@ type crdInformerInfo struct { } type kubeWatcherImpl struct { - informerFactory informers.SharedInformerFactory - stopChan chan struct{} + informerFactory informers.SharedInformerFactory + eventInformerFactory dynamicinformer.DynamicSharedInformerFactory + stopChan chan struct{} crdInformers map[crdGroupVersionResourceKind]*crdInformerInfo activeCrdInformer int64 @@ -74,6 +76,14 @@ type kubeWatcherImpl struct { exclusionRules map[string][]any } +// Events are watched with a dynamic (unstructured) informer instead of the typed corev1.Event informer. +// A typed informer decodes the response into corev1.Event, and encoding/json silently drops every field that +// is not in that struct. Some servers return Events on /api/v1 with the events.k8s.io/v1 field names +// (regarding, note, deprecatedCount, ...), which would be stored with an empty involvedObject and a count of 0, +// so they never show up in the event heatmap or in a resource's Events panel. Unstructured keeps all fields and +// normalizeEvent maps them to the core/v1 names Sloop reads. +var eventsGvr = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "events"} + var ( newCrdClient = func(kubeCfg *rest.Config) (clientset.Interface, error) { return clientset.NewForConfig(kubeCfg) } metricIngressGranularKubewatchcount = promauto.NewCounterVec(prometheus.CounterOpts{Name: "metric_ingress_event_kubewatchcount"}, []string{"namespace", "name", "kind", "reason", "type"}) @@ -84,14 +94,14 @@ var ( ) // Todo: Add additional parameters for filtering -func NewKubeWatcherSource(kubeClient kubernetes.Interface, outChan chan typed.KubeWatchResult, resync time.Duration, includeCrds bool, crdRefreshInterval time.Duration, masterURL string, kubeContext string, enableGranularMetrics bool, exclusionRules map[string][]any) (KubeWatcher, error) { +func NewKubeWatcherSource(kubeClient kubernetes.Interface, dynamicClient dynamic.Interface, outChan chan typed.KubeWatchResult, resync time.Duration, includeCrds bool, crdRefreshInterval time.Duration, masterURL string, kubeContext string, enableGranularMetrics bool, exclusionRules map[string][]any) (KubeWatcher, error) { kw := &kubeWatcherImpl{resync: resync, protection: &sync.Mutex{}} kw.stopChan = make(chan struct{}) kw.crdInformers = make(map[crdGroupVersionResourceKind]*crdInformerInfo) kw.outchan = outChan kw.exclusionRules = exclusionRules - kw.startWellKnownInformers(kubeClient, enableGranularMetrics) + kw.startWellKnownInformers(kubeClient, dynamicClient, enableGranularMetrics) if includeCrds { err := kw.startCustomInformers(masterURL, kubeContext, enableGranularMetrics) if err != nil { @@ -105,8 +115,9 @@ func NewKubeWatcherSource(kubeClient kubernetes.Interface, outChan chan typed.Ku return kw, nil } -func (i *kubeWatcherImpl) startWellKnownInformers(kubeclient kubernetes.Interface, enableGranularMetrics bool) { +func (i *kubeWatcherImpl) startWellKnownInformers(kubeclient kubernetes.Interface, dynamicClient dynamic.Interface, enableGranularMetrics bool) { i.informerFactory = informers.NewSharedInformerFactory(kubeclient, i.resync) + i.eventInformerFactory = dynamicinformer.NewDynamicSharedInformerFactory(dynamicClient, i.resync) i.informerFactory.Apps().V1().DaemonSets().Informer().AddEventHandler(i.getEventHandlerForResource("DaemonSet", enableGranularMetrics)) i.informerFactory.Apps().V1().Deployments().Informer().AddEventHandler(i.getEventHandlerForResource("Deployment", enableGranularMetrics)) @@ -114,7 +125,8 @@ func (i *kubeWatcherImpl) startWellKnownInformers(kubeclient kubernetes.Interfac i.informerFactory.Apps().V1().StatefulSets().Informer().AddEventHandler(i.getEventHandlerForResource("StatefulSet", enableGranularMetrics)) i.informerFactory.Core().V1().ConfigMaps().Informer().AddEventHandler(i.getEventHandlerForResource("ConfigMap", enableGranularMetrics)) i.informerFactory.Core().V1().Endpoints().Informer().AddEventHandler(i.getEventHandlerForResource("Endpoint", enableGranularMetrics)) - i.informerFactory.Core().V1().Events().Informer().AddEventHandler(i.getEventHandlerForResource("Event", enableGranularMetrics)) + // Not i.informerFactory.Core().V1().Events(): see eventsGvr + i.eventInformerFactory.ForResource(eventsGvr).Informer().AddEventHandler(i.getEventHandlerForResource("Event", enableGranularMetrics)) i.informerFactory.Autoscaling().V1().HorizontalPodAutoscalers().Informer().AddEventHandler(i.getEventHandlerForResource("HorizontalPodAutoscaler", enableGranularMetrics)) i.informerFactory.Batch().V1().Jobs().Informer().AddEventHandler(i.getEventHandlerForResource("Job", enableGranularMetrics)) i.informerFactory.Core().V1().Namespaces().Informer().AddEventHandler(i.getEventHandlerForResource("Namespace", enableGranularMetrics)) @@ -128,6 +140,7 @@ func (i *kubeWatcherImpl) startWellKnownInformers(kubeclient kubernetes.Interfac i.informerFactory.Storage().V1().StorageClasses().Informer().AddEventHandler(i.getEventHandlerForResource("StorageClass", enableGranularMetrics)) i.informerFactory.Admissionregistration().V1().MutatingWebhookConfigurations().Informer().AddEventHandler(i.getEventHandlerForResource("MutatingWebhookConfiguration", enableGranularMetrics)) i.informerFactory.Start(i.stopChan) + i.eventInformerFactory.Start(i.stopChan) } func (i *kubeWatcherImpl) startCustomInformers(masterURL string, kubeContext string, enableGranularMetrics bool) error { @@ -304,6 +317,11 @@ func (i *kubeWatcherImpl) reportUpdate(kind string, enableGranularMetrics bool) } func (i *kubeWatcherImpl) processUpdate(kind string, obj interface{}, watchResult *typed.KubeWatchResult, enableGranularmetrics bool) { + // Rewrite events.k8s.io/v1 shaped Events to core/v1 before they are stored, so kubeextractor + // (count, timestamps, involvedObject), the event count table and the UI can read them + if event, ok := obj.(*unstructured.Unstructured); ok && kind == "Event" { + obj = normalizeEvent(event) + } resourceJson, err := i.getResourceAsJsonString(kind, obj) if err != nil { glog.Error(err) @@ -313,7 +331,12 @@ func (i *kubeWatcherImpl) processUpdate(kind string, obj interface{}, watchResul eventExcluded := i.eventExcluded(kind, resourceJson) if eventExcluded { - objName := reflect.ValueOf(obj).Elem().FieldByName("ObjectMeta").FieldByName("Name") + // meta.Accessor works for typed and unstructured objects (Events and CRDs); reflecting on an + // ObjectMeta field panics for *unstructured.Unstructured + objName := "" + if accessor, err := meta.Accessor(obj); err == nil { + objName = accessor.GetName() + } glog.V(2).Infof("Event for object excluded: %s/%s", kind, objName) return } diff --git a/pkg/sloop/ingress/kubewatcher_test.go b/pkg/sloop/ingress/kubewatcher_test.go index d0fe6e08..4b74effe 100644 --- a/pkg/sloop/ingress/kubewatcher_test.go +++ b/pkg/sloop/ingress/kubewatcher_test.go @@ -66,6 +66,11 @@ func newTestCrdClient(reaction func(_ k8sTesting.Action) (bool, runtime.Object, } } +// newTestDynamicClient - returns a fake dynamic client that can list and watch Events +func newTestDynamicClient() *dynamicFake.FakeDynamicClient { + return dynamicFake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), map[schema.GroupVersionResource]string{eventsGvr: "EventList"}) +} + // This test (test-harness) exercises the kubewatcher from the client perspective // - start a kubewatcher // - force a k8s event in the system @@ -82,7 +87,7 @@ func Test_bigPicture(t *testing.T) { kubeContext := "" // empty string makes things work enableGranularMetrics := true exclusionRules := map[string][]any{} - kw, err := NewKubeWatcherSource(kubeClient, outChan, resync, includeCrds, time.Duration(10*time.Second), masterURL, kubeContext, enableGranularMetrics, exclusionRules) + kw, err := NewKubeWatcherSource(kubeClient, newTestDynamicClient(), outChan, resync, includeCrds, time.Duration(10*time.Second), masterURL, kubeContext, enableGranularMetrics, exclusionRules) assert.NoError(t, err) // create service and await corresponding event @@ -125,7 +130,7 @@ func Test_bigPictureWithExclusionRules(t *testing.T) { }, } - kw, err := NewKubeWatcherSource(kubeClient, outChan, resync, includeCrds, time.Duration(10*time.Second), masterURL, kubeContext, enableGranularMetrics, exclusionRules) + kw, err := NewKubeWatcherSource(kubeClient, newTestDynamicClient(), outChan, resync, includeCrds, time.Duration(10*time.Second), masterURL, kubeContext, enableGranularMetrics, exclusionRules) assert.NoError(t, err) // create namespace diff --git a/pkg/sloop/server/server.go b/pkg/sloop/server/server.go index a33a0949..d36d696a 100644 --- a/pkg/sloop/server/server.go +++ b/pkg/sloop/server/server.go @@ -103,8 +103,13 @@ func RealMain() error { if err != nil { return errors.Wrap(err, "failed to create kubernetes client") } + // Used to watch Events as unstructured objects, built from the same config as kubeClient + dynamicClient, err := ingress.MakeDynamicClient(conf.ApiServerHost, kubeContext, conf.PrivilegedAccess) + if err != nil { + return errors.Wrap(err, "failed to create kubernetes dynamic client") + } - kubeWatcherSource, err = ingress.NewKubeWatcherSource(kubeClient, kubeWatchChan, conf.KubeWatchResyncInterval, conf.WatchCrds, conf.CrdRefreshInterval, conf.ApiServerHost, kubeContext, conf.EnableGranularMetrics, conf.ExclusionRules) + kubeWatcherSource, err = ingress.NewKubeWatcherSource(kubeClient, dynamicClient, kubeWatchChan, conf.KubeWatchResyncInterval, conf.WatchCrds, conf.CrdRefreshInterval, conf.ApiServerHost, kubeContext, conf.EnableGranularMetrics, conf.ExclusionRules) if err != nil { return errors.Wrap(err, "failed to initialize kubeWatcher") }