Repository: kubernetes/kubernetes
File: staging/src/k8s.io/apiserver/pkg/storage/cacher/delegator.go (Go)

Code before:
		klog.V(3).InfoS("Cache consistency check passed", "group", c.groupResource.Group, "resource", c.groupResource.Resource, "resourceVersion", digests.ResourceVersion, "digest", digests.CacheDigest)
		metrics.StorageConsistencyCheckTotal.WithLabelValues(c.groupResource.Group, c.groupResource.Resource, "success").Inc()
		c.cacher.MarkConsistent(true)
		return
	}

	errMsgKVs := []interface{}{"group", c.groupResource.Group, "resource", c.groupResource.Resource, "resourceVersion", digests.ResourceVersion, "etcdDigest", digests.EtcdDigest, "cacheDigest", digests.CacheDigest}
	errMsgKVs = append(errMsgKVs, enrichErrorMessageKVs(digests.EtcdKeys, digests.CacheKeys)...)

	klog.ErrorS(nil, "Cache consistency check failed", errMsgKVs...)
	metrics.StorageConsistencyCheckTotal.WithLabelValues(c.groupResource.Group, c.groupResource.Resource, "failure").Inc()
	if panicOnCacheInconsistency {
		panic(fmt.Sprintf("Cache consistency check failed, group: %q, resource: %q, resourceVersion: %q, etcdDigest: %q, cacheDigest: %q", c.groupResource.Group, c.groupResource.Resource, digests.ResourceVersion, digests.EtcdDigest, digests.CacheDigest))
	}
	c.cacher.MarkConsistent(false)
}

func enrichErrorMessageKVs(etcdKeyList, cacheKeyList *[]namespaceNameRV) []interface{} {
	etcdKeySet := sets.New[namespaceNameRV](*etcdKeyList...)
	cacheKeySet := sets.New[namespaceNameRV](*cacheKeyList...)
	errMsgKVs := []interface{}{"etcdLength", len(*etcdKeyList), "cacheLength", len(*cacheKeyList)}
	missing := etcdKeySet.Difference(cacheKeySet).UnsortedList()
	additional := cacheKeySet.Difference(etcdKeySet).UnsortedList()
	// If the items are different, we log the different items without considering ordering.
	if len(missing) != 0 || len(additional) != 0 {
		errMsgKVs = append(errMsgKVs, "missingKeysInCache", missing, "additionalKeysInCache", additional)
	} else {
		// When the items are same, but orderings are different, we find the 1st item that mismatch.
		for i, keyCache := range *cacheKeyList {
			if (*etcdKeyList)[i] != keyCache {
				errMsgKVs = append(errMsgKVs, "first mismatching item in etcd list", (*etcdKeyList)[i], "first mismatching item in cache list", (*cacheKeyList)[i])
				break
			}
		}
	}
	return errMsgKVs
}

func (c *consistencyChecker) calculateDigests(ctx context.Context) (*storageDigest, error) {
	if !c.cacher.Ready() {
		return nil, fmt.Errorf("cache is not ready")
	}
	cacheDigest, cacheResourceVersion, cacheKeyList, err := c.calculateStoreDigest(ctx, c.cacher, "0", 0)
	if err != nil {
		return nil, fmt.Errorf("failed calculating cache digest: %w", err)
	}
	etcdDigest, etcdResourceVersion, etcdKeyList, err := c.calculateStoreDigest(ctx, c.etcd, cacheResourceVersion, storageWatchListPageSize)
	if err != nil {
		return nil, fmt.Errorf("failed calculating etcd digest: %w", err)
	}
	if cacheResourceVersion != etcdResourceVersion {

Review comment (bug):
By keys we usually thing about (namespace, name) tuple, not (namespace, name, RV) tripple. If there is a missing key RV we would expect to get error that RV is incorrect, not that it's missing.

Code after:
	}
	klog.ErrorS(nil, "Cache consistency check failed", "group", c.groupResource.Group, "resource", c.groupResource.Resource, "resourceVersion", digests.ResourceVersion, "etcdDigest", digests.EtcdDigest, "cacheDigest", digests.CacheDigest, "diffDetail", digests.DiffDetail)
	metrics.StorageConsistencyCheckTotal.WithLabelValues(c.groupResource.Group, c.groupResource.Resource, "failure").Inc()
	if panicOnCacheInconsistency {
		panic(fmt.Sprintf("Cache consistency check failed, group: %q, resource: %q, resourceVersion: %q, etcdDigest: %q, cacheDigest: %q, diffDetail: %v", c.groupResource.Group, c.groupResource.Resource, digests.ResourceVersion, digests.EtcdDigest, digests.CacheDigest, digests.DiffDetail))
	}
	c.cacher.MarkConsistent(false)
}

func (c *consistencyChecker) calculateDigests(ctx context.Context) (*storageDigest, error) {
	if !c.cacher.Ready() {
		return nil, fmt.Errorf("cache is not ready")
	}
	// variables for tracking the diff in consistency checker
	cacheItems := []namespaceNameRV{}
	var foundDiff *diffDetail
	var etcdIndex int

	cacheDigest, cacheResourceVersion, err := c.calculateStoreDigest(ctx, c.cacher, "0", 0, func(obj metav1.Object) {
		// collect items from cacher's list
		cacheItems = append(cacheItems, namespaceNameRV{Namespace: obj.GetNamespace(), Name: obj.GetName(), RV: obj.GetResourceVersion()})
	})
	if err != nil {
		return nil, fmt.Errorf("failed calculating cache digest: %w", err)
	}
	etcdDigest, etcdResourceVersion, err := c.calculateStoreDigest(ctx, c.etcd, cacheResourceVersion, storageWatchListPageSize, func(obj metav1.Object) {
		// compare the item in etcd's list against cacheItems which is cacher's list
		if foundDiff != nil {
			return
		}
		etcdItem := namespaceNameRV{Namespace: obj.GetNamespace(), Name: obj.GetName(), RV: obj.GetResourceVersion()}
		if len(cacheItems) <= etcdIndex {
			foundDiff = &diffDetail{Index: etcdIndex, EtcdItem: &etcdItem}
			cacheItems = nil // don't need it any more and nil it to allow GC to collect it
			return
		}
		if cacheItem := cacheItems[etcdIndex]; cacheItem != etcdItem {
			foundDiff = &diffDetail{Index: etcdIndex, EtcdItem: &etcdItem, CacheItem: &cacheItem}
			cacheItems = nil // don't need it any more and nil it to allow GC to collect it
			return
		}
		etcdIndex += 1
	})
	if err != nil {
		return nil, fmt.Errorf("failed calculating etcd digest: %w", err)
	}
	if len(cacheItems) > etcdIndex {
		cacheItem := cacheItems[etcdIndex]
		foundDiff = &diffDetail{Index: etcdIndex, CacheItem: &cacheItem}
	}
	if cacheResourceVersion != etcdResourceVersion {