From c1d461ff76fe059be2f1c94b96c70cd6473f4a95 Mon Sep 17 00:00:00 2001 From: Veer Singh <8453348+digitalveer@users.noreply.github.com> Date: Thu, 23 Jul 2026 15:58:27 -0700 Subject: [PATCH] Set the Node UID on recorded events Recorded events refer to the node with an ObjectReference that has no UID. The client already reads the Node object at startup to check that kube-apiserver is ready, and this change stores the UID from that read. The node status patch returns the Node object, so the client stores that UID too. This change removes the TODO in problem_client.go. --- pkg/exporters/k8sexporter/k8s_exporter.go | 4 +- .../problemclient/problem_client.go | 50 +++++-- .../problemclient/problem_client_test.go | 139 +++++++++++++++++- 3 files changed, 178 insertions(+), 15 deletions(-) diff --git a/pkg/exporters/k8sexporter/k8s_exporter.go b/pkg/exporters/k8sexporter/k8s_exporter.go index 17a699601..d47e04ff0 100644 --- a/pkg/exporters/k8sexporter/k8s_exporter.go +++ b/pkg/exporters/k8sexporter/k8s_exporter.go @@ -123,7 +123,9 @@ func (ke *k8sExporter) startHTTPReporting(npdo *options.NodeProblemDetectorOptio func waitForAPIServerReadyWithTimeout(ctx context.Context, c problemclient.Client, npdo *options.NodeProblemDetectorOptions) error { return wait.PollUntilContextTimeout(ctx, npdo.APIServerWaitInterval, npdo.APIServerWaitTimeout, true, func(ctx context.Context) (done bool, err error) { // If NPD can get the node object from kube-apiserver, the server is - // ready and the RBAC permission is set correctly. + // ready and the RBAC permission is set correctly. The call also caches + // the Node UID that recorded events refer to. If this check fails, the + // first node status patch caches the UID instead. if _, err := c.GetNode(ctx); err != nil { klog.Errorf("Can't get node object: %v", err) return false, err diff --git a/pkg/exporters/k8sexporter/problemclient/problem_client.go b/pkg/exporters/k8sexporter/problemclient/problem_client.go index fe5fd1820..4c4d63182 100644 --- a/pkg/exporters/k8sexporter/problemclient/problem_client.go +++ b/pkg/exporters/k8sexporter/problemclient/problem_client.go @@ -23,10 +23,12 @@ import ( "net/url" "os" "path/filepath" + "sync/atomic" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" clientset "k8s.io/client-go/kubernetes" typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" "k8s.io/client-go/tools/record" @@ -52,11 +54,14 @@ type Client interface { } type nodeProblemClient struct { - nodeName string - client typedcorev1.CoreV1Interface - clock clock.Clock - recorders map[string]record.EventRecorder - nodeRef *v1.ObjectReference + nodeName string + client typedcorev1.CoreV1Interface + clock clock.Clock + recorders map[string]record.EventRecorder + // nodeRef identifies the node in recorded events. + nodeRef *v1.ObjectReference + // cachedNodeRef holds a copy of nodeRef with the Node UID set. + cachedNodeRef atomic.Pointer[v1.ObjectReference] eventNamespace string } @@ -113,7 +118,10 @@ func (c *nodeProblemClient) SetConditions(ctx context.Context, newConditions []v return true }, func() error { - _, err := c.client.Nodes().PatchStatus(ctx, c.nodeName, patch) + node, err := c.client.Nodes().PatchStatus(ctx, c.nodeName, patch) + if err == nil { + c.cacheNodeRef(node.UID) + } return err }, ) @@ -126,13 +134,38 @@ func (c *nodeProblemClient) Eventf(eventType, source, reason, messageFmt string, recorder = getEventRecorder(c.client, c.eventNamespace, c.nodeName, source) c.recorders[source] = recorder } - recorder.Eventf(c.nodeRef, eventType, reason, messageFmt, args...) + recorder.Eventf(c.nodeRefWithUID(), eventType, reason, messageFmt, args...) } func (c *nodeProblemClient) GetNode(ctx context.Context) (*v1.Node, error) { // To reduce the load on APIServer & etcd, we are serving GET operations from // apiserver cache (the data might be slightly delayed). - return c.client.Nodes().Get(ctx, c.nodeName, metav1.GetOptions{ResourceVersion: "0"}) + node, err := c.client.Nodes().Get(ctx, c.nodeName, metav1.GetOptions{ResourceVersion: "0"}) + if err == nil { + c.cacheNodeRef(node.UID) + } + return node, err +} + +// nodeRefWithUID returns the node reference to record events against. +// +// The UID is resolved once. If the Node object is deleted and created again +// while node-problem-detector runs, events keep the first UID until restart. +func (c *nodeProblemClient) nodeRefWithUID() *v1.ObjectReference { + if ref := c.cachedNodeRef.Load(); ref != nil { + return ref + } + return c.nodeRef +} + +// cacheNodeRef stores a copy of nodeRef with the given UID set. +func (c *nodeProblemClient) cacheNodeRef(uid types.UID) { + if uid == "" || c.cachedNodeRef.Load() != nil { + return + } + ref := *c.nodeRef + ref.UID = uid + c.cachedNodeRef.CompareAndSwap(nil, &ref) } // generatePatch generates condition patch @@ -154,7 +187,6 @@ func getEventRecorder(c typedcorev1.CoreV1Interface, namespace, nodeName, source } func getNodeRef(namespace, nodeName string) *v1.ObjectReference { - // TODO(random-liu): Get node to initialize the node reference return &v1.ObjectReference{ APIVersion: "v1", Kind: "Node", diff --git a/pkg/exporters/k8sexporter/problemclient/problem_client_test.go b/pkg/exporters/k8sexporter/problemclient/problem_client_test.go index 483425cf5..439afb98e 100644 --- a/pkg/exporters/k8sexporter/problemclient/problem_client_test.go +++ b/pkg/exporters/k8sexporter/problemclient/problem_client_test.go @@ -17,28 +17,33 @@ limitations under the License. package problemclient import ( + "context" "encoding/json" "fmt" + "net/http" + "net/http/httptest" "testing" "time" "github.com/stretchr/testify/assert" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" + "k8s.io/client-go/rest" "k8s.io/client-go/tools/record" testclock "k8s.io/utils/clock/testing" ) const ( - testSource = "test" - testNode = "test-node" + testSource = "test" + testNode = "test-node" + testNodeUID = "11111111-1111-1111-1111-111111111111" ) func newFakeProblemClient() *nodeProblemClient { return &nodeProblemClient{ - nodeName: testNode, - // There is no proper fake for *client.Client for now - // TODO(random-liu): Add test for SetConditions when we have good fake for *client.Client + nodeName: testNode, clock: testclock.NewFakeClock(time.Now()), recorders: make(map[string]record.EventRecorder), nodeRef: getNodeRef("", testNode), @@ -93,3 +98,127 @@ func TestNodeRefHasAPIVersionV1(t *testing.T) { t.Errorf("expected nodeRef.APIVersion to be 'v1', got %q", client.nodeRef.APIVersion) } } + +func TestNodeRefWithUID(t *testing.T) { + client := newFakeProblemClient() + + if got := client.nodeRefWithUID().UID; got != "" { + t.Errorf("expected no UID before the node is read, got %q", got) + } + + client.cacheNodeRef(testNodeUID) + + if got := client.nodeRefWithUID().UID; got != testNodeUID { + t.Errorf("expected UID %q, got %q", testNodeUID, got) + } + if got := client.nodeRef.UID; got != "" { + t.Errorf("expected the shared nodeRef to keep no UID, got %q", got) + } +} + +func TestCacheNodeRefIgnoresEmptyUID(t *testing.T) { + client := newFakeProblemClient() + + client.cacheNodeRef("") + + if got := client.nodeRefWithUID().UID; got != "" { + t.Errorf("expected no UID, got %q", got) + } +} + +func TestCacheNodeRefKeepsFirstUID(t *testing.T) { + client := newFakeProblemClient() + + client.cacheNodeRef(testNodeUID) + client.cacheNodeRef("22222222-2222-2222-2222-222222222222") + + if got := client.nodeRefWithUID().UID; got != testNodeUID { + t.Errorf("expected the first UID %q, got %q", testNodeUID, got) + } +} + +// capturingRecorder records the object that Eventf reports the event against. +type capturingRecorder struct { + object runtime.Object +} + +func (r *capturingRecorder) Event(object runtime.Object, eventType, reason, message string) { + r.object = object +} + +func (r *capturingRecorder) Eventf(object runtime.Object, eventType, reason, messageFmt string, args ...interface{}) { + r.object = object +} + +func (r *capturingRecorder) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventType, reason, messageFmt string, args ...interface{}) { + r.object = object +} + +func TestEventfReportsNodeUID(t *testing.T) { + recorder := &capturingRecorder{} + client := newFakeProblemClient() + client.recorders[testSource] = recorder + client.cacheNodeRef(testNodeUID) + + client.Eventf(v1.EventTypeWarning, testSource, "test reason", "test message") + + ref, ok := recorder.object.(*v1.ObjectReference) + if !ok { + t.Fatalf("expected an *v1.ObjectReference, got %T", recorder.object) + } + if ref.UID != testNodeUID { + t.Errorf("expected the reported event to carry UID %q, got %q", testNodeUID, ref.UID) + } + if ref.Name != testNode { + t.Errorf("expected the reported event to carry name %q, got %q", testNode, ref.Name) + } +} + +// newProblemClientAgainstAPI returns a client that talks to a server which +// always answers with the test node. +func newProblemClientAgainstAPI(t *testing.T) *nodeProblemClient { + t.Helper() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + node := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: testNode, UID: testNodeUID}} + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(node); err != nil { + t.Errorf("failed to encode the node: %v", err) + } + })) + t.Cleanup(server.Close) + + coreClient, err := typedcorev1.NewForConfig(&rest.Config{Host: server.URL}) + if err != nil { + t.Fatalf("failed to create the core client: %v", err) + } + + client := newFakeProblemClient() + client.client = coreClient + return client +} + +func TestGetNodeCachesNodeUID(t *testing.T) { + client := newProblemClientAgainstAPI(t) + + if _, err := client.GetNode(context.Background()); err != nil { + t.Fatalf("GetNode returned an error: %v", err) + } + + if got := client.nodeRefWithUID().UID; got != testNodeUID { + t.Errorf("expected GetNode to cache UID %q, got %q", testNodeUID, got) + } +} + +func TestSetConditionsCachesNodeUID(t *testing.T) { + client := newProblemClientAgainstAPI(t) + + conditions := []v1.NodeCondition{{Type: "TestType", Status: v1.ConditionTrue}} + if err := client.SetConditions(context.Background(), conditions); err != nil { + t.Fatalf("SetConditions returned an error: %v", err) + } + + if got := client.nodeRefWithUID().UID; got != testNodeUID { + t.Errorf("expected SetConditions to cache UID %q, got %q", testNodeUID, got) + } +}