UPSTREAM: <carry>: add audit annotations to track etcd state

staging/src/k8s.io/apiserver/pkg/storage/etcd3/etcd3retry/retry_etcdclient.go

@@ -2,6 +2,8 @@ package etcd3retry
 
 import (
 	"context"
+	"k8s.io/apiserver/pkg/audit"
+	"strings"
 	"time"
 
 	etcdrpc "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
@@ -32,11 +34,18 @@ func NewRetryingEtcdStorage(delegate storage.Interface) storage.Interface {
 	return &retryClient{Interface: delegate}
 }
 
+func addEtcdAccessAuditAnnotation(ctx context.Context) {
+	// add an audit annotation indicating we reached out to etcd. This allows our post-processing to exclude requests
+	// that don't attempt to access etcd from, "how reliably is etcd" calculations.
+	audit.AddAuditAnnotation(ctx, "apiserver.internal.openshift.io/etcd-access", time.Now().Format(time.RFC3339))
+}
+
 // Create adds a new object at a key unless it already exists. 'ttl' is time-to-live
 // in seconds (0 means forever). If no error is returned and out is not nil, out will be
 // set to the read value from database.
 func (c *retryClient) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
 	return OnError(ctx, DefaultRetry, IsRetriableEtcdError, func() error {
+		addEtcdAccessAuditAnnotation(ctx)
 		return c.Interface.Create(ctx, key, obj, out, ttl)
 	})
 }
@@ -45,6 +54,7 @@ func (c *retryClient) Create(ctx context.Context, key string, obj, out runtime.O
 // If key didn't exist, it will return NotFound storage error.
 func (c *retryClient) Delete(ctx context.Context, key string, out runtime.Object, preconditions *storage.Preconditions, validateDeletion storage.ValidateObjectFunc, cachedExistingObject runtime.Object) error {
 	return OnError(ctx, DefaultRetry, IsRetriableEtcdError, func() error {
+		addEtcdAccessAuditAnnotation(ctx)
 		return c.Interface.Delete(ctx, key, out, preconditions, validateDeletion, cachedExistingObject)
 	})
 }
@@ -59,6 +69,7 @@ func (c *retryClient) Delete(ctx context.Context, key string, out runtime.Object
 func (c *retryClient) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) {
 	var ret watch.Interface
 	err := OnError(ctx, DefaultRetry, IsRetriableEtcdError, func() error {
+		addEtcdAccessAuditAnnotation(ctx)
 		var innerErr error
 		ret, innerErr = c.Interface.Watch(ctx, key, opts)
 		return innerErr
@@ -73,6 +84,7 @@ func (c *retryClient) Watch(ctx context.Context, key string, opts storage.ListOp
 // match 'opts.ResourceVersion' according 'opts.ResourceVersionMatch'.
 func (c *retryClient) Get(ctx context.Context, key string, opts storage.GetOptions, objPtr runtime.Object) error {
 	return OnError(ctx, DefaultRetry, IsRetriableEtcdError, func() error {
+		addEtcdAccessAuditAnnotation(ctx)
 		return c.Interface.Get(ctx, key, opts, objPtr)
 	})
 }
@@ -85,6 +97,7 @@ func (c *retryClient) Get(ctx context.Context, key string, opts storage.GetOptio
 // match 'opts.ResourceVersion' according 'opts.ResourceVersionMatch'.
 func (c *retryClient) GetList(ctx context.Context, key string, opts storage.ListOptions, listObj runtime.Object) error {
 	return OnError(ctx, DefaultRetry, IsRetriableEtcdError, func() error {
+		addEtcdAccessAuditAnnotation(ctx)
 		return c.Interface.GetList(ctx, key, opts, listObj)
 	})
 }
@@ -126,6 +139,7 @@ func (c *retryClient) GetList(ctx context.Context, key string, opts storage.List
 func (c *retryClient) GuaranteedUpdate(ctx context.Context, key string, destination runtime.Object, ignoreNotFound bool,
 	preconditions *storage.Preconditions, tryUpdate storage.UpdateFunc, cachedExistingObject runtime.Object) error {
 	return OnError(ctx, DefaultRetry, IsRetriableEtcdError, func() error {
+		addEtcdAccessAuditAnnotation(ctx)
 		return c.Interface.GuaranteedUpdate(ctx, key, destination, ignoreNotFound, preconditions, tryUpdate, cachedExistingObject)
 	})
 }
@@ -153,6 +167,8 @@ func OnError(ctx context.Context, backoff wait.Backoff, retriable func(error) (s
 	var retry bool
 	var retryCounter int
 	err := backoffWithRequestContext(ctx, backoff, func() (bool, error) {
+		startTime := time.Now()
+
 		err := fn()
 		if retry {
 			klog.V(1).Infof("etcd retry - counter: %v, lastErrLabel: %s lastError: %v, error: %v", retryCounter, lastErrLabel, lastErr, err)
@@ -162,6 +178,12 @@ func OnError(ctx context.Context, backoff wait.Backoff, retriable func(error) (s
 			return true, nil
 		}
 
+		// add an audit annotation if we hit a no leader condition so we can track this failure in post-processing CI steps.
+		// We only mark the first time through.  Hopefully there's enough traffic that it doesn't matter
+		if strings.Contains(err.Error(), "no leader") {
+			audit.AddAuditAnnotation(ctx, "apiserver.internal.openshift.io/no-leader", startTime.Format(time.RFC3339))
+		}
+
 		lastErrLabel, retry = retriable(err)
 		if retry {
 			lastErr = err