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

Code before:
func (c consistencyChecker) startChecking(stopCh <-chan struct{}) {
	klog.V(3).InfoS("Cache consistency check start", "group", c.groupResource.Group, "resource", c.groupResource.Resource)
	jitter := 0.5 // Period between [interval, interval * (1.0 + jitter)]
	sliding := true
	// wait.JitterUntilWithContext starts work immediately, so wait first.
	select {
	case <-time.After(wait.Jitter(ConsistencyCheckPeriod, jitter)):
	case <-stopCh:
	}
	wait.JitterUntilWithContext(wait.ContextForChannel(stopCh), c.check, ConsistencyCheckPeriod, jitter, sliding)
}

func (c *consistencyChecker) check(ctx context.Context) {
	digests, err := c.calculateDigests(ctx)
	if err != nil {
		klog.ErrorS(err, "Cache consistency check error", "group", c.groupResource.Group, "resource", c.groupResource.Resource)
		metrics.StorageConsistencyCheckTotal.WithLabelValues(c.groupResource.Group, c.groupResource.Resource, "error").Inc()
		return
	}
	if digests.CacheDigest == digests.EtcdDigest {
		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
	}
	klog.ErrorS(nil, "Cache consistency check failed", "group", c.groupResource.Group, "resource", c.groupResource.Resource, "resourceVersion", digests.ResourceVersion, "etcdDigest", digests.EtcdDigest, "cacheDigest", digests.CacheDigest, "1stDiffIndex", digests.diffDetail.index, "cacheItem", digests.diffDetail.cacheItem, "etcdItem", digests.diffDetail.etcdItem)
	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, 1stDiffIndex: %v, cacheItem: %v, etcdItem: %v", c.groupResource.Group, c.groupResource.Resource, digests.ResourceVersion, digests.EtcdDigest, digests.CacheDigest, digests.diffDetail.index, digests.diffDetail.cacheItem, digests.diffDetail.etcdItem))
	}
	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
	trackingList := []namespaceNameRV{}
	foundDiff := false
	dd := &diffDetail{index: -1}

	cacheDigest, cacheResourceVersion, err := c.calculateStoreDigest(ctx, c.cacher, "0", 0, func(obj metav1.Object) {
		// collect items from cacher's list
		trackingList = append(trackingList, 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 trackingList which is cacher's list

Review comment (bug):
ideally, we could just log diffDetail directly here and in the panic, like this:

```suggestion
	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)
```

that might require making the DiffDetail fields or the fields inside the DiffDetail type exported, and maybe adding json tags to control how they appear in structured log output

Code after:
func (c consistencyChecker) startChecking(stopCh <-chan struct{}) {
	klog.V(3).InfoS("Cache consistency check start", "group", c.groupResource.Group, "resource", c.groupResource.Resource)
	jitter := 0.5 // Period between [interval, interval * (1.0 + jitter)]
	sliding := true
	// wait.JitterUntilWithContext starts work immediately, so wait first.
	select {
	case <-time.After(wait.Jitter(ConsistencyCheckPeriod, jitter)):
	case <-stopCh:
	}
	wait.JitterUntilWithContext(wait.ContextForChannel(stopCh), c.check, ConsistencyCheckPeriod, jitter, sliding)
}

func (c *consistencyChecker) check(ctx context.Context) {
	digests, err := c.calculateDigests(ctx)
	if err != nil {
		klog.ErrorS(err, "Cache consistency check error", "group", c.groupResource.Group, "resource", c.groupResource.Resource)
		metrics.StorageConsistencyCheckTotal.WithLabelValues(c.groupResource.Group, c.groupResource.Resource, "error").Inc()
		return
	}
	if digests.CacheDigest == digests.EtcdDigest {
		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
	}
	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