<s> __author__ = '<STR_LIT>' <EOL> import mongoctl . repository as repository <EOL> from cluster import Cluster <EOL> from server import Server <EOL> from base import DocumentWrapper <EOL> from bson import DBRef <EOL> from mongoctl . mongoctl_logging import log_info <EOL> from mongoctl . utils import document_pretty_string <EOL> import time <EOL> class ShardedCluster ( Cluster ) : <EOL> def __init__ ( self , cluster_document ) : <EOL> Cluster . __init__ ( self , cluster_document ) <EOL> self . _config_members = self . _resolve_members ( "<STR_LIT>" ) <EOL> self . _shards = self . _resolve_shard_members ( ) <EOL> def _resolve_shard_members ( self ) : <EOL> member_documents = self . get_property ( "<STR_LIT>" ) <EOL> members = [ ] <EOL> if member_documents : <EOL> for mem_doc in member_documents : <EOL> member = ShardMember ( mem_doc ) <EOL> members . append ( member ) <EOL> return members <EOL> @ property <EOL> def config_members ( self ) : <EOL> return self . _config_members <EOL> def has_config_server ( self , server ) : <EOL> for member in self . config_members : <EOL> if member . get_server ( ) . id == server . id : <EOL> return True <EOL> @ property <EOL> def shards ( self ) : <EOL> return self . _shards <EOL> def has_shard ( self , shard ) : <EOL> return self . get_shard_member ( shard ) is not None <EOL> def get_shard_member ( self , shard ) : <EOL> for shard_member in self . shards : <EOL> if ( ( isinstance ( shard , Server ) and <EOL> shard_member . get_server ( ) and <EOL> shard_member . get_server ( ) . id == shard . id ) <EOL> or <EOL> ( isinstance ( shard , Cluster ) and <EOL> shard_member . get_cluster ( ) and <EOL> shard_member . get_cluster ( ) . id == shard . id ) ) : <EOL> return shard_member <EOL> def get_shard_member_by_shard_id ( self , shard_id ) : <EOL> for shard_member in self . shards : <EOL> if ( ( shard_member . get_server ( ) and <EOL> shard_member . get_server ( ) . id == shard_id ) <EOL> or ( shard_member . get_cluster ( ) and <EOL> shard_member . get_cluster ( ) . id == shard_id ) ) : <EOL> return shard_member <EOL> def get_config_member_addresses ( self ) : <EOL> addresses = [ ] <EOL> for member in self . config_members : <EOL> addresses . append ( member . get_server ( ) . get_address ( ) ) <EOL> return addresses <EOL> def get_member_addresses ( self ) : <EOL> addresses = [ ] <EOL> for member in self . config_members : <EOL> addresses . append ( member . get_server ( ) . get_address ( ) ) <EOL> return addresses <EOL> def get_shard_member_address ( self , shard_member ) : <EOL> if shard_member . get_server ( ) : <EOL> return shard_member . get_server ( ) . get_address ( ) <EOL> elif shard_member . get_cluster ( ) : <EOL> cluster_member_addresses = [ ] <EOL> for cluster_member in shard_member . get_cluster ( ) . get_members ( ) : <EOL> cluster_member_addresses . append ( <EOL> cluster_member . get_server ( ) . get_address ( ) ) <EOL> return "<STR_LIT>" % ( shard_member . get_cluster ( ) . id , <EOL> "<STR_LIT:U+002C>" . join ( cluster_member_addresses ) ) <EOL> def configure_sharded_cluster ( self ) : <EOL> sh_list = self . list_shards ( ) <EOL> if sh_list and sh_list . get ( "<STR_LIT>" ) : <EOL> log_info ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> for shard_member in self . shards : <EOL> self . add_shard ( shard_member . get_shard ( ) ) <EOL> def add_shard ( self , shard ) : <EOL> log_info ( "<STR_LIT>" % ( shard . id , self . id ) ) <EOL> if self . is_shard_configured ( shard ) : <EOL> log_info ( "<STR_LIT>" % shard . id ) <EOL> return <EOL> mongos = self . get_any_online_mongos ( ) <EOL> shard_member = self . get_shard_member ( shard ) <EOL> cmd = self . get_add_shard_command ( shard_member ) <EOL> configured_shards = self . list_shards ( ) <EOL> log_info ( "<STR_LIT>" % <EOL> document_pretty_string ( configured_shards ) ) <EOL> log_info ( "<STR_LIT>" % <EOL> ( document_pretty_string ( cmd ) , mongos . id ) ) <EOL> mongos . db_command ( cmd , "<STR_LIT>" ) <EOL> log_info ( "<STR_LIT>" % self . id ) <EOL> def get_add_shard_command ( self , shard_member ) : <EOL> return { <EOL> "<STR_LIT>" : self . get_shard_member_address ( shard_member ) <EOL> } <EOL> def remove_shard ( self , shard , unsharded_data_dest_id = None , <EOL> synchronized = False ) : <EOL> log_info ( "<STR_LIT>" % <EOL> ( shard . id , self . id ) ) <EOL> configured_shards = self . list_shards ( ) <EOL> log_info ( "<STR_LIT>" % <EOL> document_pretty_string ( configured_shards ) ) <EOL> completed = False <EOL> while not completed : <EOL> result = self . _do_remove_shard ( shard , unsharded_data_dest_id ) <EOL> completed = synchronized and ( result [ "<STR_LIT:state>" ] == "<STR_LIT>" or <EOL> not self . is_shard_configured ( shard ) ) <EOL> if not completed : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> def _do_remove_shard ( self , shard , unsharded_data_dest_id = None ) : <EOL> cmd = self . get_validate_remove_shard_command ( shard ) <EOL> mongos = self . get_any_online_mongos ( ) <EOL> log_info ( "<STR_LIT>" % <EOL> ( document_pretty_string ( cmd ) , mongos . id ) ) <EOL> result = mongos . db_command ( cmd , "<STR_LIT>" ) <EOL> log_info ( "<STR_LIT>" % result ) <EOL> if "<STR_LIT>" in result and unsharded_data_dest_id : <EOL> dest_shard_member = self . get_shard_member_by_shard_id ( <EOL> unsharded_data_dest_id ) <EOL> if not dest_shard_member : <EOL> raise Exception ( "<STR_LIT>" % <EOL> ( unsharded_data_dest_id , self . id ) ) <EOL> dest_shard = dest_shard_member . get_shard ( ) <EOL> self . move_dbs_primary ( result [ "<STR_LIT>" ] , dest_shard ) <EOL> if result . get ( '<STR_LIT:state>' ) == "<STR_LIT>" : <EOL> log_info ( "<STR_LIT>" % self . id ) <EOL> return result <EOL> def get_validate_remove_shard_command ( self , shard ) : <EOL> if not self . is_shard_configured ( shard ) : <EOL> raise Exception ( "<STR_LIT>" <EOL> "<STR_LIT>" % shard . id ) <EOL> shard_member = self . get_shard_member ( shard ) <EOL> return { <EOL> "<STR_LIT>" : shard_member . get_shard_id ( ) <EOL> } <EOL> def get_remove_shard_command ( self , shard ) : <EOL> return { <EOL> "<STR_LIT>" : shard . id <EOL> } <EOL> def list_shards ( self ) : <EOL> mongos = self . get_any_online_mongos ( ) <EOL> return mongos . db_command ( { "<STR_LIT>" : <NUM_LIT:1> } , "<STR_LIT>" ) <EOL> def is_shard_configured ( self , shard ) : <EOL> shard_list = self . list_shards ( ) <EOL> if shard_list and shard_list . get ( "<STR_LIT>" ) : <EOL> for sh in shard_list [ "<STR_LIT>" ] : <EOL> if shard . id == sh [ "<STR_LIT>" ] : <EOL> return True <EOL> def is_last_shard ( self , shard ) : <EOL> pass <EOL> def get_default_server ( self ) : <EOL> return self . get_any_online_mongos ( ) <EOL> def get_any_online_mongos ( self ) : <EOL> for member in self . get_members ( ) : <EOL> if member . get_server ( ) . is_online ( ) : <EOL> return member . get_server ( ) <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def move_dbs_primary ( self , db_names , dest_shard ) : <EOL> log_info ( "<STR_LIT>" % <EOL> ( db_names , dest_shard . id ) ) <EOL> mongos = self . get_any_online_mongos ( ) <EOL> for db_name in db_names : <EOL> move_cmd = { <EOL> "<STR_LIT>" : db_name , <EOL> "<STR_LIT:to>" : dest_shard . id <EOL> } <EOL> log_info ( "<STR_LIT>" % <EOL> ( document_pretty_string ( move_cmd ) , mongos . id ) ) <EOL> result = mongos . db_command ( move_cmd , "<STR_LIT>" ) <EOL> log_info ( "<STR_LIT>" % document_pretty_string ( result ) ) <EOL> def get_member_type ( self ) : <EOL> return ShardMember <EOL> class ShardMember ( DocumentWrapper ) : <EOL> def __init__ ( self , member_doc ) : <EOL> DocumentWrapper . __init__ ( self , member_doc ) <EOL> self . _server = None <EOL> self . _cluster = None <EOL> def get_server ( self ) : <EOL> server_doc = self . get_property ( "<STR_LIT>" ) <EOL> if not server_doc : <EOL> return <EOL> if self . _server is None : <EOL> if server_doc is not None : <EOL> if type ( server_doc ) is DBRef : <EOL> self . _server = repository . lookup_server ( server_doc . id ) <EOL> return self . _server <EOL> def get_cluster ( self ) : <EOL> cluster_doc = self . get_property ( "<STR_LIT>" ) <EOL> if not cluster_doc : <EOL> return <EOL> if self . _cluster is None : <EOL> if cluster_doc is not None : <EOL> if type ( cluster_doc ) is DBRef : <EOL> self . _cluster = repository . lookup_cluster ( cluster_doc . id ) <EOL> return self . _cluster <EOL> def get_shard_id ( self ) : <EOL> return self . get_shard ( ) . id <EOL> def get_shard ( self ) : <EOL> if self . get_server ( ) : <EOL> return self . get_server ( ) <EOL> elif self . get_cluster ( ) : <EOL> return self . get_cluster ( ) </s>
