MongoDB 系统时钟跳变引发的风波

{"ts" : Timestamp(1560861342, 2),"t" : NumberLong(12),"h" : NumberLong("7983167552279045735"),"v" : 2,"op" : "d","ns" : "app.T_AppInfo","o" : {"_id" : ObjectId("5d08da9ebe3cb8c01ea48a25")        }}

db.currentOp({"ns" : "local.oplog.rs"})
{"desc" : "conn611866","client" : "192.168.138.77:51842","clientMetadata" : {"driver" : {"name" : "NetworkInterfaceASIO-RS","version" : "3.4.10"            }    },"active" : true,"opid" : 20648187,"secs_running" : 0,"microsecs_running" : NumberLong(519601),"op" : "getmore","ns" : "local.oplog.rs","query" : {"getMore" : NumberLong("16712800432"),"collection" : "oplog.rs","maxTimeMS" : NumberLong(5000),"term" : NumberLong(2),"lastKnownCommittedOpTime" : {"ts" : Timestamp(1560842637, 2),"t" : NumberLong(2)            }    },"originatingCommand" : {"find" : "oplog.rs","filter" : {"ts" : {"$gte" : Timestamp(1560406790, 2)                    }            },"tailable" : true,"oplogReplay" : true,"awaitData" : true,"maxTimeMS" : NumberLong(60000),"term" : NumberLong(2),"readConcern" : {"afterOpTime" : {"ts" : Timestamp(1560406790, 2),"t" : NumberLong(1)                    }            }    },"planSummary" : "COLLSCAN",}

shard0:PRIMARY> db.test.insert({"justForTest": true})shard0:PRIMARY> db.getSiblingDB("local").oplog.rs.find({ns: "test.test"}).sort({$natural: -1}).limit(1).pretty(){"ts" : Timestamp(1560842490, 2),"t" : NumberLong(2),"h" : NumberLong("-1966048951433407860"),"v" : 2,"op" : "i","ns" : "test.test","o" : {"_id" : ObjectId("5d088723b0a0777f7326df57"),"justForTest" : true        }}

shard0:PRIMARY> new Date(1560842490*1000)ISODate("2019-06-18T07:21:30Z")

# date '+%Y-%m-%d %H:%M:%S'2019-06-18 07:21:26

# date -s 09:00:00Tue Jun 18 09:00:00 UTC 2019

shard0:PRIMARY> db.test.insert({"justForTest": true})shard0:PRIMARY> db.getSiblingDB("local").oplog.rs.find({ns: "test.test"}).sort({$natural: -1}).limit(1).pretty(){"ts" : Timestamp(1560848723, 1),"t" : NumberLong(4),"h" : NumberLong("-6994951573635880200"),"v" : 2,"op" : "i","ns" : "test.test","o" : {"_id" : ObjectId("5d08a953b9963dbc8476d6b7"),"justForTest" : true        }}
shard0:PRIMARY> new Date(1560848723*1000)ISODate("2019-06-18T09:05:23Z")

host-192-168-138-148:~ # date -s 07:00:00Tue Jun 18 07:00:00 UTC 2019

shard0:PRIMARY> db.test.insert({"justForTest": true})shard0:PRIMARY> db.getSiblingDB("local").oplog.rs.find({ns: "test.test"}).sort({$natural: -1}).limit(1).pretty(){"ts" : Timestamp(1560848864, 92),"t" : NumberLong(4),"h" : NumberLong("3290816976088149103"),"v" : 2,"op" : "i","ns" : "test.test","o" : {"_id" : ObjectId("5d088c1eb9963dbc8476d6b8"),"justForTest" : true        }}shard0:PRIMARY> new Date(1560848864*1000)ISODate("2019-06-18T09:07:44Z")

//获取下一个时间戳Timestamp getNextGlobalTimestamp(unsigned count) {//系统时间值const unsigned now = durationCount( getGlobalServiceContext()->getFastClockSource()->now().toDurationSinceEpoch());    ...// 对当前上下文的Timestamp 自增计数auto first = globalTimestamp.fetchAndAdd(count);auto currentTimestamp = first + count;  // What we just set it to.unsigned globalSecs = Timestamp(currentTimestamp).getSecs();// 若上下文时间大于系统时间，且同一时刻的计数器 超过2^31-1(2147483647)时，进行报错if (MONGO_unlikely(globalSecs > now) && Timestamp(currentTimestamp).getInc() >= 1U << 31) {        mongo::severe() << "clock skew detected, prev: " << globalSecs << " now: " << now;        fassertFailed(17449);    }

# dateTue Jun 18 09:00:12 UTC 2019# date -s 08:00:00Tue Jun 18 08:00:00 UTC 2019

shard0:SECONDARY> db.isMaster(){"hosts" : ["192.168.138.77:30071","192.168.138.148:30071","192.168.138.55:30071"        ],"setName" : "shard0","setVersion" : 1,"ismaster" : false,"secondary" : true,"primary" : "192.168.138.148:30071","me" : "192.168.138.55:30071",        ..."readOnly" : false,"ok" : 1}
=== 没有发生变化，仍然是备节点
shard0:SECONDARY>

# dateTue Jun 18 08:02:45 UTC 2019# date -s 09:00:00Tue Jun 18 09:00:00 UTC 2019

shard0:SECONDARY> db.isMaster(){"hosts" : ["192.168.138.77:30071","192.168.138.148:30071","192.168.138.55:30071"        ],"setName" : "shard0","setVersion" : 1,"ismaster" : true,"secondary" : false,"primary" : "192.168.138.55:30071","me" : "192.168.138.55:30071","electionId" : ObjectId("7fffffff0000000000000008"),        ..."readOnly" : false,"ok" : 1}
=== 发生变化，切换为主节点
shard0:PRIMARY>

I REPL     [ReplicationExecutor] Starting an election, since we've seen no PRIMARY in the past 10000msI REPL     [ReplicationExecutor] conducting a dry run election to see if we could be electedI REPL     [ReplicationExecutor] VoteRequester(term 7 dry run) received a yes vote from 192.168.138.77:30071; response message: { term: 7, voteGranted: true, reason: "", ok: 1.0 }I REPL     [ReplicationExecutor] dry election run succeeded, running for electionI REPL     [ReplicationExecutor] VoteRequester(term 8) received a yes vote from 192.168.138.77:30071; response message: { term: 8, voteGranted: true, reason: "", ok: 1.0 }I REPL     [ReplicationExecutor] election succeeded, assuming primary role in term 8I REPL     [ReplicationExecutor] transition to PRIMARYI REPL     [ReplicationExecutor] Entering primary catch-up mode.I REPL     [ReplicationExecutor] Caught up to the latest optime known via heartbeats after becoming primary.I REPL     [ReplicationExecutor] Exited primary catch-up mode.I REPL     [rsBackgroundSync] Replication producer stopped after oplog fetcher finished returning a batch from our sync source.  Abandoning this batch of oplog entries and re-evaluating our sync source.I REPL     [SyncSourceFeedback] SyncSourceFeedback error sending update to 192.168.138.148:30071: InvalidSyncSource: Sync source was cleared. Was 192.168.138.148:30071I REPL     [rsSync] transition to primary complete; database writes are now permittedI REPL     [ReplicationExecutor] Member 192.168.138.148:30071 is now in state SECONDARY

void ReplicationCoordinatorImpl::_cancelAndRescheduleElectionTimeout_inlock() {//如果上一个定时器回调存在，则直接取消if (_handleElectionTimeoutCbh.isValid()) {        _replExecutor.cancel(_handleElectionTimeoutCbh);        ..    }    ...//触发调度，when时间点为 now + electionTimeout + randomOffset//到了时间就执行_startElectSelfIfEligibleV1函数，发起选举    _handleElectionTimeoutCbh =        _scheduleWorkAt(when,             stdx::bind(&ReplicationCoordinatorImpl::_startElectSelfIfEligibleV1,this,                  StartElectionV1Reason::kElectionTimeout));}

StatusWith ReplicationExecutor::scheduleWorkAt(    Date_t when, const CallbackFn& work) {    stdx::lock_guard lk(_mutex);    WorkQueue temp;    StatusWith cbHandle = enqueueWork_inlock(&temp, work);    ...
    WorkQueue::iterator insertBefore = _sleepersQueue.begin();//根据调度时间找到插入位置while (insertBefore != _sleepersQueue.end() && insertBefore->readyDate <= when)        ++insertBefore;//将任务置入_sleepersQueue队列    _sleepersQueue.splice(insertBefore, temp, temp.begin());    ...return cbHandle;}

//运行线程 -- 持续获取队列任务void ReplicationExecutor::run() {    ...//循环获取任务执行while ((work = getWork()).first.callback.isValid()) {//发起任务..    }}
//获取可用的任务ReplicationExecutor::getWork() {    stdx::unique_lock lk(_mutex);while (true) {//取当前时间const Date_t now = _networkInterface->now();//将_sleepersQueue队列中到时间的任务置入_readyQueue队列(唤醒)        Date_t nextWakeupDate = scheduleReadySleepers_inlock(now);//存在任务执行，跳出循环if (!_readyQueue.empty()) {break;        } else if (_inShutdown) {return std::make_pair(WorkItem(), CallbackHandle());        }        lk.unlock();//没有合适的任务，继续等待if (nextWakeupDate == Date_t::max()) {            _networkInterface->waitForWork();        } else {            _networkInterface->waitForWorkUntil(nextWakeupDate);        }        lk.lock();    }//返回待执行任务const WorkItem work = *_readyQueue.begin();return std::make_pair(work, cbHandle);}//将到时间的任务唤醒，写入_readyQueue队列Date_t ReplicationExecutor::scheduleReadySleepers_inlock(const Date_t now) {    WorkQueue::iterator iter = _sleepersQueue.begin();//从头部开始，找到最后一个调度时间小于等于当前时间(需要执行)的任务while ((iter != _sleepersQueue.end()) && (iter->readyDate callback);        callback->_isSleeper = false;        ++iter;    }//转移队列    _readyQueue.splice(_readyQueue.end(), _sleepersQueue, _sleepersQueue.begin(), iter);if (iter == _sleepersQueue.end()) {// indicate no sleeper to wait forreturn Date_t::max();    }return iter->readyDate;}