@W-24431985: Watch Events with an unstructured informer so events.k8s.io/v1 fields are not dropped

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
+}

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()
+}

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(