diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageStats.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageStats.java
index 6546dfde0f8..9abc65cc973 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageStats.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageStats.java
@@ -85,16 +85,16 @@ class DbLedgerStorageStats {
     private final OpStatsLogger readEntryStats;
     @StatsDoc(
             name = READ_ENTRY_LOCATIONS_INDEX_TIME,
-            help = "time spent reading entries from the locations index of the db ledger storage engine",
+            help = "operation stats of reading entries from the locations index of the db ledger storage engine",
             parent = READ_ENTRY
     )
-    private final Counter readFromLocationIndexTime;
+    private final OpStatsLogger readFromLocationIndexTime;
     @StatsDoc(
             name = READ_ENTRYLOG_TIME,
-            help = "time spent reading entries from the entry log files of the db ledger storage engine",
+            help = "operation stats of reading entries from the entry log files of the db ledger storage engine",
             parent = READ_ENTRY
     )
-    private final Counter readFromEntryLogTime;
+    private final OpStatsLogger readFromEntryLogTime;
     @StatsDoc(
             name = WRITE_CACHE_HITS,
             help = "number of write cache hits (on reads)",
@@ -131,9 +131,9 @@ class DbLedgerStorageStats {
     private final OpStatsLogger readAheadBatchSizeStats;
     @StatsDoc(
             name = READAHEAD_TIME,
-            help = "Time spent on readahead operations"
+            help = "operation stats of readahead operations"
     )
-    private final Counter readAheadTime;
+    private final OpStatsLogger readAheadTime;
     @StatsDoc(
         name = FLUSH,
         help = "operation stats of flushing write cache to entry log files"
@@ -203,15 +203,15 @@ class DbLedgerStorageStats {
                          Supplier<Long> readCacheCountSupplier) {
         addEntryStats = stats.getThreadScopedOpStatsLogger(ADD_ENTRY);
         readEntryStats = stats.getThreadScopedOpStatsLogger(READ_ENTRY);
-        readFromLocationIndexTime = stats.getThreadScopedCounter(READ_ENTRY_LOCATIONS_INDEX_TIME);
-        readFromEntryLogTime = stats.getThreadScopedCounter(READ_ENTRYLOG_TIME);
+        readFromLocationIndexTime = stats.getThreadScopedOpStatsLogger(READ_ENTRY_LOCATIONS_INDEX_TIME);
+        readFromEntryLogTime = stats.getThreadScopedOpStatsLogger(READ_ENTRYLOG_TIME);
         readCacheHitCounter = stats.getCounter(READ_CACHE_HITS);
         readCacheMissCounter = stats.getCounter(READ_CACHE_MISSES);
         writeCacheHitCounter = stats.getCounter(WRITE_CACHE_HITS);
         writeCacheMissCounter = stats.getCounter(WRITE_CACHE_MISSES);
         readAheadBatchCountStats = stats.getOpStatsLogger(READAHEAD_BATCH_COUNT);
         readAheadBatchSizeStats = stats.getOpStatsLogger(READAHEAD_BATCH_SIZE);
-        readAheadTime = stats.getThreadScopedCounter(READAHEAD_TIME);
+        readAheadTime = stats.getThreadScopedOpStatsLogger(READAHEAD_TIME);
         flushStats = stats.getOpStatsLogger(FLUSH);
         flushEntryLogStats = stats.getOpStatsLogger(FLUSH_ENTRYLOG);
         flushLocationIndexStats = stats.getOpStatsLogger(FLUSH_LOCATIONS_INDEX);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
index 6ce2d4b4f54..d9936371093 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
@@ -638,17 +638,19 @@ private ByteBuf doGetEntry(long ledgerId, long entryId) throws IOException, Book
 
                 throw new NoEntryException(ledgerId, entryId);
             }
-        } finally {
-            dbLedgerStorageStats.getReadFromLocationIndexTime().addLatency(
-                    MathUtils.elapsedNanos(locationIndexStartNano), TimeUnit.NANOSECONDS);
+            recordSuccessfulEvent(dbLedgerStorageStats.getReadFromLocationIndexTime(), locationIndexStartNano);
+        } catch (NoEntryException e) {
+            recordFailedEvent(dbLedgerStorageStats.getReadFromLocationIndexTime(), locationIndexStartNano);
+            throw e;
         }
 
         long readEntryStartNano = MathUtils.nowInNano();
         try {
             entry = entryLogger.readEntry(ledgerId, entryId, entryLocation);
-        } finally {
-            dbLedgerStorageStats.getReadFromEntryLogTime().addLatency(
-                    MathUtils.elapsedNanos(readEntryStartNano), TimeUnit.NANOSECONDS);
+            recordSuccessfulEvent(dbLedgerStorageStats.getReadFromEntryLogTime(), readEntryStartNano);
+        } catch (Exception e) {
+            recordFailedEvent(dbLedgerStorageStats.getReadFromEntryLogTime(), readEntryStartNano);
+            throw e;
         }
 
         readCache.put(ledgerId, entryId, entry);
@@ -696,15 +698,16 @@ private void fillReadAheadCache(long originalLedgerId, long firstEntryId, long f
                     ReferenceCountUtil.release(entry);
                 }
             }
+            dbLedgerStorageStats.getReadAheadBatchCountStats().registerSuccessfulValue(count);
+            dbLedgerStorageStats.getReadAheadBatchSizeStats().registerSuccessfulValue(size);
+            recordSuccessfulEvent(dbLedgerStorageStats.getReadAheadTime(), readAheadStartNano);
         } catch (Exception e) {
             if (log.isDebugEnabled()) {
                 log.debug("Exception during read ahead for ledger: {}: e", originalLedgerId, e);
             }
-        } finally {
-            dbLedgerStorageStats.getReadAheadBatchCountStats().registerSuccessfulValue(count);
-            dbLedgerStorageStats.getReadAheadBatchSizeStats().registerSuccessfulValue(size);
-            dbLedgerStorageStats.getReadAheadTime().addLatency(
-                    MathUtils.elapsedNanos(readAheadStartNano), TimeUnit.NANOSECONDS);
+            dbLedgerStorageStats.getReadAheadBatchCountStats().registerFailedValue(count);
+            dbLedgerStorageStats.getReadAheadBatchSizeStats().registerFailedValue(size);
+            recordFailedEvent(dbLedgerStorageStats.getReadAheadTime(), readAheadStartNano);
         }
     }
 
@@ -763,21 +766,31 @@ public ByteBuf getLastEntry(long ledgerId) throws IOException, BookieException {
         dbLedgerStorageStats.getWriteCacheMissCounter().inc();
 
         // Search the last entry in storage
+        long lastEntryId;
+        long entryLocation;
         long locationIndexStartNano = MathUtils.nowInNano();
-        long lastEntryId = entryLocationIndex.getLastEntryInLedger(ledgerId);
-        if (log.isDebugEnabled()) {
-            log.debug("Found last entry for ledger {} in db: {}", ledgerId, lastEntryId);
-        }
+        try {
+            lastEntryId = entryLocationIndex.getLastEntryInLedger(ledgerId);
+            if (log.isDebugEnabled()) {
+                log.debug("Found last entry for ledger {} in db: {}", ledgerId, lastEntryId);
+            }
 
-        long entryLocation = entryLocationIndex.getLocation(ledgerId, lastEntryId);
-        dbLedgerStorageStats.getReadFromLocationIndexTime().addLatency(
-                MathUtils.elapsedNanos(locationIndexStartNano), TimeUnit.NANOSECONDS);
+            entryLocation = entryLocationIndex.getLocation(ledgerId, lastEntryId);
+            recordSuccessfulEvent(dbLedgerStorageStats.getReadFromLocationIndexTime(), locationIndexStartNano);
+        } catch (NoEntryException e) {
+            recordFailedEvent(dbLedgerStorageStats.getReadFromLocationIndexTime(), locationIndexStartNano);
+            throw e;
+        }
 
         long readEntryStartNano = MathUtils.nowInNano();
-        ByteBuf content = entryLogger.readEntry(ledgerId, lastEntryId, entryLocation);
-        dbLedgerStorageStats.getReadFromEntryLogTime().addLatency(
-                MathUtils.elapsedNanos(readEntryStartNano), TimeUnit.NANOSECONDS);
-        return content;
+        try {
+            ByteBuf content = entryLogger.readEntry(ledgerId, lastEntryId, entryLocation);
+            recordSuccessfulEvent(dbLedgerStorageStats.getReadFromEntryLogTime(), readEntryStartNano);
+            return content;
+        } catch (Exception e) {
+            recordFailedEvent(dbLedgerStorageStats.getReadFromEntryLogTime(), readEntryStartNano);
+            throw e;
+        }
     }
 
     @VisibleForTesting