SERVER-51021 Add currentOp reporting for ReshardingRecipientService Instances

5 files changed, 108 insertions, 10 deletions

diff --git a/jstests/sharding/resharding_metrics.js b/jstests/sharding/resharding_metrics.js
index 051a809b379..cf0d3b7f070 100644
--- a/jstests/sharding/resharding_metrics.js
+++ b/jstests/sharding/resharding_metrics.js
@@ -109,7 +109,27 @@ function verifyCurrentOpOutput(reshardingTest, inputCollection) {
         });
     });
 
-    // TODO SERVER-51021 verify currentOp output for recipients
+    reshardingTest.recipientShardNames.forEach(function(shardName) {
+        checkCurrentOp(new Mongo(topology.shards[shardName].primary), shardName, "Recipient", {
+            "type": "op",
+            "op": "command",
+            "ns": kNamespace,
+            "originatingCommand": undefined,
+            "totalOperationTimeElapsed": undefined,
+            "remainingOperationTimeEstimated": undefined,
+            "approxDocumentsToCopy": undefined,
+            "approxBytesToCopy": undefined,
+            "documentsCopied": undefined,
+            "bytesCopied": undefined,
+            "totalCopyTimeElapsed": undefined,
+            "oplogEntriesFetched": undefined,
+            "oplogEntriesApplied": undefined,
+            "totalApplyTimeElapsed": undefined,
+            "recipientState": undefined,
+            "opStatus": "actively running",
+        });
+    });
+
     // TODO SERVER-50976 verify currentOp output for the coordinator
 }
 
diff --git a/src/mongo/db/s/resharding/resharding_metrics.cpp b/src/mongo/db/s/resharding/resharding_metrics.cpp
index 6e4833813d1..d7063bf07ad 100644
--- a/src/mongo/db/s/resharding/resharding_metrics.cpp
+++ b/src/mongo/db/s/resharding/resharding_metrics.cpp
@@ -288,8 +288,8 @@ void ReshardingMetrics::OperationMetrics::append(BSONObjBuilder* bob, Role role)
             bob->append(kCompletionStatus, OperationStatus_serializer(operationStatus));
             break;
         case Role::kRecipient:
-            // TODO SERVER-51021
-            MONGO_UNREACHABLE;
+            bob->append(kRecipientState, RecipientState_serializer(recipientState));
+            bob->append(kCompletionStatus, OperationStatus_serializer(operationStatus));
             break;
         case Role::kCoordinator:
             // TODO SERVER-50976
diff --git a/src/mongo/db/s/resharding/resharding_metrics_test.cpp b/src/mongo/db/s/resharding/resharding_metrics_test.cpp
index eea7789e682..f22584eec0a 100644
--- a/src/mongo/db/s/resharding/resharding_metrics_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_metrics_test.cpp
@@ -256,4 +256,75 @@ TEST_F(ReshardingMetricsTest, CurrentOpReportForDonor) {
     ASSERT_BSONOBJ_EQ(expected, report);
 }
 
+TEST_F(ReshardingMetricsTest, CurrentOpReportForRecipient) {
+    const auto kRecipientState = RecipientStateEnum::kCloning;
+
+    constexpr auto kDocumentsToCopy = 500;
+    constexpr auto kDocumentsCopied = kDocumentsToCopy * 0.5;
+    static_assert(kDocumentsToCopy >= kDocumentsCopied);
+
+    constexpr auto kBytesToCopy = 8192;
+    constexpr auto kBytesCopied = kBytesToCopy * 0.5;
+    static_assert(kBytesToCopy >= kBytesCopied);
+
+    constexpr auto kDelayBeforeCloning = Seconds(2);
+    getMetrics()->onStart();
+    advanceTime(kDelayBeforeCloning);
+
+    constexpr auto kTimeSpentCloning = Seconds(3);
+    getMetrics()->setRecipientState(kRecipientState);
+    getMetrics()->setDocumentsToCopy(kDocumentsToCopy, kBytesToCopy);
+    advanceTime(kTimeSpentCloning);
+    getMetrics()->onDocumentsCopied(kDocumentsCopied, kBytesCopied);
+
+    const auto kTimeToCopyRemainingSeconds =
+        durationCount<Seconds>(kTimeSpentCloning) * (kBytesToCopy / kBytesCopied - 1);
+    const auto kRemainingOperationTimeSeconds =
+        durationCount<Seconds>(kTimeSpentCloning) + 2 * kTimeToCopyRemainingSeconds;
+
+    const ReshardingMetrics::ReporterOptions options(
+        ReshardingMetrics::ReporterOptions::Role::kRecipient,
+        UUID::parse("12345678-1234-1234-1234-123456789def").getValue(),
+        NamespaceString("db", "collection"),
+        BSON("id" << 1),
+        false);
+
+    const auto expected =
+        fromjson(fmt::format("{{ type: \"op\","
+                             "desc: \"ReshardingRecipientService {0}\","
+                             "op: \"command\","
+                             "ns: \"{1}\","
+                             "originatingCommand: {{ reshardCollection: \"{1}\","
+                             "key: {2},"
+                             "unique: {3},"
+                             "collation: {{ locale: \"simple\" }} }},"
+                             "totalOperationTimeElapsed: {4},"
+                             "remainingOperationTimeEstimated: {5},"
+                             "approxDocumentsToCopy: {6},"
+                             "documentsCopied: {7},"
+                             "approxBytesToCopy: {8},"
+                             "bytesCopied: {9},"
+                             "totalCopyTimeElapsed: {10},"
+                             "oplogEntriesFetched: 0,"
+                             "oplogEntriesApplied: 0,"
+                             "totalApplyTimeElapsed: 0,"
+                             "recipientState: \"{11}\","
+                             "opStatus: \"actively running\" }}",
+                             options.id.toString(),
+                             options.nss.toString(),
+                             options.shardKey.toString(),
+                             options.unique ? "true" : "false",
+                             durationCount<Seconds>(kDelayBeforeCloning + kTimeSpentCloning),
+                             kRemainingOperationTimeSeconds,
+                             kDocumentsToCopy,
+                             kDocumentsCopied,
+                             kBytesToCopy,
+                             kBytesCopied,
+                             durationCount<Seconds>(kTimeSpentCloning),
+                             RecipientState_serializer(kRecipientState)));
+
+    const auto report = getMetrics()->reportForCurrentOp(options);
+    ASSERT_BSONOBJ_EQ(expected, report);
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service.cpp b/src/mongo/db/s/resharding/resharding_recipient_service.cpp
index b6ac78b8da9..548072bc3d4 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service.cpp
+++ b/src/mongo/db/s/resharding/resharding_recipient_service.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/repl/read_concern_args.h"
 #include "mongo/db/s/migration_destination_manager.h"
 #include "mongo/db/s/resharding/resharding_collection_cloner.h"
+#include "mongo/db/s/resharding/resharding_metrics.h"
 #include "mongo/db/s/resharding/resharding_server_parameters_gen.h"
 #include "mongo/db/s/resharding_util.h"
 #include "mongo/db/s/shard_key_util.h"
@@ -238,6 +239,17 @@ void ReshardingRecipientService::RecipientStateMachine::interrupt(Status status)
     }
 }
 
+boost::optional<BSONObj> ReshardingRecipientService::RecipientStateMachine::reportForCurrentOp(
+    MongoProcessInterface::CurrentOpConnectionsMode,
+    MongoProcessInterface::CurrentOpSessionsMode) noexcept {
+    ReshardingMetrics::ReporterOptions options(ReshardingMetrics::ReporterOptions::Role::kRecipient,
+                                               _id,
+                                               _recipientDoc.getNss(),
+                                               _recipientDoc.getReshardingKey().toBSON(),
+                                               false);
+    return ReshardingMetrics::get(cc().getServiceContext())->reportForCurrentOp(options);
+}
+
 void ReshardingRecipientService::RecipientStateMachine::onReshardingFieldsChanges(
     const TypeCollectionReshardingFields& reshardingFields) {
     auto coordinatorState = reshardingFields.getState();
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service.h b/src/mongo/db/s/resharding/resharding_recipient_service.h
index 0967b4bf930..045a55bc1a8 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service.h
+++ b/src/mongo/db/s/resharding/resharding_recipient_service.h
@@ -107,14 +107,9 @@ public:
         return _completionPromise.getFuture();
     }
 
-    /**
-     * TODO(SERVER-51021) Report ReshardingRecipientService Instances in currentOp().
-     */
     boost::optional<BSONObj> reportForCurrentOp(
-        MongoProcessInterface::CurrentOpConnectionsMode connMode,
-        MongoProcessInterface::CurrentOpSessionsMode sessionMode) noexcept final {
-        return boost::none;
-    }
+        MongoProcessInterface::CurrentOpConnectionsMode,
+        MongoProcessInterface::CurrentOpSessionsMode) noexcept override;
 
     void onReshardingFieldsChanges(const TypeCollectionReshardingFields& reshardingFields);