MB-42306 [1/2]: Add onDiskPrepareBytes to vbucket_state

To allow a VBucket to track the space consumed by SyncWrite prepares,
add a new onDiskPrepareBytes member variable to vbucket_state.

* onDiskPrepareBytes: Total number of bytes that SyncWrite prepares
  take on-disk (including any compression the KVStore may have
  applied).

This is updated in a similar way to onDiskPrepares (the number of
prepares on disk).

Change-Id: I836e837ba2d2ad049cfaf606ea2eca8f8bc4b93d
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/139272
Well-Formed: Build Bot <[email protected]>
Tested-by: Dave Rigby <[email protected]>
Reviewed-by: Jim Walker <[email protected]>

Author: daverigby

engines/ep/src/couch-kvstore/couch-kvstore.cc

@@ -311,6 +311,12 @@ struct kvstats_ctx {
      * the persisted VB state before commit
      */
     size_t onDiskPrepareDelta = 0;
+
+    /**
+     * Delta of onDiskPrepareBytes that we should add to the value tracked in
+     * the persisted VB state before commit.
+     */
+    ssize_t onDiskPrepareBytesDelta = 0;
 };
 
 CouchKVStore::CouchKVStore(KVStoreConfig& config)
@@ -948,6 +954,7 @@ static int time_purge_hook(Db* d, DocInfo* info, sized_buf item, void* ctx_p) {
                 }
                 if (metadata->isPrepare()) {
                     ctx->stats.preparesPurged++;
+                    ctx->stats.prepareBytesPurged += info->size;
                 }
                 return COUCHSTORE_COMPACT_DROP_ITEM;
             } else if (info->deleted) {
@@ -998,6 +1005,7 @@ static int time_purge_hook(Db* d, DocInfo* info, sized_buf item, void* ctx_p) {
             if (metadata->isPrepare()) {
                 if (info->db_seq <= ctx->highCompletedSeqno) {
                     ctx->stats.preparesPurged++;
+                    ctx->stats.prepareBytesPurged += info->size;
                     return COUCHSTORE_COMPACT_DROP_ITEM;
                 }
 
@@ -1232,6 +1240,8 @@ bool CouchKVStore::compactDBInternal(compaction_ctx* hook_ctx,
         cachedDeleteCount[vbid.get()] = info.deleted_count;
         cachedDocCount[vbid.get()] = info.doc_count;
         state->onDiskPrepares -= hook_ctx->stats.preparesPurged;
+        state->updateOnDiskPrepareBytes(
+                -int64_t(hook_ctx->stats.prepareBytesPurged));
     }
 
     logger.debug("INFO: created new couch db file, name:{} rev:{}",
@@ -1269,7 +1279,8 @@ couchstore_error_t CouchKVStore::compactPrecommitCallback(Db& db,
 
     // we need to update the _local/vbstate to update the number of
     // on_disk_prepared to match whatever we purged
-    if (ctx.stats.preparesPurged != 0) {
+    if ((ctx.stats.preparesPurged != 0) ||
+        (ctx.stats.prepareBytesPurged != 0)) {
         LocalDoc* ldoc = nullptr;
         sized_buf id{(char*)"_local/vbstate", sizeof("_local/vbstate") - 1};
         auto err = couchstore_open_local_document(
@@ -1296,12 +1307,27 @@ couchstore_error_t CouchKVStore::compactPrecommitCallback(Db& db,
         }
         couchstore_free_local_document(ldoc);
 
-        auto iter = json.find("on_disk_prepares");
-        // only update if it's there..
-        if (iter != json.end()) {
-            auto onDiskPrepares = std::stoull(iter->get<std::string>()) -
+        bool updateVbState = false;
+
+        auto prepares = json.find("on_disk_prepares");
+        if (prepares != json.end()) {
+            auto onDiskPrepares = std::stoull(prepares->get<std::string>()) -
                                   ctx.stats.preparesPurged;
-            json["on_disk_prepares"] = std::to_string(onDiskPrepares);
+            *prepares = std::to_string(onDiskPrepares);
+            updateVbState = true;
+        }
+
+        auto prepareBytes = json.find("on_disk_prepare_bytes");
+        if (prepareBytes != json.end()) {
+            auto onDiskPrepareBytes =
+                    std::stoull(prepareBytes->get<std::string>()) -
+                    ctx.stats.prepareBytesPurged;
+            *prepareBytes = std::to_string(onDiskPrepareBytes);
+            updateVbState = true;
+        }
+
+        // only update if at least one of the prepare stats is already there
+        if (updateVbState) {
             const auto doc = json.dump();
             LocalDoc newDoc{};
             newDoc.id = id;
@@ -2230,6 +2256,12 @@ static void saveDocsCallback(const DocInfo* oldInfo,
                 std::to_string(itemCountDelta));
     }
 
+    if (newKey.isPrepared()) {
+        const ssize_t newSize = newInfo->size;
+        const ssize_t oldSize = oldInfo ? oldInfo->size : 0;
+        cbCtx->onDiskPrepareBytesDelta += (newSize - oldSize);
+    }
+
     // Do not need to update high seqno if we are calling this for a prepare and
     // it will error if we do so return early.
     if (!newKey.isCommitted()) {
@@ -2326,6 +2358,7 @@ couchstore_error_t CouchKVStore::saveDocs(Vbid vbid,
                           std::placeholders::_2));
 
         state->onDiskPrepares += kvctx.onDiskPrepareDelta;
+        state->updateOnDiskPrepareBytes(kvctx.onDiskPrepareBytesDelta);
         errCode = saveVBState(db, *state);
         if (errCode != COUCHSTORE_SUCCESS) {
             logger.warn("CouchKVStore::saveDocs: saveVBState error:{} [{}]",

engines/ep/src/ep_bucket.cc

@@ -1111,14 +1111,15 @@ void EPBucket::compactInternal(const CompactionConfig& config,
 
     EP_LOG_INFO(
             "Compaction of {} done ({}). "
-            "purged tombstones:{}, prepares:{}, "
+            "purged tombstones:{}, prepares:{}, prepareBytes:{} "
             "collection_items_erased:alive:{},deleted:{}, "
             "size/items/tombstones/purge_seqno pre{{{}, {}, {}, {}}}, "
             "post{{{}, {}, {}, {}}}",
             config.db_file_id,
             result ? "ok" : "failed",
             ctx.stats.tombstonesPurged,
             ctx.stats.preparesPurged,
+            ctx.stats.prepareBytesPurged,
             ctx.stats.collectionsItemsPurged,
             ctx.stats.collectionsDeletedItemsPurged,
             ctx.stats.pre.size,

engines/ep/src/kvstore.cc

@@ -198,6 +198,7 @@ bool KVStore::updateCachedVBState(Vbid vbid, const vbucket_state& newState) {
                     newState.highPreparedSeqno;
             vbState->maxVisibleSeqno = newState.maxVisibleSeqno;
             vbState->onDiskPrepares = newState.onDiskPrepares;
+            vbState->setOnDiskPrepareBytes(newState.getOnDiskPrepareBytes());
         } else {
             state_change_detected = false;
         }

engines/ep/src/kvstore.h

@@ -103,6 +103,7 @@ struct CompactionStats {
     size_t collectionsDeletedItemsPurged = 0;
     uint64_t tombstonesPurged = 0;
     uint64_t preparesPurged = 0;
+    uint64_t prepareBytesPurged = 0;
     FileInfo pre;
     FileInfo post;
 };

engines/ep/src/magma-kvstore/magma-kvstore.cc

@@ -1435,6 +1435,7 @@ int MagmaKVStore::saveDocs(Collections::VB::Flush& collectionsFlush,
         if (vbstate) {
             vbstate->highSeqno = lastSeqno;
             vbstate->onDiskPrepares += kvctx.onDiskPrepareDelta;
+            vbstate->onDiskPrepareBytes += kvctx.onDiskPrepareBytesDelta;
 
             auto magmaInfo = cachedMagmaInfo[vbid.get()].get();
             magmaInfo->docCount += ninserts - ndeletes;
@@ -2162,6 +2163,8 @@ bool MagmaKVStore::compactDB(compaction_ctx* ctx) {
     {
         std::lock_guard<std::shared_timed_mutex> lock(kvHandle->vbstateMutex);
         cachedVBStates[vbid.get()]->onDiskPrepares -= ctx->stats.preparesPurged;
+        cachedVBStates[vbid.get()]->onDiskPrepareBytes -=
+                ctx->stats.prepareBytesPurged;
         cachedVBStates[vbid.get()]->purgeSeqno = ctx->max_purged_seq;
         cachedMagmaInfo[vbid.get()]->docCount -=
                 ctx->stats.collectionsItemsPurged + ctx->stats.preparesPurged;

engines/ep/src/vbucket_state.cc

@@ -99,6 +99,7 @@ void vbucket_state::reset() {
     highPreparedSeqno = 0;
     maxVisibleSeqno = 0;
     onDiskPrepares = 0;
+    onDiskPrepareBytes = 0;
     transition = vbucket_transition_state{};
 }
 
@@ -120,6 +121,7 @@ bool vbucket_state::operator==(const vbucket_state& other) const {
                 other.highPreparedSeqno);
     rv = rv && (maxVisibleSeqno == other.maxVisibleSeqno);
     rv = rv && (onDiskPrepares == other.onDiskPrepares);
+    rv = rv && (onDiskPrepareBytes == other.onDiskPrepareBytes);
     rv = rv && (checkpointType == other.checkpointType);
     rv = rv && (transition == other.transition);
     return rv;
@@ -129,6 +131,14 @@ bool vbucket_state::operator!=(const vbucket_state& other) const {
     return !(*this == other);
 }
 
+void vbucket_state::updateOnDiskPrepareBytes(int64_t delta) {
+    // Note: onDiskPrepareBytes was only added in 6.6.1 (where it is
+    // initialized to zero). As such, if using files created before
+    // then, we need to ensure that onDiskPrepareBytes is capped at zero
+    // and doesn't underflow.
+    onDiskPrepareBytes += std::max(-int64_t(onDiskPrepareBytes), delta);
+}
+
 void to_json(nlohmann::json& json, const vbucket_state& vbs) {
     // First add all required fields.
     // Note that integers are stored as strings to avoid any undesired
@@ -152,6 +162,8 @@ void to_json(nlohmann::json& json, const vbucket_state& vbs) {
             {"high_prepared_seqno", std::to_string(vbs.highPreparedSeqno)},
             {"max_visible_seqno", std::to_string(vbs.maxVisibleSeqno)},
             {"on_disk_prepares", std::to_string(vbs.onDiskPrepares)},
+            {"on_disk_prepare_bytes",
+             std::to_string(vbs.getOnDiskPrepareBytes())},
             {"checkpoint_type", to_string(vbs.checkpointType)}};
 
     to_json(json, vbs.transition);
@@ -238,6 +250,10 @@ void from_json(const nlohmann::json& j, vbucket_state& vbs) {
     // Note: We don't track on disk prepares pre-6.5
     vbs.onDiskPrepares = std::stoll(j.value("on_disk_prepares", "0"));
 
+    // Note: We don't track on disk prepare total size pre-6.6.1.
+    vbs.setOnDiskPrepareBytes(
+            std::stoll(j.value("on_disk_prepare_bytes", "0")));
+
     // Note: We don't track checkpoint type pre-6.5
     auto checkpointType = j.find("checkpoint_type");
     if (checkpointType != j.end()) {

engines/ep/src/vbucket_state.h

@@ -89,11 +89,26 @@ struct vbucket_state {
      * v3: Mad-Hatter. high_completed_seqno and high_prepared_seqno added along
      *     with counter for number of prepares on disk. Checkpoint-ID no longer
      *     stored (and ignored during upgrade)
+     * v4: 6.6.1, added with MB-32670. Adds onDiskPrepareBytes.
      */
-    static constexpr int CurrentVersion = 3;
+    static constexpr int CurrentVersion = 4;
 
     bool needsToBePersisted(const vbucket_state& vbstate);
 
+    uint64_t getOnDiskPrepareBytes() const {
+        return onDiskPrepareBytes;
+    }
+
+    void setOnDiskPrepareBytes(int64_t value) {
+        onDiskPrepareBytes = value;
+    }
+
+    /**
+     * Apply the given delta to the value of onDiskPrepareBytes. Clamps
+     * onDiskPrepareBytes at zero (avoids underflow).
+     */
+    void updateOnDiskPrepareBytes(int64_t delta);
+
     void reset();
 
     bool operator==(const vbucket_state& other) const;
@@ -186,6 +201,19 @@ struct vbucket_state {
      */
     uint64_t onDiskPrepares = 0;
 
+private:
+    /**
+     * Size in bytes of on disk prepares (Pending SyncWrites). Required to
+     * estimate the size of completed prepares as part of calculating the
+     * couchstore 'stale' data overhead.
+     * Note this is private as care needs to be taken when modifying it to
+     * ensure it doesn't underflow.
+     *
+     * Added for SyncReplication (MB-42306) in 6.6.1.
+     */
+    uint64_t onDiskPrepareBytes = 0;
+
+public:
     /**
      * The type of the most recently persisted snapshot disk. Required as we
      * cannot rely on the HCS in the middle of a snapshot to optimise warmup and