Merge pull request #1325 from DigitalVeer/node-event-uid

Set the Node UID on recorded events

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

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",

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