<s> import logging <EOL> import math <EOL> import os <EOL> from oslo_config import cfg <EOL> from cloudferry . lib . copy_engines import base <EOL> from cloudferry . lib . utils import files <EOL> from cloudferry . lib . utils import local <EOL> from cloudferry . lib . utils import retrying <EOL> from cloudferry . lib . utils import sizeof_format <EOL> from cloudferry . lib . utils import ssh_util <EOL> LOG = logging . getLogger ( __name__ ) <EOL> CONF = cfg . CONF <EOL> class ScpCopier ( base . BaseCopier ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> def run_scp ( self , host_src , path_src , host_dst , path_dst , gateway = None ) : <EOL> """<STR_LIT>""" <EOL> cmd = "<STR_LIT>" <EOL> if CONF . migrate . direct_transfer : <EOL> src = path_src <EOL> runner = self . runner ( host_src , '<STR_LIT:src>' , gateway ) <EOL> run = runner . run <EOL> else : <EOL> src = '<STR_LIT>' . format ( <EOL> src_user = CONF . src . ssh_user , <EOL> src_host = host_src , <EOL> src_path = path_src , <EOL> ) <EOL> run = local . run <EOL> dst = '<STR_LIT>' . format ( <EOL> dst_user = CONF . dst . ssh_user , <EOL> dst_host = host_dst , <EOL> dst_path = path_dst , <EOL> ) <EOL> kwargs = { '<STR_LIT>' : host_src , <EOL> '<STR_LIT>' : path_src , <EOL> '<STR_LIT>' : host_dst , <EOL> '<STR_LIT>' : path_dst , <EOL> '<STR_LIT>' : gateway } <EOL> retrier = retrying . Retry ( <EOL> max_attempts = CONF . migrate . retry , <EOL> predicate = self . verify , <EOL> predicate_kwargs = kwargs , <EOL> ) <EOL> LOG . info ( "<STR_LIT>" , path_src , host_dst ) <EOL> try : <EOL> retrier . run ( run , cmd . format ( <EOL> ssh_opts = ssh_util . default_ssh_options ( ) , <EOL> src = src , <EOL> dst = dst ) , <EOL> capture_output = False ) <EOL> except retrying . MaxAttemptsReached : <EOL> self . clean_dst ( host_dst , path_dst ) <EOL> raise base . FileCopyError ( host_src = host_src , <EOL> path_src = path_src , <EOL> host_dst = host_dst , <EOL> path_dst = path_dst ) <EOL> def transfer ( self , data ) : <EOL> host_src = data [ '<STR_LIT>' ] <EOL> path_src = data [ '<STR_LIT>' ] <EOL> host_dst = data [ '<STR_LIT>' ] <EOL> path_dst = data [ '<STR_LIT>' ] <EOL> gateway = data . get ( '<STR_LIT>' ) <EOL> src_runner = self . runner ( host_src , '<STR_LIT:src>' , gateway ) <EOL> dst_runner = self . runner ( host_dst , '<STR_LIT>' , gateway ) <EOL> block_size = CONF . migrate . ssh_chunk_size <EOL> file_size = files . remote_file_size_mb ( src_runner , path_src ) <EOL> num_blocks = int ( math . ceil ( float ( file_size ) / block_size ) ) <EOL> src_temp_dir = os . path . join ( os . path . basename ( path_src ) , '<STR_LIT>' ) <EOL> dst_temp_dir = os . path . join ( os . path . basename ( path_dst ) , '<STR_LIT>' ) <EOL> partial_files = [ ] <EOL> with files . RemoteDir ( src_runner , src_temp_dir ) as src_temp , files . RemoteDir ( dst_runner , dst_temp_dir ) as dst_temp : <EOL> for i in xrange ( num_blocks ) : <EOL> part = os . path . basename ( path_src ) + '<STR_LIT>' . format ( i = i ) <EOL> part_path = os . path . join ( src_temp . dirname , part ) <EOL> files . remote_split_file ( src_runner , path_src , part_path , i , <EOL> block_size ) <EOL> gzipped_path = files . remote_gzip ( src_runner , part_path ) <EOL> gzipped_filename = os . path . basename ( gzipped_path ) <EOL> dst_gzipped_path = os . path . join ( dst_temp . dirname , <EOL> gzipped_filename ) <EOL> self . run_scp ( host_src , gzipped_path , host_dst , <EOL> dst_gzipped_path , gateway ) <EOL> files . remote_unzip ( dst_runner , dst_gzipped_path ) <EOL> partial_files . append ( os . path . join ( dst_temp . dirname , part ) ) <EOL> for i in xrange ( num_blocks ) : <EOL> files . remote_join_file ( dst_runner , path_dst , partial_files [ i ] , <EOL> i , block_size ) <EOL> if not self . verify ( host_src , path_src , host_dst , path_dst , gateway ) : <EOL> self . clean_dst ( host_dst , path_dst ) <EOL> raise base . FileCopyError ( ** data ) <EOL> def verify ( self , host_src , path_src , host_dst , path_dst , gateway = None ) : <EOL> """<STR_LIT>""" <EOL> src_runner = self . runner ( host_src , '<STR_LIT:src>' , gateway ) <EOL> dst_runner = self . runner ( host_dst , '<STR_LIT>' , gateway ) <EOL> src_size = files . remote_file_size ( src_runner , path_src ) <EOL> dst_size = files . remote_file_size ( dst_runner , path_dst ) <EOL> if src_size != dst_size : <EOL> LOG . warning ( "<STR_LIT>" , <EOL> path_src , sizeof_format . sizeof_fmt ( src_size ) , <EOL> path_dst , sizeof_format . sizeof_fmt ( dst_size ) ) <EOL> return False <EOL> if CONF . migrate . copy_with_md5_verification : <EOL> LOG . info ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> path_src , sizeof_format . sizeof_fmt ( src_size ) , host_src ) <EOL> src_md5 = files . remote_md5_sum ( src_runner , path_src ) <EOL> LOG . info ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> path_dst , sizeof_format . sizeof_fmt ( dst_size ) , host_dst ) <EOL> dst_md5 = files . remote_md5_sum ( dst_runner , path_dst ) <EOL> if src_md5 != dst_md5 : <EOL> LOG . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , path_src , src_md5 , path_dst , dst_md5 ) <EOL> return False <EOL> return True </s>
