UPSTREAM: <carry>: 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,21 +281,6 @@ 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,18 +182,3 @@ 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/cacher.go

@@ -506,6 +506,18 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 	} else {
 		readyGeneration, err = c.ready.waitAndReadGeneration(ctx)
 		if err != nil {
+			if err == ErrStorageInitializing || strings.HasPrefix(err.Error(), "storage is (re)initializing") {
+				// Add audit annotations with OpenShift prefix
+				audit.AddAuditAnnotation(ctx, "openshift.io/watch-reject-reason", "storage_initializing")
+
+				msg := err.Error()
+				if len(msg) > 1024 {
+					msg = msg[:1024] + "…"
+				}
+				audit.AddAuditAnnotation(ctx, "openshift.io/watch-reject-message", msg)
+			}
+
+			// Return HTTP 503 (ServiceUnavailable) to the client
 			return nil, errors.NewServiceUnavailable(err.Error())
 		}
 	}

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

@@ -23,7 +23,6 @@ import (
 	"sync"
 	"time"
 
-	"k8s.io/apiserver/pkg/audit"
 	"k8s.io/utils/clock"
 )
 
@@ -87,6 +86,7 @@ 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()
@@ -95,22 +95,18 @@ 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()
-
-		if err != nil {
-			if errors.Is(err, ErrStorageInitializing) {
-				audit.AddAuditAnnotationForRejectWithReason(ctx, "storage_initializing")
-				audit.AddAuditAnnotationForRejectMessage(ctx, err.Error())
-			}
-			return 0, err
-		}
-
-		return generation, nil
+		return generation, err
 	}
 }