Add audit annotation for watch rejections due to storage initialization

Signed-off-by: Shaza Aldawamneh <[email protected]>

Author: ShazaAldawamneh

staging/src/k8s.io/apiserver/pkg/audit/context.go

@@ -281,6 +281,21 @@ func AddAuditAnnotation(ctx context.Context, key, value string) {
 	addAuditAnnotationLocked(ac, key, value)
 }
 
+// AddAuditAnnotationForRejectWithReason records an audit annotation
+// indicating that a watch request was rejected for the given reason.
+func AddAuditAnnotationForRejectWithReason(ctx context.Context, reason string) {
+	AddAuditAnnotation(ctx, "audit.k8s.io/watch-reject-reason", reason)
+}
+
+// AddAuditAnnotationForRejectMessage adds a human-readable message annotation
+// truncated to 1024 characters to avoid excessive log size.
+func AddAuditAnnotationForRejectMessage(ctx context.Context, msg string) {
+	if len(msg) > 1024 {
+		msg = msg[:1024] + "…"
+	}
+	AddAuditAnnotation(ctx, "audit.k8s.io/watch-reject-message", msg)
+}
+
 // AddAuditAnnotations is a bulk version of AddAuditAnnotation. Refer to AddAuditAnnotation for
 // restrictions on when this can be called.
 // keysAndValues are the key-value pairs to add, and must have an even number of items.

staging/src/k8s.io/apiserver/pkg/audit/context_test.go

@@ -182,3 +182,18 @@ func withAuditContextAndLevel(ctx context.Context, t *testing.T, l auditinternal
 	}
 	return ctx
 }
+func TestAddAuditAnnotationForRejectWithReason(t *testing.T) {
+	ctx := WithAuditContext(context.Background())
+	AddAuditAnnotationForRejectWithReason(ctx, "storage_initializing")
+	AddAuditAnnotationForRejectMessage(ctx, "storage is (re)initializing")
+
+	ac := AuditContextFrom(ctx)
+	annotations := ac.GetEventAnnotations()
+
+	if got := annotations["audit.k8s.io/watch-reject-reason"]; got != "storage_initializing" {
+		t.Errorf("expected reason 'storage_initializing', got %q", got)
+	}
+	if got := annotations["audit.k8s.io/watch-reject-message"]; got != "storage is (re)initializing" {
+		t.Errorf("expected message 'storage is (re)initializing', got %q", got)
+	}
+}

staging/src/k8s.io/apiserver/pkg/storage/cacher/ready.go

@@ -18,13 +18,20 @@ package cacher
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
 
+	"k8s.io/apiserver/pkg/audit"
 	"k8s.io/utils/clock"
 )
 
+// ErrStorageInitializing is returned when the cacher is still initializing.
+// This allows callers to detect this specific condition and handle it
+// (e.g., add an audit annotation or return HTTP 429).
+var ErrStorageInitializing = errors.New("storage is (re)initializing")
+
 type status int
 
 const (
@@ -80,7 +87,6 @@ func (r *ready) wait(ctx context.Context) error {
 // of times we entered ready state if Ready and error otherwise.
 func (r *ready) waitAndReadGeneration(ctx context.Context) (int, error) {
 	for {
-		// r.done() only blocks if state is Pending
 		select {
 		case <-ctx.Done():
 			return 0, ctx.Err()
@@ -89,18 +95,22 @@ func (r *ready) waitAndReadGeneration(ctx context.Context) (int, error) {
 
 		r.lock.RLock()
 		if r.state == Pending {
-			// since we allow to switch between the states Pending and Ready
-			// if there is a quick transition from Pending -> Ready -> Pending
-			// a process that was waiting can get unblocked and see a Pending
-			// state again. If the state is Pending we have to wait again to
-			// avoid an inconsistent state on the system, with some processes not
-			// waiting despite the state moved back to Pending.
 			r.lock.RUnlock()
 			continue
 		}
+
 		generation, err := r.readGenerationLocked()
 		r.lock.RUnlock()
-		return generation, err
+
+		if err != nil {
+			if errors.Is(err, ErrStorageInitializing) {
+				audit.AddAuditAnnotationForRejectWithReason(ctx, "storage_initializing")
+				audit.AddAuditAnnotationForRejectMessage(ctx, err.Error())
+			}
+			return 0, err
+		}
+
+		return generation, nil
 	}
 }
 
@@ -122,10 +132,9 @@ func (r *ready) readGenerationLocked() (int, error) {
 	switch r.state {
 	case Pending:
 		if r.lastErr == nil {
-			return 0, fmt.Errorf("storage is (re)initializing")
-		} else {
-			return 0, fmt.Errorf("storage is (re)initializing: %w", r.lastErr)
+			return 0, ErrStorageInitializing
 		}
+		return 0, fmt.Errorf("%w: %v", ErrStorageInitializing, r.lastErr)
 	case Ready:
 		return r.generation, nil
 	case Stopped: