SERVER-36501 serverStatus support for prepared transactions

7 files changed, 236 insertions, 11 deletions

diff --git a/src/mongo/db/server_transactions_metrics.cpp b/src/mongo/db/server_transactions_metrics.cpp
index 3386bc19c42..627953e54fb 100644
--- a/src/mongo/db/server_transactions_metrics.cpp
+++ b/src/mongo/db/server_transactions_metrics.cpp
@@ -113,7 +113,31 @@ void ServerTransactionsMetrics::incrementTotalCommitted() {
     _totalCommitted.fetchAndAdd(1);
 }
 
-boost::optional<repl::OpTime> ServerTransactionsMetrics::getOldestActiveOpTime() const {
+unsigned long long ServerTransactionsMetrics::getTotalPrepared() const {
+    return _totalPrepared.load();
+}
+
+void ServerTransactionsMetrics::incrementTotalPrepared() {
+    _totalPrepared.fetchAndAdd(1);
+}
+
+unsigned long long ServerTransactionsMetrics::getTotalPreparedThenCommitted() const {
+    return _totalPreparedThenCommitted.load();
+}
+
+void ServerTransactionsMetrics::incrementTotalPreparedThenCommitted() {
+    _totalPreparedThenCommitted.fetchAndAdd(1);
+}
+
+unsigned long long ServerTransactionsMetrics::getTotalPreparedThenAborted() const {
+    return _totalPreparedThenAborted.load();
+}
+
+void ServerTransactionsMetrics::incrementTotalPreparedThenAborted() {
+    _totalPreparedThenAborted.fetchAndAdd(1);
+}
+
+boost::optional<repl::OpTime> ServerTransactionsMetrics::_calculateOldestActiveOpTime() const {
     if (_oldestActiveOplogEntryOpTimes.empty()) {
         return boost::none;
     }
@@ -140,6 +164,7 @@ void ServerTransactionsMetrics::addActiveOpTime(repl::OpTime oldestOplogEntryOpT
                             << "oldestNonMajorityCommittedOpTimes."
                             << "oldestOplogEntryOpTime: "
                             << oldestOplogEntryOpTime.toString());
+    _oldestActiveOplogEntryOpTime = _calculateOldestActiveOpTime();
 }
 
 void ServerTransactionsMetrics::removeActiveOpTime(repl::OpTime oldestOplogEntryOpTime,
@@ -183,6 +208,7 @@ void ServerTransactionsMetrics::removeActiveOpTime(repl::OpTime oldestOplogEntry
                             << oldestOplogEntryOpTime.toString()
                             << "finishOpTime: "
                             << finishOpTime->toString());
+    _oldestActiveOplogEntryOpTime = _calculateOldestActiveOpTime();
 }
 
 boost::optional<repl::OpTime> ServerTransactionsMetrics::getOldestNonMajorityCommittedOpTime()
@@ -219,6 +245,10 @@ ServerTransactionsMetrics::getFinishOpTimeOfOldestNonMajCommitted_forTest() cons
     return oldestNonMajorityCommittedOpTime;
 }
 
+boost::optional<repl::OpTime> ServerTransactionsMetrics::getOldestActiveOpTime() const {
+    return _oldestActiveOplogEntryOpTime;
+}
+
 unsigned int ServerTransactionsMetrics::getTotalActiveOpTimes() const {
     return _oldestActiveOplogEntryOpTimes.size();
 }
@@ -230,6 +260,15 @@ void ServerTransactionsMetrics::updateStats(TransactionsStats* stats) {
     stats->setTotalAborted(_totalAborted.load());
     stats->setTotalCommitted(_totalCommitted.load());
     stats->setTotalStarted(_totalStarted.load());
+    stats->setTotalPrepared(_totalPrepared.load());
+    stats->setTotalPreparedThenCommitted(_totalPreparedThenCommitted.load());
+    stats->setTotalPreparedThenAborted(_totalPreparedThenAborted.load());
+    // To avoid compression loss, we have Timestamp(0, 0) be the default value if no oldest active
+    // transaction optime is stored.
+    Timestamp oldestActiveOplogEntryTimestamp = (_oldestActiveOplogEntryOpTime != boost::none)
+        ? _oldestActiveOplogEntryOpTime->getTimestamp()
+        : Timestamp();
+    stats->setOldestActiveOplogEntryTimestamp(oldestActiveOplogEntryTimestamp);
 }
 
 class TransactionsSSS : public ServerStatusSection {
diff --git a/src/mongo/db/server_transactions_metrics.h b/src/mongo/db/server_transactions_metrics.h
index 7dbc208e0c6..be872863067 100644
--- a/src/mongo/db/server_transactions_metrics.h
+++ b/src/mongo/db/server_transactions_metrics.h
@@ -72,6 +72,15 @@ public:
     unsigned long long getTotalCommitted() const;
     void incrementTotalCommitted();
 
+    unsigned long long getTotalPrepared() const;
+    void incrementTotalPrepared();
+
+    unsigned long long getTotalPreparedThenCommitted() const;
+    void incrementTotalPreparedThenCommitted();
+
+    unsigned long long getTotalPreparedThenAborted() const;
+    void incrementTotalPreparedThenAborted();
+
     /**
      * Returns the OpTime of the oldest oplog entry written across all open transactions.
      * Returns boost::none if there are no transaction oplog entry OpTimes stored.
@@ -130,6 +139,12 @@ public:
     void updateStats(TransactionsStats* stats);
 
 private:
+    /**
+     * Returns the first and oldest optime in the ordered set of active oplog entry optimes.
+     * Returns boost::none if there are no transaction oplog entry optimes stored.
+     */
+    boost::optional<repl::OpTime> _calculateOldestActiveOpTime() const;
+
     // The number of multi-document transactions currently active.
     AtomicUInt64 _currentActive{0};
 
@@ -148,6 +163,18 @@ private:
     // The total number of multi-document transaction commits.
     AtomicUInt64 _totalCommitted{0};
 
+    // The total number of prepared transactions since the last server startup.
+    AtomicUInt64 _totalPrepared{0};
+
+    // The total number of prepared transaction commits.
+    AtomicUInt64 _totalPreparedThenCommitted{0};
+
+    // The total number of prepared transaction aborts.
+    AtomicUInt64 _totalPreparedThenAborted{0};
+
+    // The optime of the oldest oplog entry for any active transaction.
+    boost::optional<repl::OpTime> _oldestActiveOplogEntryOpTime;
+
     // Maintain the oldest oplog entry OpTime across all active transactions. Currently, we only
     // write an oplog entry for an ongoing transaction if it is in the `prepare` state. By
     // maintaining an ordered set of OpTimes, the OpTime at the beginning will be the oldest.
diff --git a/src/mongo/db/transaction_metrics_observer.cpp b/src/mongo/db/transaction_metrics_observer.cpp
index eea17ad5379..28a6f26e696 100644
--- a/src/mongo/db/transaction_metrics_observer.cpp
+++ b/src/mongo/db/transaction_metrics_observer.cpp
@@ -100,7 +100,8 @@ void TransactionMetricsObserver::onCommit(ServerTransactionsMetrics* serverTrans
                                           TickSource* tickSource,
                                           boost::optional<repl::OpTime> oldestOplogEntryOpTime,
                                           boost::optional<repl::OpTime> commitOpTime,
-                                          Top* top) {
+                                          Top* top,
+                                          bool wasPrepared) {
     invariant((oldestOplogEntryOpTime != boost::none && commitOpTime != boost::none) ||
               (oldestOplogEntryOpTime == boost::none && commitOpTime == boost::none));
     //
@@ -123,6 +124,10 @@ void TransactionMetricsObserver::onCommit(ServerTransactionsMetrics* serverTrans
     serverTransactionsMetrics->decrementCurrentOpen();
     serverTransactionsMetrics->decrementCurrentActive();
 
+    if (wasPrepared) {
+        serverTransactionsMetrics->incrementTotalPreparedThenCommitted();
+    }
+
     auto duration =
         durationCount<Microseconds>(_singleTransactionStats.getDuration(tickSource, curTick));
     top->incrementGlobalTransactionLatencyStats(static_cast<uint64_t>(duration));
@@ -137,7 +142,8 @@ void TransactionMetricsObserver::onAbortActive(ServerTransactionsMetrics* server
                                                TickSource* tickSource,
                                                boost::optional<repl::OpTime> oldestOplogEntryOpTime,
                                                boost::optional<repl::OpTime> abortOpTime,
-                                               Top* top) {
+                                               Top* top,
+                                               bool wasPrepared) {
     invariant((oldestOplogEntryOpTime != boost::none && abortOpTime != boost::none) ||
               (oldestOplogEntryOpTime == boost::none && abortOpTime == boost::none));
 
@@ -157,6 +163,10 @@ void TransactionMetricsObserver::onAbortActive(ServerTransactionsMetrics* server
     //
     serverTransactionsMetrics->decrementCurrentActive();
 
+    if (wasPrepared) {
+        serverTransactionsMetrics->incrementTotalPreparedThenAborted();
+    }
+
     // Remove this transaction's oldest oplog entry OpTime if one was written.
     if (oldestOplogEntryOpTime) {
         serverTransactionsMetrics->removeActiveOpTime(*oldestOplogEntryOpTime, abortOpTime);
@@ -215,11 +225,13 @@ void TransactionMetricsObserver::_onAbort(ServerTransactionsMetrics* serverTrans
 }
 
 void TransactionMetricsObserver::onPrepare(ServerTransactionsMetrics* serverTransactionsMetrics,
-                                           repl::OpTime prepareOpTime) {
+                                           repl::OpTime prepareOpTime,
+                                           TickSource::Tick curTick) {
     // Since we currently only write an oplog entry for an in progress transaction when it is in
     // the prepare state, the prepareOpTime is currently the oldest OpTime written to the
     // oplog for this transaction.
     serverTransactionsMetrics->addActiveOpTime(prepareOpTime);
+    serverTransactionsMetrics->incrementTotalPrepared();
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/transaction_metrics_observer.h b/src/mongo/db/transaction_metrics_observer.h
index 715aa14244f..6600615ceac 100644
--- a/src/mongo/db/transaction_metrics_observer.h
+++ b/src/mongo/db/transaction_metrics_observer.h
@@ -77,7 +77,8 @@ public:
                   TickSource* tickSource,
                   boost::optional<repl::OpTime> oldestOplogEntryOpTime,
                   boost::optional<repl::OpTime> commitOpTime,
-                  Top* top);
+                  Top* top,
+                  bool wasPrepared);
 
     /**
      * Updates relevant metrics when an active transaction aborts. Also removes this transaction's
@@ -89,7 +90,8 @@ public:
                        TickSource* tickSource,
                        boost::optional<repl::OpTime> oldestOplogEntryOpTime,
                        boost::optional<repl::OpTime> abortOpTime,
-                       Top* top);
+                       Top* top,
+                       bool wasPrepared);
 
     /**
      * Updates relevant metrics when an inactive transaction aborts. Also removes this transaction's
@@ -109,7 +111,8 @@ public:
      * an active transaction, to the oldestActiveOplogEntryTS set.
      */
     void onPrepare(ServerTransactionsMetrics* serverTransactionsMetrics,
-                   repl::OpTime prepareOpTime);
+                   repl::OpTime prepareOpTime,
+                   TickSource::Tick curTick);
 
     /**
      * Updates relevant metrics when an operation running on the transaction completes. An operation
diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp
index 17d440103a2..68e85ca5ec8 100644
--- a/src/mongo/db/transaction_participant.cpp
+++ b/src/mongo/db/transaction_participant.cpp
@@ -835,8 +835,10 @@ Timestamp TransactionParticipant::prepareTransaction(OperationContext* opCtx,
     // but this will change when we allow multiple oplog entries per transaction.
     {
         stdx::lock_guard<stdx::mutex> lm(_metricsMutex);
+        const auto tickSource = getGlobalServiceContext()->getTickSource();
         _transactionMetricsObserver.onPrepare(ServerTransactionsMetrics::get(opCtx),
-                                              *_oldestOplogEntryOpTime);
+                                              *_oldestOplogEntryOpTime,
+                                              tickSource->getTicks());
     }
 
     return prepareOplogSlot.opTime.getTimestamp();
@@ -1012,7 +1014,7 @@ void TransactionParticipant::_finishCommitTransaction(WithLock lk, OperationCont
     if (_speculativeTransactionReadOpTime > clientInfo.getLastOp()) {
         clientInfo.setLastOp(_speculativeTransactionReadOpTime);
     }
-
+    const bool isCommittingWithPrepare = _txnState.isCommittingWithPrepare(lk);
     _txnState.transitionTo(lk, TransactionState::kCommitted);
 
     {
@@ -1022,7 +1024,8 @@ void TransactionParticipant::_finishCommitTransaction(WithLock lk, OperationCont
                                              tickSource,
                                              _oldestOplogEntryOpTime,
                                              _finishOpTime,
-                                             &Top::get(getGlobalServiceContext()));
+                                             &Top::get(getGlobalServiceContext()),
+                                             isCommittingWithPrepare);
         _transactionMetricsObserver.onTransactionOperation(
             opCtx->getClient(), CurOp::get(opCtx)->debug().additiveMetrics);
     }
@@ -1194,7 +1197,8 @@ void TransactionParticipant::_abortTransactionOnSession(WithLock wl) {
             tickSource,
             _oldestOplogEntryOpTime,
             _finishOpTime,
-            &Top::get(getGlobalServiceContext()));
+            &Top::get(getGlobalServiceContext()),
+            _txnState.isPrepared(lm));
     }
 
     _transactionOperationBytes = 0;
diff --git a/src/mongo/db/transaction_participant_test.cpp b/src/mongo/db/transaction_participant_test.cpp
index d8e68714408..8330d1dc187 100644
--- a/src/mongo/db/transaction_participant_test.cpp
+++ b/src/mongo/db/transaction_participant_test.cpp
@@ -1682,6 +1682,17 @@ TEST_F(TransactionsMetricsTest, IncrementTotalStartedUponStartTransaction) {
               beforeTransactionStart + 1U);
 }
 
+TEST_F(TransactionsMetricsTest, IncrementPreparedTransaction) {
+    OperationContextSessionMongod opCtxSession(opCtx(), true, makeSessionInfo());
+    auto txnParticipant = TransactionParticipant::get(opCtx());
+    unsigned long long beforePrepareCount =
+        ServerTransactionsMetrics::get(opCtx())->getTotalPrepared();
+    txnParticipant->unstashTransactionResources(opCtx(), "prepareTransaction");
+    txnParticipant->prepareTransaction(opCtx(), {});
+
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getTotalPrepared(), beforePrepareCount + 1U);
+}
+
 TEST_F(TransactionsMetricsTest, IncrementTotalCommittedOnCommit) {
     OperationContextSessionMongod opCtxSession(opCtx(), true, makeSessionInfo());
     auto txnParticipant = TransactionParticipant::get(opCtx());
@@ -1696,6 +1707,23 @@ TEST_F(TransactionsMetricsTest, IncrementTotalCommittedOnCommit) {
     ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getTotalCommitted(), beforeCommitCount + 1U);
 }
 
+TEST_F(TransactionsMetricsTest, IncrementTotalPreparedThenCommitted) {
+    OperationContextSessionMongod opCtxSession(opCtx(), true, makeSessionInfo());
+    auto txnParticipant = TransactionParticipant::get(opCtx());
+    txnParticipant->unstashTransactionResources(opCtx(), "commitTransaction");
+    const auto prepareTimestamp = txnParticipant->prepareTransaction(opCtx(), {});
+
+    unsigned long long beforePreparedThenCommittedCount =
+        ServerTransactionsMetrics::get(opCtx())->getTotalPreparedThenCommitted();
+
+    txnParticipant->commitPreparedTransaction(opCtx(), prepareTimestamp);
+
+    ASSERT_TRUE(txnParticipant->transactionIsCommitted());
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getTotalPreparedThenCommitted(),
+              beforePreparedThenCommittedCount + 1U);
+}
+
+
 TEST_F(TransactionsMetricsTest, IncrementTotalAbortedUponAbort) {
     OperationContextSessionMongod opCtxSession(opCtx(), true, makeSessionInfo());
     auto txnParticipant = TransactionParticipant::get(opCtx());
@@ -1710,6 +1738,22 @@ TEST_F(TransactionsMetricsTest, IncrementTotalAbortedUponAbort) {
     ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getTotalAborted(), beforeAbortCount + 1U);
 }
 
+TEST_F(TransactionsMetricsTest, IncrementTotalPreparedThenAborted) {
+    unsigned long long beforePreparedThenAbortedCount =
+        ServerTransactionsMetrics::get(opCtx())->getTotalPreparedThenAborted();
+
+    OperationContextSessionMongod opCtxSession(opCtx(), true, makeSessionInfo());
+    auto txnParticipant = TransactionParticipant::get(opCtx());
+    txnParticipant->unstashTransactionResources(opCtx(), "prepareTransaction");
+    txnParticipant->prepareTransaction(opCtx(), {});
+
+    txnParticipant->abortActiveTransaction(opCtx());
+    ASSERT(txnParticipant->transactionIsAborted());
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getTotalPreparedThenAborted(),
+              beforePreparedThenAbortedCount + 1U);
+}
+
+
 TEST_F(TransactionsMetricsTest, TrackTotalOpenTransactionsWithAbort) {
     unsigned long long beforeTransactionStart =
         ServerTransactionsMetrics::get(opCtx())->getCurrentOpen();
@@ -1854,6 +1898,91 @@ TEST_F(TransactionsMetricsTest, TrackTotalActiveAndInactiveTransactionsWithUnsta
     ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentInactive(), beforeInactiveCounter);
 }
 
+TEST_F(TransactionsMetricsTest, TrackCurrentActiveAndInactivePreparedTransactionsOnCommit) {
+    unsigned long long beforeActivePreparedCounter =
+        ServerTransactionsMetrics::get(opCtx())->getCurrentActive();
+    unsigned long long beforeInactivePreparedCounter =
+        ServerTransactionsMetrics::get(opCtx())->getCurrentInactive();
+    OperationContextSessionMongod opCtxSession(opCtx(), true, makeSessionInfo());
+    unsigned long long beforePrepareCount =
+        ServerTransactionsMetrics::get(opCtx())->getTotalPrepared();
+    unsigned long long beforePreparedThenCommittedCount =
+        ServerTransactionsMetrics::get(opCtx())->getTotalPreparedThenCommitted();
+
+    // Tests that unstashing a transaction puts it into an active state.
+    auto txnParticipant = TransactionParticipant::get(opCtx());
+    txnParticipant->unstashTransactionResources(opCtx(), "prepareTransaction");
+    const auto prepareTimestamp = txnParticipant->prepareTransaction(opCtx(), {});
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentActive(),
+              beforeActivePreparedCounter + 1U);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentInactive(),
+              beforeInactivePreparedCounter);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getTotalPrepared(), beforePrepareCount + 1U);
+
+    // Tests that the first stash decrements the active counter and increments the inactive counter.
+    txnParticipant->stashTransactionResources(opCtx());
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentActive(),
+              beforeActivePreparedCounter);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentInactive(),
+              beforeInactivePreparedCounter + 1U);
+
+    // Tests that unstashing increments the active counter and decrements the inactive counter.
+    txnParticipant->unstashTransactionResources(opCtx(), "commitTransaction");
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentActive(),
+              beforeActivePreparedCounter + 1U);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentInactive(),
+              beforeInactivePreparedCounter);
+
+    // Tests that committing decrements the active counter only.
+    txnParticipant->commitPreparedTransaction(opCtx(), prepareTimestamp);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentActive(),
+              beforeActivePreparedCounter);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentInactive(),
+              beforeInactivePreparedCounter);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getTotalPreparedThenCommitted(),
+              beforePreparedThenCommittedCount + 1U);
+}
+
+TEST_F(TransactionsMetricsTest,
+       TrackCurrentActiveAndInactivePreparedTransactionsWithUnstashedAbort) {
+    unsigned long long beforeActivePreparedCounter =
+        ServerTransactionsMetrics::get(opCtx())->getCurrentActive();
+    unsigned long long beforeInactivePreparedCounter =
+        ServerTransactionsMetrics::get(opCtx())->getCurrentInactive();
+    OperationContextSessionMongod opCtxSession(opCtx(), true, makeSessionInfo());
+
+    auto txnParticipant = TransactionParticipant::get(opCtx());
+
+    // Tests that unstashing a transaction increments the active counter only.
+    txnParticipant->unstashTransactionResources(opCtx(), "prepareTransaction");
+    txnParticipant->prepareTransaction(opCtx(), {});
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentActive(),
+              beforeActivePreparedCounter + 1U);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentInactive(),
+              beforeInactivePreparedCounter);
+
+    // Tests that stashing a prepared transaction decrements the active counter and increments the
+    // inactive counter.
+    txnParticipant->stashTransactionResources(opCtx());
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentActive(),
+              beforeActivePreparedCounter);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentInactive(),
+              beforeInactivePreparedCounter + 1U);
+
+    // Tests that aborting a stashed prepared transaction decrements the inactive counter only.
+    txnParticipant->unstashTransactionResources(opCtx(), "abortTransaction");
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentActive(),
+              beforeActivePreparedCounter + 1U);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentInactive(),
+              beforeInactivePreparedCounter);
+    txnParticipant->abortActiveTransaction(opCtx());
+    ASSERT(txnParticipant->transactionIsAborted());
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentActive(),
+              beforeActivePreparedCounter);
+    ASSERT_EQ(ServerTransactionsMetrics::get(opCtx())->getCurrentInactive(),
+              beforeInactivePreparedCounter);
+}
+
 TEST_F(TransactionsMetricsTest, SingleTransactionStatsDurationShouldBeSetUponCommit) {
     auto tickSource = initMockTickSource();
 
diff --git a/src/mongo/db/transactions_stats.idl b/src/mongo/db/transactions_stats.idl
index b47f49d2b6b..13c9843d1a5 100644
--- a/src/mongo/db/transactions_stats.idl
+++ b/src/mongo/db/transactions_stats.idl
@@ -70,3 +70,14 @@ structs:
       totalStarted:
         type: long
         default: 0
+      totalPrepared:
+        type: long
+        default: 0
+      totalPreparedThenCommitted:
+        type: long
+        default: 0
+      totalPreparedThenAborted:
+        type: long
+        default: 0
+      oldestActiveOplogEntryTimestamp:
+        type: timestamp