DATAREDIS-1082 - Capture per cache statistics.

Original Pull Request: #545

src/main/java/org/springframework/data/redis/cache/CacheStatistics.java

@@ -15,28 +15,37 @@
  */
 package org.springframework.data.redis.cache;
 
+import java.time.Instant;
 import java.util.concurrent.TimeUnit;
 
 /**
- * Cache statistics for a Redis Cache.
+ * Cache statistics for a {@link RedisCache}. <br />
+ * <strong>NOTE:</strong> {@link CacheStatistics} only serve local (in memory) data and do not collect any server
+ * statistics.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.4
  */
 public interface CacheStatistics {
 
+	/**
+	 * @return the name of the {@link RedisCache}.
+	 */
+	String getCacheName();
+
 	/**
 	 * @return number of put operations on the cache.
 	 */
-	long getStores();
+	long getPuts();
 
 	/**
-	 * @return number of get operations.
+	 * @return the total number of get operations including both {@link #getHits() hits} and {@link #getMisses() misses}.
 	 */
-	long getRetrievals();
+	long getGets();
 
 	/**
-	 * @return number of cache get hits.
+	 * @return the number of cache get hits.
 	 */
 	long getHits();
 
@@ -45,14 +54,39 @@ public interface CacheStatistics {
 	 */
 	long getMisses();
 
+	/**
+	 * @return the number of {@link #getGets() gets} that have not yet been answered (neither {@link #getHits() hit} nor
+	 *         {@link #getMisses() miss}).
+	 */
+	default long getPending() {
+		return getGets() - (getHits() + getMisses());
+	}
+
 	/**
 	 * @return number of cache removals.
 	 */
-	long getRemovals();
+	long getDeletes();
 
 	/**
 	 * @param unit the time unit to report the lock wait duration.
 	 * @return lock duration using the given {@link TimeUnit} if the cache is configured to use locking.
 	 */
 	long getLockWaitDuration(TimeUnit unit);
+
+	/**
+	 * @return initial point in time when started statistics capturing.
+	 */
+	Instant getSince();
+
+	/**
+	 * @return instantaneous point in time of last statistics counter reset. Equals {@link #getSince()} if never resetted.
+	 */
+	Instant getLastReset();
+
+	/**
+	 * @return the statistics time.
+	 */
+	default Instant getTime() {
+		return Instant.now();
+	}
 }

src/main/java/org/springframework/data/redis/cache/CacheStatisticsCollector.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.cache;
+
+/**
+ * The statistics collector supports capturing of relevant {@link RedisCache} operations such as
+ * {@literal hits & misses}.
+ *
+ * @author Christoph Strobl
+ * @since 2.4
+ */
+public interface CacheStatisticsCollector extends CacheStatisticsProvider {
+
+	/**
+	 * Increase the counter for {@literal put operations} of the given cache.
+	 * 
+	 * @param cacheName must not be {@literal null}.
+	 */
+	void incPuts(String cacheName);
+
+	/**
+	 * Increase the counter for {@literal get operations} of the given cache.
+	 * 
+	 * @param cacheName must not be {@literal null}.
+	 */
+	void incGets(String cacheName);
+
+	/**
+	 * Increase the counter for {@literal get operations with result} of the given cache.
+	 * 
+	 * @param cacheName must not be {@literal null}.
+	 */
+	void incHits(String cacheName);
+
+	/**
+	 * Increase the counter for {@literal get operations without result} of the given cache.
+	 * 
+	 * @param cacheName must not be {@literal null}.
+	 */
+	void incMisses(String cacheName);
+
+	/**
+	 * Increase the counter for {@literal delete operations} of the given cache.
+	 * 
+	 * @param cacheName must not be {@literal null}.
+	 */
+	default void incDeletes(String cacheName) {
+		incDeletesBy(cacheName, 1);
+	}
+
+	/**
+	 * Increase the counter for {@literal delete operations} of the given cache by the given value.
+	 * 
+	 * @param cacheName must not be {@literal null}.
+	 */
+	void incDeletesBy(String cacheName, int value);
+
+	/**
+	 * Increase the gauge for {@literal sync lock duration} of the cache by the given nanoseconds.
+	 * 
+	 * @param cacheName must not be {@literal null}.
+	 */
+	void incLockTime(String cacheName, long durationNS);
+
+	/**
+	 * Reset the all counters and gauges of for the given cache.
+	 * 
+	 * @param cacheName must not be {@literal null}.
+	 */
+	void reset(String cacheName);
+
+	/**
+	 * @return a {@link CacheStatisticsCollector} that performs no action.
+	 */
+	static CacheStatisticsCollector none() {
+		return NoOpCacheStatisticsCollector.INSTANCE;
+	}
+
+	/**
+	 * @return a default {@link CacheStatisticsCollector} implementation.
+	 */
+	static CacheStatisticsCollector instance/*clearlyNeedsBetterNaming*/() {
+		return new DefaultCacheStatisticsCollector();
+	}
+}

src/main/java/org/springframework/data/redis/cache/CacheStatisticsProvider.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.cache;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.4
+ */
+public interface CacheStatisticsProvider {
+
+	/**
+	 * Obtain snapshot of the captured statistics.
+	 *
+	 * @param cacheName must not be {@literal null}.
+	 * @return never {@literal null}.
+	 */
+	CacheStatistics getCacheStatistics(String cacheName);
+}

src/main/java/org/springframework/data/redis/cache/DefaultCacheStatistics.java

@@ -1,125 +0,0 @@
-/*
- * Copyright 2020 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.springframework.data.redis.cache;
-
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.LongAdder;
-
-import org.springframework.util.Assert;
-
-/**
- * Default mutable {@link CacheStatistics} implementation.
- *
- * @author Mark Paluch
- * @since 2.4
- */
-class DefaultCacheStatistics implements CacheStatistics {
-
-	private final LongAdder stores = new LongAdder();
-	private final LongAdder retrievals = new LongAdder();
-	private final LongAdder hits = new LongAdder();
-	private final LongAdder misses = new LongAdder();
-	private final LongAdder removals = new LongAdder();
-	private final LongAdder lockWaitTimeNs = new LongAdder();
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.cache.CacheStatistics#getStores()
-	 */
-	@Override
-	public long getStores() {
-		return stores.sum();
-	}
-
-	void incrementStores() {
-		stores.increment();
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.cache.CacheStatistics#getRetrievals()
-	 */
-	@Override
-	public long getRetrievals() {
-		return retrievals.sum();
-	}
-
-	void incrementRetrievals() {
-		retrievals.increment();
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.cache.CacheStatistics#getHits()
-	 */
-	@Override
-	public long getHits() {
-		return hits.sum();
-	}
-
-	void incrementHits() {
-		hits.increment();
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.cache.CacheStatistics#getMisses()
-	 */
-	@Override
-	public long getMisses() {
-		return misses.sum();
-	}
-
-	void incrementMisses() {
-		misses.increment();
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.cache.CacheStatistics#getRemovals()
-	 */
-	@Override
-	public long getRemovals() {
-		return removals.sum();
-	}
-
-	void incrementRemovals() {
-		incrementRemovals(1);
-	}
-
-	/**
-	 * @param x number of removals to add.
-	 */
-	void incrementRemovals(int x) {
-		removals.add(x);
-	}
-
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.cache.CacheStatistics#getLockWaitDuration(java.util.concurrent.TimeUnit)
-	 */
-	@Override
-	public long getLockWaitDuration(TimeUnit unit) {
-
-		Assert.notNull(unit, "TimeUnit must not be null");
-
-		return unit.convert(lockWaitTimeNs.sum(), TimeUnit.NANOSECONDS);
-	}
-
-	void incrementLockWaitTime(long waitTimeNs) {
-		lockWaitTimeNs.add(waitTimeNs);
-	}
-}

src/main/java/org/springframework/data/redis/cache/DefaultCacheStatisticsCollector.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.cache;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Default {@link CacheStatisticsCollector} implementation holding synchronized per cache
+ * {@link MutableCacheStatistics}.
+ * 
+ * @author Christoph Strobl
+ * @since 2.4
+ */
+class DefaultCacheStatisticsCollector implements CacheStatisticsCollector {
+
+	private final Map<String, MutableCacheStatistics> stats = new ConcurrentHashMap<>();
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incPuts(java.lang.String)
+	 */
+	@Override
+	public void incPuts(String cacheName) {
+		statsFor(cacheName).incPuts();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incGets(java.lang.String)
+	 */
+	@Override
+	public void incGets(String cacheName) {
+		statsFor(cacheName).incGets();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incHits(java.lang.String)
+	 */
+	@Override
+	public void incHits(String cacheName) {
+		statsFor(cacheName).incHits();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incMisses(java.lang.String)
+	 */
+	@Override
+	public void incMisses(String cacheName) {
+		statsFor(cacheName).incMisses();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incDeletesBy(java.lang.String, int)
+	 */
+	@Override
+	public void incDeletesBy(String cacheName, int value) {
+		statsFor(cacheName).incDeletes(value);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incLockTime(java.lang.String)
+	 */
+	@Override
+	public void incLockTime(String name, long durationNS) {
+		statsFor(name).incLockWaitTime(durationNS);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.reset(java.lang.String)
+	 */
+	@Override
+	public void reset(String cacheName) {
+		statsFor(cacheName).reset();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.getCacheStatistics(java.lang.String)
+	 */
+	@Override
+	public CacheStatistics getCacheStatistics(String cacheName) {
+		return statsFor(cacheName).captureSnapshot();
+	}
+
+	private MutableCacheStatistics statsFor(String cacheName) {
+		return stats.computeIfAbsent(cacheName, MutableCacheStatistics::new);
+	}
+}

src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java

@@ -51,7 +51,7 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 
 	private final RedisConnectionFactory connectionFactory;
 	private final Duration sleepTime;
-	private final DefaultCacheStatistics statistics;
+	private final CacheStatisticsCollector statistics;
 
 	/**
 	 * @param connectionFactory must not be {@literal null}.
@@ -66,13 +66,25 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 	 *          to disable locking.
 	 */
 	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime) {
+		this(connectionFactory, sleepTime, CacheStatisticsCollector.none());
+	}
+
+	/**
+	 * @param connectionFactory must not be {@literal null}.
+	 * @param sleepTime sleep time between lock request attempts. Must not be {@literal null}. Use {@link Duration#ZERO}
+	 *          to disable locking.
+	 * @param cacheStatisticsCollector must not be {@literal null}.
+	 */
+	DefaultRedisCacheWriter(RedisConnectionFactory connectionFactory, Duration sleepTime,
+			CacheStatisticsCollector cacheStatisticsCollector) {
 
 		Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
 		Assert.notNull(sleepTime, "SleepTime must not be null!");
+		Assert.notNull(cacheStatisticsCollector, "CacheStatisticsCollector must not be null!");
 
 		this.connectionFactory = connectionFactory;
 		this.sleepTime = sleepTime;
-		this.statistics = new DefaultCacheStatistics();
+		this.statistics = cacheStatisticsCollector;
 	}
 
 	/*
@@ -97,7 +109,7 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 			return "OK";
 		});
 
-		statistics.incrementStores();
+		statistics.incPuts(name);
 	}
 
 	/*
@@ -112,12 +124,12 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 
 		byte[] result = execute(name, connection -> connection.get(key));
 
-		statistics.incrementRetrievals();
+		statistics.incGets(name);
 
 		if (result != null) {
-			statistics.incrementHits();
+			statistics.incHits(name);
 		} else {
-			statistics.incrementMisses();
+			statistics.incMisses(name);
 		}
 
 		return result;
@@ -147,7 +159,7 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 						connection.pExpire(key, ttl.toMillis());
 					}
 
-					statistics.incrementStores();
+					statistics.incPuts(name);
 					return null;
 				}
 
@@ -172,7 +184,7 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 		Assert.notNull(key, "Key must not be null!");
 
 		execute(name, connection -> connection.del(key));
-		statistics.incrementRemovals();
+		statistics.incDeletes(name);
 	}
 
 	/*
@@ -200,7 +212,7 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 						.toArray(new byte[0][]);
 
 				if (keys.length > 0) {
-					statistics.incrementRemovals(keys.length);
+					statistics.incDeletesBy(name, keys.length);
 					connection.del(keys);
 				}
 			} finally {
@@ -216,11 +228,25 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.cache.RedisCacheWriter#getStatistics()
+	 * @see org.springframework.data.redis.cache.CacheStatisticsProvider#getCacheStatistics(java.lang.String)
 	 */
 	@Override
-	public DefaultCacheStatistics getStatistics() {
-		return statistics;
+	public CacheStatistics getCacheStatistics(String cacheName) {
+		return statistics.getCacheStatistics(cacheName);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.RedisCacheWriter#clearStatistics(java.lang.String)
+	 */
+	@Override
+	public void clearStatistics(String name) {
+		statistics.reset(name);
+	}
+
+	@Override
+	public RedisCacheWriter with(CacheStatisticsCollector cacheStatisticsCollector) {
+		return new DefaultRedisCacheWriter(connectionFactory, sleepTime, cacheStatisticsCollector);
 	}
 
 	/**
@@ -303,7 +329,7 @@ class DefaultRedisCacheWriter implements RedisCacheWriter {
 			throw new PessimisticLockingFailureException(String.format("Interrupted while waiting to unlock cache %s", name),
 					ex);
 		} finally {
-			statistics.incrementLockWaitTime(System.nanoTime() - lockWaitTimeNs);
+			statistics.incLockTime(name, System.nanoTime() - lockWaitTimeNs);
 		}
 	}
 

src/main/java/org/springframework/data/redis/cache/MutableCacheStatistics.java

@@ -0,0 +1,304 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.cache;
+
+import java.time.Instant;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.springframework.util.Assert;
+
+/**
+ * Default mutable {@link CacheStatistics} implementation.
+ *
+ * @author Mark Paluch
+ * @since 2.4
+ */
+class MutableCacheStatistics implements CacheStatistics {
+
+	private final String cacheName;
+
+	private final Instant aliveSince = Instant.now();
+	private Instant lastReset = aliveSince;
+
+	private final LongAdder puts = new LongAdder();
+	private final LongAdder gets = new LongAdder();
+	private final LongAdder hits = new LongAdder();
+	private final LongAdder misses = new LongAdder();
+	private final LongAdder deletes = new LongAdder();
+	private final LongAdder lockWaitTimeNs = new LongAdder();
+
+	MutableCacheStatistics(String cacheName) {
+		this.cacheName = cacheName;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * org.springframework.data.redis.cache.CacheStatistics#getCacheName()
+	 */
+	@Override
+	public String getCacheName() {
+		return cacheName;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatistics#getPuts()
+	 */
+	@Override
+	public long getPuts() {
+		return puts.sum();
+	}
+
+	void incPuts() {
+		puts.increment();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatistics#getRetrievals()
+	 */
+	@Override
+	public long getGets() {
+		return gets.sum();
+	}
+
+	void incGets() {
+		gets.increment();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatistics#getHits()
+	 */
+	@Override
+	public long getHits() {
+		return hits.sum();
+	}
+
+	void incHits() {
+		hits.increment();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatistics#getMisses()
+	 */
+	@Override
+	public long getMisses() {
+		return misses.sum();
+	}
+
+	void incMisses() {
+		misses.increment();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatistics#getRemovals()
+	 */
+	@Override
+	public long getDeletes() {
+		return deletes.sum();
+	}
+
+	void incDeletes() {
+		incDeletes(1);
+	}
+
+	/**
+	 * @param x number of removals to add.
+	 */
+	void incDeletes(int x) {
+		deletes.add(x);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatistics#getLockWaitDuration(java.util.concurrent.TimeUnit)
+	 */
+	@Override
+	public long getLockWaitDuration(TimeUnit unit) {
+
+		Assert.notNull(unit, "TimeUnit must not be null");
+
+		return unit.convert(lockWaitTimeNs.sum(), TimeUnit.NANOSECONDS);
+	}
+
+	@Override
+	public Instant getSince() {
+		return this.aliveSince;
+	}
+
+	@Override
+	public Instant getLastReset() {
+		return lastReset;
+	}
+
+	void incLockWaitTime(long waitTimeNs) {
+		lockWaitTimeNs.add(waitTimeNs);
+	}
+
+	void reset() {
+
+		lastReset = Instant.now();
+
+		puts.reset();
+		gets.reset();
+		hits.reset();
+		misses.reset();
+		deletes.reset();
+		lockWaitTimeNs.reset();
+	}
+
+	CacheStatistics captureSnapshot() {
+		return new Snapshot(this);
+	}
+
+	/**
+	 * {@link CacheStatistics} value object holding snapshot data.
+	 */
+	private static class Snapshot implements CacheStatistics {
+
+		private final String cacheName;
+		private final long puts;
+		private final long gets;
+		private final long hits;
+		private final long misses;
+		private final long deletes;
+		private final long lockWaitTimeNS;
+		private final long pending;
+		private final Instant time;
+		private final Instant since;
+		private final Instant lastReset;
+
+		Snapshot(CacheStatistics statistics) {
+
+			cacheName = statistics.getCacheName();
+			gets = statistics.getGets();
+			hits = statistics.getHits();
+			misses = statistics.getMisses();
+			puts = statistics.getPuts();
+			deletes = statistics.getDeletes();
+			pending = gets - (hits + misses);
+
+			lockWaitTimeNS = statistics.getLockWaitDuration(TimeUnit.NANOSECONDS);
+
+			time = Instant.now();
+			since = Instant.from(statistics.getSince());
+			lastReset = Instant.from(statistics.getLastReset());
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getCacheName()
+		 */
+		@Override
+		public String getCacheName() {
+			return cacheName;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getPuts()
+		 */
+		@Override
+		public long getPuts() {
+			return puts;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getGets()
+		 */
+		@Override
+		public long getGets() {
+			return gets;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getHits()
+		 */
+		@Override
+		public long getHits() {
+			return hits;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getMisses()
+		 */
+		@Override
+		public long getMisses() {
+			return misses;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getPending()
+		 */
+		@Override
+		public long getPending() {
+			return pending;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getDeletes()
+		 */
+		@Override
+		public long getDeletes() {
+			return deletes;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getCacheName(java.util.concurrent.TimeUnit)
+		 */
+		@Override
+		public long getLockWaitDuration(TimeUnit unit) {
+			return unit.convert(lockWaitTimeNS, TimeUnit.NANOSECONDS);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getSince()
+		 */
+		@Override
+		public Instant getSince() {
+			return since;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getLastReset()
+		 */
+		@Override
+		public Instant getLastReset() {
+			return lastReset;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * org.springframework.data.redis.cache.CacheStatistics#getTime()
+		 */
+		@Override
+		public Instant getTime() {
+			return time;
+		}
+	}
+}

src/main/java/org/springframework/data/redis/cache/NoOpCacheStatisticsCollector.java

@@ -0,0 +1,164 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.cache;
+
+import java.time.Instant;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.util.ObjectUtils;
+
+/**
+ * {@link CacheStatisticsCollector} implementation that does not capture anything but throws an
+ * {@link IllegalStateException} when {@link #getCacheStatistics(String) obtaining} {@link CacheStatistics} for a cache.
+ * 
+ * @author Christoph Strobl
+ * @since 2.4
+ */
+enum NoOpCacheStatisticsCollector implements CacheStatisticsCollector {
+
+	INSTANCE;
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incPuts(java.lang.String)
+	 */
+	@Override
+	public void incPuts(String cacheName) {}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incGets(java.lang.String)
+	 */
+	@Override
+	public void incGets(String cacheName) {}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incHits(java.lang.String)
+	 */
+	@Override
+	public void incHits(String cacheName) {}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incMisses(java.lang.String)
+	 */
+	@Override
+	public void incMisses(String cacheName) {}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incDeletesBy(java.lang.String)
+	 */
+	@Override
+	public void incDeletesBy(String cacheName, int value) {}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.incLockTime(java.lang.String)
+	 */
+	@Override
+	public void incLockTime(String name, long durationNS) {}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.reset(java.lang.String)
+	 */
+	@Override
+	public void reset(String cacheName) {}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.cache.CacheStatisticsCollector.getCacheStatistics(java.lang.String)
+	 */
+	@Override
+	public CacheStatistics getCacheStatistics(String cacheName) {
+		return new EmptyStats(cacheName);
+	}
+
+	private static class EmptyStats implements CacheStatistics {
+
+		private final String cacheName;
+
+		EmptyStats(String cacheName) {
+			this.cacheName = cacheName;
+		}
+
+		@Override
+		public String getCacheName() {
+			return cacheName;
+		}
+
+		@Override
+		public long getPuts() {
+			return -1;
+		}
+
+		@Override
+		public long getGets() {
+			return -1;
+		}
+
+		@Override
+		public long getHits() {
+			return -1;
+		}
+
+		@Override
+		public long getMisses() {
+			return -1;
+		}
+
+		@Override
+		public long getDeletes() {
+			return -1;
+		}
+
+		@Override
+		public long getLockWaitDuration(TimeUnit unit) {
+			return -1;
+		}
+
+		@Override
+		public Instant getSince() {
+			return Instant.EPOCH;
+		}
+
+		@Override
+		public Instant getLastReset() {
+			return getSince();
+		}
+
+		@Override
+		public boolean equals(Object o) {
+
+			if (this == o) {
+				return true;
+			}
+			if (o == null || getClass() != o.getClass()) {
+				return false;
+			}
+
+			EmptyStats that = (EmptyStats) o;
+			return ObjectUtils.nullSafeEquals(cacheName, that.cacheName);
+		}
+
+		@Override
+		public int hashCode() {
+			return ObjectUtils.nullSafeHashCode(cacheName);
+		}
+	}
+}

src/main/java/org/springframework/data/redis/cache/RedisCache.java

@@ -195,13 +195,23 @@ public class RedisCache extends AbstractValueAdaptingCache {
 
 	/**
 	 * Return the {@link CacheStatistics} for this cache instance. Statistics are accumulated per cache instance and not
-	 * from the backing Redis data store. Cache statistics are accumulated starting from the time a cache is created.
+	 * from the backing Redis data store.
 	 *
 	 * @return statistics object for this {@link RedisCache}.
 	 * @since 2.4
+	 * @throws IllegalStateException if the {@link RedisCacheWriter} does not support statistics.
 	 */
 	public CacheStatistics getStatistics() {
-		return cacheWriter.getStatistics();
+		return cacheWriter.getCacheStatistics(getName());
+	}
+
+	/**
+	 * Reset all statistics counters and gauges for this cache.
+	 *
+	 * @since 2.4
+	 */
+	public void clearStatistics() {
+		cacheWriter.clearStatistics(getName());
 	}
 
 	/**

src/main/java/org/springframework/data/redis/cache/RedisCacheManager.java

@@ -289,6 +289,7 @@ public class RedisCacheManager extends AbstractTransactionSupportingCacheManager
 	public static class RedisCacheManagerBuilder {
 
 		private @Nullable RedisCacheWriter cacheWriter;
+		private CacheStatisticsCollector statisticsCollector = CacheStatisticsCollector.none();
 		private RedisCacheConfiguration defaultCacheConfiguration = RedisCacheConfiguration.defaultCacheConfig();
 		private final Map<String, RedisCacheConfiguration> initialCaches = new LinkedHashMap<>();
 		private boolean enableTransactions;
@@ -456,6 +457,25 @@ public class RedisCacheManager extends AbstractTransactionSupportingCacheManager
 			return Optional.ofNullable(this.initialCaches.get(cacheName));
 		}
 
+		/**
+		 * @return
+		 * @since 2.4
+		 */
+		public RedisCacheManagerBuilder enableStatistics() {
+			return statisticsCollector(CacheStatisticsCollector.instance());
+		}
+
+		/**
+		 * @return
+		 * @since 2.4
+		 */
+		private RedisCacheManagerBuilder statisticsCollector(CacheStatisticsCollector statisticsCollector) {
+
+			Assert.notNull(statisticsCollector, "StatisticsCollector must not be null!");
+			this.statisticsCollector = statisticsCollector;
+			return this;
+		}
+
 		/**
 		 * Create new instance of {@link RedisCacheManager} with configuration options applied.
 		 *
@@ -463,9 +483,16 @@ public class RedisCacheManager extends AbstractTransactionSupportingCacheManager
 		 */
 		public RedisCacheManager build() {
 
-			Assert.state(cacheWriter != null, "CacheWriter must not be null! You can provide one via 'RedisCacheManagerBuilder#cacheWriter(RedisCacheWriter)'.");
+			Assert.state(cacheWriter != null,
+					"CacheWriter must not be null! You can provide one via 'RedisCacheManagerBuilder#cacheWriter(RedisCacheWriter)'.");
 
-			RedisCacheManager cm = new RedisCacheManager(cacheWriter, defaultCacheConfiguration, initialCaches,
+			RedisCacheWriter theCacheWriter = cacheWriter;
+
+			if (!statisticsCollector.equals(CacheStatisticsCollector.none())) {
+				theCacheWriter = cacheWriter.with(statisticsCollector);
+			}
+
+			RedisCacheManager cm = new RedisCacheManager(theCacheWriter, defaultCacheConfiguration, initialCaches,
 					allowInFlightCacheCreation);
 
 			cm.setTransactionAware(enableTransactions);

src/main/java/org/springframework/data/redis/cache/RedisCacheWriter.java

@@ -31,7 +31,7 @@ import org.springframework.util.Assert;
  * @author Mark Paluch
  * @since 2.0
  */
-public interface RedisCacheWriter {
+public interface RedisCacheWriter extends CacheStatisticsProvider {
 
 	/**
 	 * Create new {@link RedisCacheWriter} without locking behavior.
@@ -108,11 +108,18 @@ public interface RedisCacheWriter {
 	void clean(String name, byte[] pattern);
 
 	/**
-	 * Return the {@link CacheStatistics} for this cache instance. Statistics are accumulated per cache instance and not
-	 * from the backing Redis data store. Cache statistics are accumulated starting from the time a cache is created.
+	 * Reset all statistics counters and gauges for this cache.
 	 *
-	 * @return statistics object for this {@link RedisCache}.
 	 * @since 2.4
 	 */
-	CacheStatistics getStatistics();
+	void clearStatistics(String name);
+
+	/**
+	 * Obtain a {@link RedisCacheWriter} using the given {@link CacheStatisticsCollector} to collect metrics.
+	 *
+	 * @param cacheStatisticsCollector must not be {@literal null}.
+	 * @return new instance of {@link RedisCacheWriter}.
+	 */
+	RedisCacheWriter with(CacheStatisticsCollector cacheStatisticsCollector);
+
 }