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