#NEWLINE# def _replicate_object( self , partition , object_file , node_id ) : #NEWLINE# #INDENT# '\n        Replicate the db, choosing method based on whether or not it\n        already exists on peers.\n\n        :param partition: partition to be replicated to\n        :param object_file: DB file name to be replicated\n        :param node_id: node id of the node to be replicated to\n        ' #NEWLINE# start_time = now = time . time ( ) #NEWLINE# self . logger . debug ( 'Replicating db %s' , object_file ) #NEWLINE# self . stats [ 'attempted' ] += 1 #NEWLINE# self . logger . increment ( 'attempts' ) #NEWLINE# shouldbehere = True #NEWLINE# try : #NEWLINE# #INDENT# broker = self . brokerclass ( object_file , pending_timeout = 30 ) #NEWLINE# broker . reclaim ( ( now - self . reclaim_age ) , ( now - ( self . reclaim_age * 2 ) ) ) #NEWLINE# info = broker . get_replication_info ( ) #NEWLINE# bpart = self . ring . get_part ( info [ 'account' ] , info . get ( 'container' ) ) #NEWLINE# if ( bpart != int ( partition ) ) : #NEWLINE# #INDENT# partition = bpart #NEWLINE# shouldbehere = False #NEWLINE# name = ( '/' + quote ( info [ 'account' ] ) ) #NEWLINE# if ( 'container' in info ) : #NEWLINE# #INDENT# name += ( '/' + quote ( info [ 'container' ] ) ) #NEWLINE# #UNINDENT# self . logger . error ( ( 'Found %s for %s when it should be on partition %s; will replicate out and remove.' % ( object_file , name , bpart ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# except ( Exception , Timeout ) as e : #NEWLINE# #INDENT# if ( 'no such table' in str ( e ) ) : #NEWLINE# #INDENT# self . logger . error ( _ ( 'Quarantining DB %s' ) , object_file ) #NEWLINE# quarantine_db ( broker . db_file , broker . db_type ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# self . logger . exception ( _ ( 'ERROR reading db %s' ) , object_file ) #NEWLINE# #UNINDENT# nodes = self . ring . get_part_nodes ( int ( partition ) ) #NEWLINE# self . _add_failure_stats ( [ ( failure_dev [ 'replication_ip' ] , failure_dev [ 'device' ] ) for failure_dev in nodes ] ) #NEWLINE# self . logger . increment ( 'failures' ) #NEWLINE# return #NEWLINE# #UNINDENT# delete_timestamp = Timestamp ( ( info . get ( 'delete_timestamp' ) or 0 ) ) #NEWLINE# put_timestamp = Timestamp ( ( info . get ( 'put_timestamp' ) or 0 ) ) #NEWLINE# if ( ( delete_timestamp < ( now - self . reclaim_age ) ) and ( delete_timestamp > put_timestamp ) and ( info [ 'count' ] in ( None , '' , 0 , '0' ) ) ) : #NEWLINE# #INDENT# if self . report_up_to_date ( info ) : #NEWLINE# #INDENT# self . delete_db ( broker ) #NEWLINE# #UNINDENT# self . logger . timing_since ( 'timing' , start_time ) #NEWLINE# return #NEWLINE# #UNINDENT# responses = [ ] #NEWLINE# failure_devs_info = set ( ) #NEWLINE# nodes = self . ring . get_part_nodes ( int ( partition ) ) #NEWLINE# local_dev = None #NEWLINE# for node in nodes : #NEWLINE# #INDENT# if ( node [ 'id' ] == node_id ) : #NEWLINE# #INDENT# local_dev = node #NEWLINE# break #NEWLINE# #UNINDENT# #UNINDENT# if shouldbehere : #NEWLINE# #INDENT# shouldbehere = bool ( [ n for n in nodes if ( n [ 'id' ] == node_id ) ] ) #NEWLINE# #UNINDENT# i = 0 #NEWLINE# while ( ( i < len ( nodes ) ) and ( nodes [ i ] [ 'id' ] != node_id ) ) : #NEWLINE# #INDENT# i += 1 #NEWLINE# #UNINDENT# repl_nodes = ( nodes [ ( i + 1 ) : ] + nodes [ : i ] ) #NEWLINE# more_nodes = self . ring . get_more_nodes ( int ( partition ) ) #NEWLINE# if ( not local_dev ) : #NEWLINE# #INDENT# for node in more_nodes : #NEWLINE# #INDENT# if ( node [ 'id' ] == node_id ) : #NEWLINE# #INDENT# local_dev = node #NEWLINE# break #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# for node in repl_nodes : #NEWLINE# #INDENT# different_region = False #NEWLINE# if ( local_dev and ( local_dev [ 'region' ] != node [ 'region' ] ) ) : #NEWLINE# #INDENT# different_region = True #NEWLINE# #UNINDENT# success = False #NEWLINE# try : #NEWLINE# #INDENT# success = self . _repl_to_node ( node , broker , partition , info , different_region ) #NEWLINE# #UNINDENT# except DriveNotMounted : #NEWLINE# #INDENT# repl_nodes . append ( next ( more_nodes ) ) #NEWLINE# self . logger . error ( _ ( 'ERROR Remote drive not mounted %s' ) , node ) #NEWLINE# #UNINDENT# except ( Exception , Timeout ) : #NEWLINE# #INDENT# self . logger . exception ( _ ( 'ERROR syncing %(file)s with node %(node)s' ) , { #NEWLINE# #INDENT# 'file' : object_file , #NEWLINE# 'node' : node , #NEWLINE# #UNINDENT# } ) #NEWLINE# #UNINDENT# if ( not success ) : #NEWLINE# #INDENT# failure_devs_info . add ( ( node [ 'replication_ip' ] , node [ 'device' ] ) ) #NEWLINE# #UNINDENT# self . logger . increment ( ( 'successes' if success else 'failures' ) ) #NEWLINE# responses . append ( success ) #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# self . _post_replicate_hook ( broker , info , responses ) #NEWLINE# #UNINDENT# except ( Exception , Timeout ) : #NEWLINE# #INDENT# self . logger . exception ( 'UNHANDLED EXCEPTION: in post replicate hook for %s' , broker . db_file ) #NEWLINE# #UNINDENT# if ( ( not shouldbehere ) and all ( responses ) ) : #NEWLINE# #INDENT# if ( not self . delete_db ( broker ) ) : #NEWLINE# #INDENT# failure_devs_info . update ( [ ( failure_dev [ 'replication_ip' ] , failure_dev [ 'device' ] ) for failure_dev in repl_nodes ] ) #NEWLINE# #UNINDENT# #UNINDENT# target_devs_info = set ( [ ( target_dev [ 'replication_ip' ] , target_dev [ 'device' ] ) for target_dev in repl_nodes ] ) #NEWLINE# self . stats [ 'success' ] += len ( ( target_devs_info - failure_devs_info ) ) #NEWLINE# self . _add_failure_stats ( failure_devs_info ) #NEWLINE# self . logger . timing_since ( 'timing' , start_time )