SERVER-19855 Include min OpTime with shard version

This change adds the OpTime of the chunk manager as of the time its cached
metadata was loaded along with each versioned request sent from MongoS.
This includes write commands and the setShardVersion command.

The OpTime is only sent as part of this change. There will be a follow-up
change to add code to interpret this information on the MongoD side.

9 files changed, 79 insertions, 124 deletions

diff --git a/src/mongo/s/write_ops/batch_write_op.cpp b/src/mongo/s/write_ops/batch_write_op.cpp
index 113dd9133f8..ba7ddae8d95 100644
--- a/src/mongo/s/write_ops/batch_write_op.cpp
+++ b/src/mongo/s/write_ops/batch_write_op.cpp
@@ -87,11 +87,13 @@ static int compareEndpoints(const ShardEndpoint* endpointA, const ShardEndpoint*
     if (shardNameDiff != 0)
         return shardNameDiff;
 
-    long shardVersionDiff = endpointA->shardVersion.toLong() - endpointB->shardVersion.toLong();
+    long shardVersionDiff = endpointA->shardVersion.getVersion().toLong() -
+        endpointB->shardVersion.getVersion().toLong();
     if (shardVersionDiff != 0)
         return shardVersionDiff;
 
-    int shardEpochDiff = endpointA->shardVersion.epoch().compare(endpointB->shardVersion.epoch());
+    int shardEpochDiff = endpointA->shardVersion.getVersion().epoch().compare(
+        endpointB->shardVersion.getVersion().epoch());
     return shardEpochDiff;
 }
 
@@ -456,7 +458,8 @@ void BatchWriteOp::buildBatchRequest(const TargetedWriteBatch& targetedBatch,
 
     unique_ptr<BatchedRequestMetadata> requestMetadata(new BatchedRequestMetadata());
     requestMetadata->setShardName(targetedBatch.getEndpoint().shardName);
-    requestMetadata->setShardVersion(targetedBatch.getEndpoint().shardVersion);
+    requestMetadata->setShardVersion(
+        ChunkVersionAndOpTime(targetedBatch.getEndpoint().shardVersion));
     requestMetadata->setSession(0);
     request->setMetadata(requestMetadata.release());
 }
diff --git a/src/mongo/s/write_ops/batched_delete_request_test.cpp b/src/mongo/s/write_ops/batched_delete_request_test.cpp
index 87d883f1b0b..c8f71652599 100644
--- a/src/mongo/s/write_ops/batched_delete_request_test.cpp
+++ b/src/mongo/s/write_ops/batched_delete_request_test.cpp
@@ -46,32 +46,24 @@ TEST(BatchedDeleteRequest, Basic) {
         BSON(BatchedDeleteDocument::query(BSON("a" << 1)) << BatchedDeleteDocument::limit(1))
         << BSON(BatchedDeleteDocument::query(BSON("b" << 1)) << BatchedDeleteDocument::limit(1)));
 
-    BSONObj writeConcernObj = BSON("w" << 1);
-
-    // The BSON_ARRAY macro doesn't support Timestamps.
-    BSONArrayBuilder arrBuilder;
-    arrBuilder.append(Timestamp(1, 1));
-    arrBuilder.append(OID::gen());
-    BSONArray shardVersionArray = arrBuilder.arr();
-
-    BSONObj origDeleteRequestObj =
-        BSON(BatchedDeleteRequest::collName("test")
-             << BatchedDeleteRequest::deletes() << deleteArray
-             << BatchedDeleteRequest::writeConcern(writeConcernObj)
-             << BatchedDeleteRequest::ordered(true) << BatchedDeleteRequest::metadata()
-             << BSON(BatchedRequestMetadata::shardName("shard000")
-                     << BatchedRequestMetadata::shardVersion() << shardVersionArray
-                     << BatchedRequestMetadata::session(0)));
+    BSONObj origDeleteRequestObj = BSON(
+        BatchedDeleteRequest::collName("test")
+        << BatchedDeleteRequest::deletes() << deleteArray
+        << BatchedDeleteRequest::writeConcern(BSON("w" << 1)) << BatchedDeleteRequest::ordered(true)
+        << BatchedDeleteRequest::metadata() << BSON("shardName"
+                                                    << "shard000"
+                                                    << "shardVersion"
+                                                    << BSON_ARRAY(Timestamp(1, 2) << OID::gen())
+                                                    << "ts" << Timestamp(3, 4) << "t" << 5
+                                                    << "session" << 0LL));
 
     string errMsg;
     BatchedDeleteRequest request;
-    bool ok = request.parseBSON("foo", origDeleteRequestObj, &errMsg);
-    ASSERT_TRUE(ok);
+    ASSERT_TRUE(request.parseBSON("foo", origDeleteRequestObj, &errMsg));
 
     ASSERT_EQ("foo.test", request.getNS().ns());
 
-    BSONObj genDeleteRequestObj = request.toBSON();
-    ASSERT_EQUALS(0, genDeleteRequestObj.woCompare(origDeleteRequestObj));
+    ASSERT_EQUALS(origDeleteRequestObj, request.toBSON());
 }
 
 }  // namespace
diff --git a/src/mongo/s/write_ops/batched_insert_request_test.cpp b/src/mongo/s/write_ops/batched_insert_request_test.cpp
index e31503d431f..09e8e950c08 100644
--- a/src/mongo/s/write_ops/batched_insert_request_test.cpp
+++ b/src/mongo/s/write_ops/batched_insert_request_test.cpp
@@ -45,32 +45,24 @@ namespace {
 TEST(BatchedInsertRequest, Basic) {
     BSONArray insertArray = BSON_ARRAY(BSON("a" << 1) << BSON("b" << 1));
 
-    BSONObj writeConcernObj = BSON("w" << 1);
-
-    // The BSON_ARRAY macro doesn't support Timestamps.
-    BSONArrayBuilder arrBuilder;
-    arrBuilder.append(Timestamp(1, 1));
-    arrBuilder.append(OID::gen());
-    BSONArray shardVersionArray = arrBuilder.arr();
-
-    BSONObj origInsertRequestObj =
-        BSON(BatchedInsertRequest::collName("test")
-             << BatchedInsertRequest::documents() << insertArray
-             << BatchedInsertRequest::writeConcern(writeConcernObj)
-             << BatchedInsertRequest::ordered(true) << BatchedInsertRequest::metadata()
-             << BSON(BatchedRequestMetadata::shardName("shard0000")
-                     << BatchedRequestMetadata::shardVersion() << shardVersionArray
-                     << BatchedRequestMetadata::session(0)));
+    BSONObj origInsertRequestObj = BSON(
+        BatchedInsertRequest::collName("test")
+        << BatchedInsertRequest::documents() << insertArray
+        << BatchedInsertRequest::writeConcern(BSON("w" << 1)) << BatchedInsertRequest::ordered(true)
+        << BatchedInsertRequest::metadata() << BSON("shardName"
+                                                    << "shard000"
+                                                    << "shardVersion"
+                                                    << BSON_ARRAY(Timestamp(1, 2) << OID::gen())
+                                                    << "ts" << Timestamp(3, 4) << "t" << 5
+                                                    << "session" << 0LL));
 
     string errMsg;
     BatchedInsertRequest request;
-    bool ok = request.parseBSON("foo", origInsertRequestObj, &errMsg);
-    ASSERT_TRUE(ok);
+    ASSERT_TRUE(request.parseBSON("foo", origInsertRequestObj, &errMsg));
 
     ASSERT_EQ("foo.test", request.getNS().ns());
 
-    BSONObj genInsertRequestObj = request.toBSON();
-    ASSERT_EQUALS(0, genInsertRequestObj.woCompare(origInsertRequestObj));
+    ASSERT_EQUALS(origInsertRequestObj, request.toBSON());
 }
 
 TEST(BatchedInsertRequest, GenIDAll) {
diff --git a/src/mongo/s/write_ops/batched_request_metadata.cpp b/src/mongo/s/write_ops/batched_request_metadata.cpp
index a2e21f99e4c..78ecefbb9df 100644
--- a/src/mongo/s/write_ops/batched_request_metadata.cpp
+++ b/src/mongo/s/write_ops/batched_request_metadata.cpp
@@ -26,6 +26,8 @@
  *    it in the license file.
  */
 
+#include "mongo/platform/basic.h"
+
 #include "mongo/s/write_ops/batched_request_metadata.h"
 
 #include "mongo/db/field_parser.h"
@@ -55,9 +57,8 @@ BSONObj BatchedRequestMetadata::toBSON() const {
     if (_isShardNameSet)
         metadataBuilder << shardName(_shardName);
 
-    if (_shardVersion.get()) {
-        // ChunkVersion wants to be an array.
-        metadataBuilder.append(shardVersion(), static_cast<BSONArray>(_shardVersion->toBSON()));
+    if (_shardVersion) {
+        _shardVersion.get().appendForCommands(&metadataBuilder);
     }
 
     if (_isSessionSet)
@@ -80,12 +81,12 @@ bool BatchedRequestMetadata::parseBSON(const BSONObj& source, string* errMsg) {
     _isShardNameSet = fieldState == FieldParser::FIELD_SET;
 
     {
-        std::unique_ptr<ChunkVersion> tempChunkVersion(new ChunkVersion);
-        fieldState = FieldParser::extract(source, shardVersion, tempChunkVersion.get(), errMsg);
-        if (fieldState == FieldParser::FIELD_INVALID)
+        auto verAndOpTStatus = ChunkVersionAndOpTime::parseFromBSONForCommands(source);
+        if (!verAndOpTStatus.isOK()) {
             return false;
-        if (fieldState == FieldParser::FIELD_SET)
-            _shardVersion.swap(tempChunkVersion);
+        }
+
+        _shardVersion = verAndOpTStatus.getValue();
     }
 
     fieldState = FieldParser::extract(source, session, &_session, errMsg);
@@ -113,10 +114,7 @@ string BatchedRequestMetadata::toString() const {
 void BatchedRequestMetadata::cloneTo(BatchedRequestMetadata* other) const {
     other->_shardName = _shardName;
     other->_isShardNameSet = _isShardNameSet;
-
-    if (other->_shardVersion.get())
-        _shardVersion->cloneTo(other->_shardVersion.get());
-
+    other->_shardVersion = _shardVersion;
     other->_session = _session;
     other->_isSessionSet = _isSessionSet;
 }
@@ -126,36 +124,21 @@ void BatchedRequestMetadata::setShardName(StringData shardName) {
     _isShardNameSet = true;
 }
 
-void BatchedRequestMetadata::unsetShardName() {
-    _isShardNameSet = false;
-}
-
-bool BatchedRequestMetadata::isShardNameSet() const {
-    return _isShardNameSet;
-}
-
 const string& BatchedRequestMetadata::getShardName() const {
     dassert(_isShardNameSet);
     return _shardName;
 }
 
-void BatchedRequestMetadata::setShardVersion(const ChunkVersion& shardVersion) {
-    unique_ptr<ChunkVersion> temp(new ChunkVersion);
-    shardVersion.cloneTo(temp.get());
-    _shardVersion.reset(temp.release());
-}
-
-void BatchedRequestMetadata::unsetShardVersion() {
-    _shardVersion.reset();
+void BatchedRequestMetadata::setShardVersion(const ChunkVersionAndOpTime& shardVersion) {
+    _shardVersion = shardVersion;
 }
 
 bool BatchedRequestMetadata::isShardVersionSet() const {
-    return _shardVersion.get() != NULL;
+    return _shardVersion.is_initialized();
 }
 
 const ChunkVersion& BatchedRequestMetadata::getShardVersion() const {
-    dassert(_shardVersion.get());
-    return *_shardVersion;
+    return _shardVersion.get().getVersion();
 }
 
 void BatchedRequestMetadata::setSession(long long session) {
diff --git a/src/mongo/s/write_ops/batched_request_metadata.h b/src/mongo/s/write_ops/batched_request_metadata.h
index 6e31cb60713..2b885e9018d 100644
--- a/src/mongo/s/write_ops/batched_request_metadata.h
+++ b/src/mongo/s/write_ops/batched_request_metadata.h
@@ -28,34 +28,33 @@
 
 #pragma once
 
+#include <boost/optional.hpp>
 #include <string>
 
-#include "mongo/base/disallow_copying.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/s/bson_serializable.h"
 #include "mongo/s/chunk_version.h"
 
 namespace mongo {
-class BatchedRequestMetadata : public BSONSerializable {
-    MONGO_DISALLOW_COPYING(BatchedRequestMetadata);
 
+class BatchedRequestMetadata {
 public:
     static const BSONField<std::string> shardName;
     static const BSONField<ChunkVersion> shardVersion;
     static const BSONField<long long> session;
 
     BatchedRequestMetadata();
-    virtual ~BatchedRequestMetadata();
+    ~BatchedRequestMetadata();
 
     //
     // bson serializable interface implementation
     //
 
-    virtual bool isValid(std::string* errMsg) const;
-    virtual BSONObj toBSON() const;
-    virtual bool parseBSON(const BSONObj& source, std::string* errMsg);
-    virtual void clear();
-    virtual std::string toString() const;
+    bool isValid(std::string* errMsg) const;
+    BSONObj toBSON() const;
+    bool parseBSON(const BSONObj& source, std::string* errMsg);
+    void clear();
+    std::string toString() const;
 
     void cloneTo(BatchedRequestMetadata* other) const;
 
@@ -64,14 +63,12 @@ public:
     //
 
     void setShardName(StringData shardName);
-    void unsetShardName();
-    bool isShardNameSet() const;
     const std::string& getShardName() const;
 
-    void setShardVersion(const ChunkVersion& shardVersion);
-    void unsetShardVersion();
+    void setShardVersion(const ChunkVersionAndOpTime& shardVersion);
     bool isShardVersionSet() const;
     const ChunkVersion& getShardVersion() const;
+    const repl::OpTime& getOpTime() const;
 
     void setSession(long long session);
     void unsetSession();
@@ -84,10 +81,11 @@ private:
     bool _isShardNameSet;
 
     // (O)  version for this collection on a given shard
-    std::unique_ptr<ChunkVersion> _shardVersion;
+    boost::optional<ChunkVersionAndOpTime> _shardVersion;
 
     // (O)  session number the inserts belong to
     long long _session;
     bool _isSessionSet;
 };
-}
+
+}  // namespace mongo
diff --git a/src/mongo/s/write_ops/batched_request_metadata_test.cpp b/src/mongo/s/write_ops/batched_request_metadata_test.cpp
index f35bcee4ef9..978ee865045 100644
--- a/src/mongo/s/write_ops/batched_request_metadata_test.cpp
+++ b/src/mongo/s/write_ops/batched_request_metadata_test.cpp
@@ -40,22 +40,16 @@ using std::string;
 namespace {
 
 TEST(BatchedRequestMetadata, Basic) {
-    // The BSON_ARRAY macro doesn't support Timestamps.
-    BSONArrayBuilder arrBuilder;
-    arrBuilder.append(Timestamp(1, 1));
-    arrBuilder.append(OID::gen());
-    BSONArray shardVersionArray = arrBuilder.arr();
-
-    BSONObj metadataObj(BSON(BatchedRequestMetadata::shardName("shard0000")
-                             << BatchedRequestMetadata::shardVersion() << shardVersionArray
-                             << BatchedRequestMetadata::session(100)));
+    BSONObj metadataObj(BSON("shardName"
+                             << "shard0000"
+                             << "shardVersion" << BSON_ARRAY(Timestamp(1, 2) << OID::gen()) << "ts"
+                             << Timestamp(3, 4) << "t" << 5 << "session" << 0LL));
 
     string errMsg;
     BatchedRequestMetadata metadata;
     ASSERT_TRUE(metadata.parseBSON(metadataObj, &errMsg));
 
-    BSONObj genMetadataObj = metadata.toBSON();
-    ASSERT_EQUALS(metadataObj, genMetadataObj);
+    ASSERT_EQUALS(metadataObj, metadata.toBSON());
 }
 
 }  // namespace
diff --git a/src/mongo/s/write_ops/batched_update_request_test.cpp b/src/mongo/s/write_ops/batched_update_request_test.cpp
index 04994733537..af7bb5aadfb 100644
--- a/src/mongo/s/write_ops/batched_update_request_test.cpp
+++ b/src/mongo/s/write_ops/batched_update_request_test.cpp
@@ -50,32 +50,24 @@ TEST(BatchedUpdateRequest, Basic) {
                 << BatchedUpdateDocument::updateExpr(BSON("$set" << BSON("b" << 2)))
                 << BatchedUpdateDocument::multi(false) << BatchedUpdateDocument::upsert(false)));
 
-    BSONObj writeConcernObj = BSON("w" << 1);
-
-    // The BSON_ARRAY macro doesn't support Timestamps.
-    BSONArrayBuilder arrBuilder;
-    arrBuilder.append(Timestamp(1, 1));
-    arrBuilder.append(OID::gen());
-    BSONArray shardVersionArray = arrBuilder.arr();
-
-    BSONObj origUpdateRequestObj =
-        BSON(BatchedUpdateRequest::collName("test")
-             << BatchedUpdateRequest::updates() << updateArray
-             << BatchedUpdateRequest::writeConcern(writeConcernObj)
-             << BatchedUpdateRequest::ordered(true) << BatchedUpdateRequest::metadata()
-             << BSON(BatchedRequestMetadata::shardName("shard0000")
-                     << BatchedRequestMetadata::shardVersion() << shardVersionArray
-                     << BatchedRequestMetadata::session(0)));
+    BSONObj origUpdateRequestObj = BSON(
+        BatchedUpdateRequest::collName("test")
+        << BatchedUpdateRequest::updates() << updateArray
+        << BatchedUpdateRequest::writeConcern(BSON("w" << 1)) << BatchedUpdateRequest::ordered(true)
+        << BatchedUpdateRequest::metadata() << BSON("shardName"
+                                                    << "shard000"
+                                                    << "shardVersion"
+                                                    << BSON_ARRAY(Timestamp(1, 2) << OID::gen())
+                                                    << "ts" << Timestamp(3, 4) << "t" << 5
+                                                    << "session" << 0LL));
 
     string errMsg;
     BatchedUpdateRequest request;
-    bool ok = request.parseBSON("foo", origUpdateRequestObj, &errMsg);
-    ASSERT_TRUE(ok);
+    ASSERT_TRUE(request.parseBSON("foo", origUpdateRequestObj, &errMsg));
 
     ASSERT_EQ("foo.test", request.getNS().ns());
 
-    BSONObj genUpdateRequestObj = request.toBSON();
-    ASSERT_EQUALS(0, genUpdateRequestObj.woCompare(origUpdateRequestObj));
+    ASSERT_EQUALS(origUpdateRequestObj, request.toBSON());
 }
 
 }  // namespace
diff --git a/src/mongo/s/write_ops/write_op.cpp b/src/mongo/s/write_ops/write_op.cpp
index 25f9f13b3aa..8cd20fa4865 100644
--- a/src/mongo/s/write_ops/write_op.cpp
+++ b/src/mongo/s/write_ops/write_op.cpp
@@ -127,7 +127,8 @@ Status WriteOp::targetWrites(OperationContext* txn,
         if (endpoints.size() == 1u) {
             targetedWrites->push_back(new TargetedWrite(*endpoint, ref));
         } else {
-            ShardEndpoint broadcastEndpoint(endpoint->shardName, ChunkVersion::IGNORED());
+            ShardEndpoint broadcastEndpoint(endpoint->shardName,
+                                            ChunkVersionAndOpTime(ChunkVersion::IGNORED()));
             targetedWrites->push_back(new TargetedWrite(broadcastEndpoint, ref));
         }
 
diff --git a/src/mongo/s/write_ops/write_op_test.cpp b/src/mongo/s/write_ops/write_op_test.cpp
index e488c068e47..233aa824337 100644
--- a/src/mongo/s/write_ops/write_op_test.cpp
+++ b/src/mongo/s/write_ops/write_op_test.cpp
@@ -218,11 +218,11 @@ TEST(WriteOpTests, TargetMultiAllShards) {
     ASSERT_EQUALS(targeted.size(), 3u);
     sortByEndpoint(&targeted);
     ASSERT_EQUALS(targeted[0]->endpoint.shardName, endpointA.shardName);
-    ASSERT(ChunkVersion::isIgnoredVersion(targeted[0]->endpoint.shardVersion));
+    ASSERT(ChunkVersion::isIgnoredVersion(targeted[0]->endpoint.shardVersion.getVersion()));
     ASSERT_EQUALS(targeted[1]->endpoint.shardName, endpointB.shardName);
-    ASSERT(ChunkVersion::isIgnoredVersion(targeted[1]->endpoint.shardVersion));
+    ASSERT(ChunkVersion::isIgnoredVersion(targeted[1]->endpoint.shardVersion.getVersion()));
     ASSERT_EQUALS(targeted[2]->endpoint.shardName, endpointC.shardName);
-    ASSERT(ChunkVersion::isIgnoredVersion(targeted[2]->endpoint.shardVersion));
+    ASSERT(ChunkVersion::isIgnoredVersion(targeted[2]->endpoint.shardVersion.getVersion()));
 
     writeOp.noteWriteComplete(*targeted[0]);
     writeOp.noteWriteComplete(*targeted[1]);