276 lines
9.9 KiB
C++
276 lines
9.9 KiB
C++
// @file shard_version.cpp
|
|
|
|
/**
|
|
* Copyright (C) 2010 10gen Inc.
|
|
*
|
|
* This program is free software: you can redistribute it and/or modify
|
|
* it under the terms of the GNU Affero General Public License, version 3,
|
|
* as published by the Free Software Foundation.
|
|
*
|
|
* This program is distributed in the hope that it will be useful,
|
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
* GNU Affero General Public License for more details.
|
|
*
|
|
* You should have received a copy of the GNU Affero General Public License
|
|
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
*/
|
|
|
|
#include "pch.h"
|
|
|
|
#include "chunk.h"
|
|
#include "config.h"
|
|
#include "grid.h"
|
|
#include "util.h"
|
|
#include "shard.h"
|
|
#include "writeback_listener.h"
|
|
|
|
#include "shard_version.h"
|
|
|
|
namespace mongo {
|
|
|
|
// when running in sharded mode, use chunk shard version control
|
|
|
|
static bool isVersionable( DBClientBase * conn );
|
|
static bool initShardVersion( DBClientBase & conn, BSONObj& result );
|
|
static bool checkShardVersion( DBClientBase & conn , const string& ns , bool authoritative = false , int tryNumber = 1 );
|
|
static void resetShardVersion( DBClientBase * conn );
|
|
|
|
void installChunkShardVersioning() {
|
|
//
|
|
// Overriding no-op behavior in shardconnection.cpp
|
|
//
|
|
// TODO: Better encapsulate this mechanism.
|
|
//
|
|
isVersionableCB = isVersionable;
|
|
initShardVersionCB = initShardVersion;
|
|
checkShardVersionCB = checkShardVersion;
|
|
resetShardVersionCB = resetShardVersion;
|
|
}
|
|
|
|
struct ConnectionShardStatus {
|
|
|
|
typedef unsigned long long S;
|
|
|
|
ConnectionShardStatus()
|
|
: _mutex( "ConnectionShardStatus" ) {
|
|
}
|
|
|
|
bool isInitialized( DBClientBase * conn ){
|
|
scoped_lock lk( _mutex );
|
|
return _init.find( conn ) != _init.end();
|
|
}
|
|
|
|
void setInitialized( DBClientBase * conn ){
|
|
// At this point, conn may be deleted, *do not access*
|
|
scoped_lock lk( _mutex );
|
|
_init.insert( conn );
|
|
}
|
|
|
|
S getSequence( DBClientBase * conn , const string& ns ) {
|
|
scoped_lock lk( _mutex );
|
|
return _map[conn][ns];
|
|
}
|
|
|
|
void setSequence( DBClientBase * conn , const string& ns , const S& s ) {
|
|
scoped_lock lk( _mutex );
|
|
_map[conn][ns] = s;
|
|
}
|
|
|
|
void reset( DBClientBase * conn ) {
|
|
scoped_lock lk( _mutex );
|
|
_map.erase( conn );
|
|
_init.erase( conn );
|
|
}
|
|
|
|
// protects _maps
|
|
mongo::mutex _mutex;
|
|
|
|
// a map from a connection into ChunkManager's sequence number for each namespace
|
|
map<DBClientBase*, map<string,unsigned long long> > _map;
|
|
set<DBClientBase*> _init;
|
|
|
|
} connectionShardStatus;
|
|
|
|
void resetShardVersion( DBClientBase * conn ) {
|
|
connectionShardStatus.reset( conn );
|
|
}
|
|
|
|
bool isVersionable( DBClientBase* conn ){
|
|
return conn->type() == ConnectionString::MASTER || conn->type() == ConnectionString::SET;
|
|
}
|
|
|
|
DBClientBase* getVersionable( DBClientBase* conn ){
|
|
|
|
switch ( conn->type() ) {
|
|
case ConnectionString::INVALID:
|
|
massert( 15904, str::stream() << "cannot set version on invalid connection " << conn->toString(), false );
|
|
return NULL;
|
|
case ConnectionString::MASTER:
|
|
return conn;
|
|
case ConnectionString::PAIR:
|
|
massert( 15905, str::stream() << "cannot set version or shard on pair connection " << conn->toString(), false );
|
|
return NULL;
|
|
case ConnectionString::SYNC:
|
|
massert( 15906, str::stream() << "cannot set version or shard on sync connection " << conn->toString(), false );
|
|
return NULL;
|
|
case ConnectionString::SET:
|
|
DBClientReplicaSet* set = (DBClientReplicaSet*) conn;
|
|
return &( set->masterConn() );
|
|
}
|
|
|
|
assert( false );
|
|
return NULL;
|
|
}
|
|
|
|
extern OID serverID;
|
|
|
|
bool initShardVersion( DBClientBase& conn_in, BSONObj& result ){
|
|
|
|
WriteBackListener::init( conn_in );
|
|
|
|
DBClientBase* conn = getVersionable( &conn_in );
|
|
assert( conn ); // errors thrown above
|
|
|
|
BSONObjBuilder cmdBuilder;
|
|
|
|
cmdBuilder.append( "setShardVersion" , "" );
|
|
cmdBuilder.appendBool( "init", true );
|
|
cmdBuilder.append( "configdb" , configServer.modelServer() );
|
|
cmdBuilder.appendOID( "serverID" , &serverID );
|
|
cmdBuilder.appendBool( "authoritative" , true );
|
|
|
|
BSONObj cmd = cmdBuilder.obj();
|
|
|
|
LOG(1) << "initializing shard connection to " << conn->toString() << endl;
|
|
LOG(2) << "initial sharding settings : " << cmd << endl;
|
|
|
|
bool ok = conn->runCommand( "admin" , cmd , result );
|
|
|
|
// Conn may be deleted here - *do not access again* - css is an exception, since just uses ptr address
|
|
connectionShardStatus.setInitialized( conn );
|
|
conn = NULL;
|
|
|
|
// HACK for backwards compatibility with v1.8.x, v2.0.0 and v2.0.1
|
|
// Result is false, but will still initialize serverID and configdb
|
|
// Not master does not initialize serverID and configdb, but we ignore since if the connection is not master,
|
|
// we are not setting the shard version at all
|
|
if( ! ok && ! result["errmsg"].eoo() && ( result["errmsg"].String() == "need to specify namespace"/* 2.0.1/2 */ ||
|
|
result["errmsg"].String() == "need to speciy namespace" /* 1.8 */ ||
|
|
result["errmsg"].String() == "not master" /* both */ ) )
|
|
{
|
|
ok = true;
|
|
}
|
|
|
|
LOG(3) << "initial sharding result : " << result << endl;
|
|
|
|
return ok;
|
|
|
|
}
|
|
|
|
/**
|
|
* @return true if had to do something
|
|
*/
|
|
bool checkShardVersion( DBClientBase& conn_in , const string& ns , bool authoritative , int tryNumber ) {
|
|
// TODO: cache, optimize, etc...
|
|
|
|
WriteBackListener::init( conn_in );
|
|
|
|
DBConfigPtr conf = grid.getDBConfig( ns );
|
|
if ( ! conf )
|
|
return false;
|
|
|
|
DBClientBase* conn = getVersionable( &conn_in );
|
|
assert(conn); // errors thrown above
|
|
|
|
if( ! connectionShardStatus.isInitialized( conn ) ){
|
|
BSONObj result;
|
|
uassert( 15918, str::stream() << "cannot initialize version on shard " << conn->getServerAddress() << causedBy( result.toString() ), initShardVersion( *conn, result ) );
|
|
}
|
|
|
|
unsigned long long officialSequenceNumber = 0;
|
|
|
|
ChunkManagerPtr manager;
|
|
const bool isSharded = conf->isSharded( ns );
|
|
if ( isSharded ) {
|
|
manager = conf->getChunkManagerIfExists( ns , authoritative );
|
|
// It's possible the chunk manager was reset since we checked whether sharded was true,
|
|
// so must check this here.
|
|
if( manager ) officialSequenceNumber = manager->getSequenceNumber();
|
|
}
|
|
|
|
// has the ChunkManager been reloaded since the last time we updated the connection-level version?
|
|
// (ie., last time we issued the setShardVersions below)
|
|
unsigned long long sequenceNumber = connectionShardStatus.getSequence(conn,ns);
|
|
if ( sequenceNumber == officialSequenceNumber ) {
|
|
return false;
|
|
}
|
|
|
|
|
|
ShardChunkVersion version = 0;
|
|
if ( isSharded && manager ) {
|
|
version = manager->getVersion( Shard::make( conn->getServerAddress() ) );
|
|
}
|
|
|
|
if( version == 0 ){
|
|
LOG(2) << "resetting shard version of " << ns << " on " << conn->getServerAddress() << ", " <<
|
|
( ! isSharded ? "no longer sharded" :
|
|
( ! manager ? "no chunk manager found" :
|
|
"version is zero" ) ) << endl;
|
|
}
|
|
|
|
LOG(2) << " have to set shard version for conn: " << conn << " ns:" << ns
|
|
<< " my last seq: " << sequenceNumber << " current: " << officialSequenceNumber
|
|
<< " version: " << version << " manager: " << manager.get()
|
|
<< endl;
|
|
|
|
BSONObj result;
|
|
// Save the server address, cannot access if fails
|
|
string serverAddress = conn->getServerAddress();
|
|
if ( setShardVersion( *conn , ns , version , authoritative , result ) ) {
|
|
// success!
|
|
LOG(1) << " setShardVersion success: " << result << endl;
|
|
connectionShardStatus.setSequence( conn , ns , officialSequenceNumber );
|
|
return true;
|
|
}
|
|
|
|
// At this point, it is no longer safe to use the pointer to conn, we do not know its state
|
|
conn = NULL;
|
|
|
|
LOG(1) << " setShardVersion failed!\n" << result << endl;
|
|
|
|
if ( result["need_authoritative"].trueValue() )
|
|
massert( 10428 , "need_authoritative set but in authoritative mode already" , ! authoritative );
|
|
|
|
if ( ! authoritative ) {
|
|
checkShardVersion( conn_in , ns , 1 , tryNumber + 1 );
|
|
return true;
|
|
}
|
|
|
|
if ( result["reloadConfig"].trueValue() ) {
|
|
if( result["version"].timestampTime() == 0 ){
|
|
// reload db
|
|
conf->reload();
|
|
}
|
|
else {
|
|
// reload config
|
|
conf->getChunkManager( ns , true );
|
|
}
|
|
}
|
|
|
|
const int maxNumTries = 7;
|
|
if ( tryNumber < maxNumTries ) {
|
|
LOG( tryNumber < ( maxNumTries / 2 ) ? 1 : 0 )
|
|
<< "going to retry checkShardVersion host: " << serverAddress << " " << result << endl;
|
|
sleepmillis( 10 * tryNumber );
|
|
checkShardVersion( conn_in , ns , true , tryNumber + 1 );
|
|
return true;
|
|
}
|
|
|
|
string errmsg = str::stream() << "setShardVersion failed host: " << serverAddress << " " << result;
|
|
log() << " " << errmsg << endl;
|
|
massert( 10429 , errmsg , 0 );
|
|
return true;
|
|
}
|
|
|
|
} // namespace mongo
|