<s> """<STR_LIT>""" <EOL> import errno <EOL> import logging <EOL> from eventlet import event <EOL> from eventlet import greenthread <EOL> from eventlet import queue <EOL> from eventlet import timeout <EOL> from oslo_vmware . _i18n import _ <EOL> from oslo_vmware import constants <EOL> from oslo_vmware import exceptions <EOL> from oslo_vmware . objects import datastore as ds_obj <EOL> from oslo_vmware import rw_handles <EOL> from oslo_vmware import vim_util <EOL> LOG = logging . getLogger ( __name__ ) <EOL> IMAGE_SERVICE_POLL_INTERVAL = <NUM_LIT:5> <EOL> FILE_READ_WRITE_TASK_SLEEP_TIME = <NUM_LIT> <EOL> BLOCKING_QUEUE_SIZE = <NUM_LIT:10> <EOL> class BlockingQueue ( queue . LightQueue ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , max_size , max_transfer_size ) : <EOL> """<STR_LIT>""" <EOL> queue . LightQueue . __init__ ( self , max_size ) <EOL> self . _max_transfer_size = max_transfer_size <EOL> self . _transferred = <NUM_LIT:0> <EOL> def read ( self , chunk_size ) : <EOL> """<STR_LIT>""" <EOL> if ( self . _max_transfer_size is <NUM_LIT:0> or <EOL> self . _transferred < self . _max_transfer_size ) : <EOL> data_item = self . get ( ) <EOL> self . _transferred += len ( data_item ) <EOL> return data_item <EOL> else : <EOL> LOG . debug ( "<STR_LIT>" , self . _transferred ) <EOL> return "<STR_LIT>" <EOL> def write ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . put ( data ) <EOL> def seek ( self , offset , whence = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> raise IOError ( errno . ESPIPE , "<STR_LIT>" ) <EOL> def tell ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _transferred <EOL> def close ( self ) : <EOL> pass <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" <EOL> class ImageWriter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , context , input_file , image_service , image_id , <EOL> image_meta = None ) : <EOL> """<STR_LIT>""" <EOL> if not image_meta : <EOL> image_meta = { } <EOL> self . _context = context <EOL> self . _input_file = input_file <EOL> self . _image_service = image_service <EOL> self . _image_id = image_id <EOL> self . _image_meta = image_meta <EOL> self . _running = False <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _done = event . Event ( ) <EOL> def _inner ( ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:image>' : self . _image_id , <EOL> '<STR_LIT>' : self . _image_meta } ) <EOL> try : <EOL> self . _image_service . update ( self . _context , <EOL> self . _image_id , <EOL> self . _image_meta , <EOL> data = self . _input_file ) <EOL> self . _running = True <EOL> while self . _running : <EOL> LOG . debug ( "<STR_LIT>" , <EOL> self . _image_id ) <EOL> image_meta = self . _image_service . show ( self . _context , <EOL> self . _image_id ) <EOL> image_status = image_meta . get ( '<STR_LIT:status>' ) <EOL> if image_status == '<STR_LIT>' : <EOL> self . stop ( ) <EOL> LOG . debug ( "<STR_LIT>" , <EOL> self . _image_id ) <EOL> self . _done . send ( True ) <EOL> elif image_status == '<STR_LIT>' : <EOL> self . stop ( ) <EOL> excep_msg = ( _ ( "<STR_LIT>" ) % <EOL> self . _image_id ) <EOL> LOG . error ( excep_msg ) <EOL> excep = exceptions . ImageTransferException ( excep_msg ) <EOL> self . _done . send_exception ( excep ) <EOL> elif image_status in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:image>' : self . _image_id , <EOL> '<STR_LIT:state>' : image_status , <EOL> '<STR_LIT>' : IMAGE_SERVICE_POLL_INTERVAL } ) <EOL> greenthread . sleep ( IMAGE_SERVICE_POLL_INTERVAL ) <EOL> else : <EOL> self . stop ( ) <EOL> excep_msg = ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % <EOL> { '<STR_LIT:image>' : self . _image_id , <EOL> '<STR_LIT:state>' : image_status } ) <EOL> LOG . error ( excep_msg ) <EOL> excep = exceptions . ImageTransferException ( excep_msg ) <EOL> self . _done . send_exception ( excep ) <EOL> except Exception as excep : <EOL> self . stop ( ) <EOL> excep_msg = ( _ ( "<STR_LIT>" ) % <EOL> self . _image_id ) <EOL> LOG . exception ( excep_msg ) <EOL> excep = exceptions . ImageTransferException ( excep_msg , excep ) <EOL> self . _done . send_exception ( excep ) <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:source>' : self . _input_file , <EOL> '<STR_LIT:image>' : self . _image_id } ) <EOL> greenthread . spawn ( _inner ) <EOL> return self . _done <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( "<STR_LIT>" , <EOL> self . _image_id ) <EOL> self . _running = False <EOL> def wait ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _done . wait ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def __str__ ( self ) : <EOL> string = "<STR_LIT>" % ( self . _input_file , <EOL> self . _image_id ) <EOL> return string <EOL> class FileReadWriteTask ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , input_file , output_file ) : <EOL> """<STR_LIT>""" <EOL> self . _input_file = input_file <EOL> self . _output_file = output_file <EOL> self . _running = False <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _done = event . Event ( ) <EOL> def _inner ( ) : <EOL> """<STR_LIT>""" <EOL> self . _running = True <EOL> while self . _running : <EOL> try : <EOL> data = self . _input_file . read ( rw_handles . READ_CHUNKSIZE ) <EOL> if not data : <EOL> LOG . debug ( "<STR_LIT>" ) <EOL> self . stop ( ) <EOL> self . _done . send ( True ) <EOL> self . _output_file . write ( data ) <EOL> if hasattr ( self . _input_file , "<STR_LIT>" ) : <EOL> self . _input_file . update_progress ( ) <EOL> if hasattr ( self . _output_file , "<STR_LIT>" ) : <EOL> self . _output_file . update_progress ( ) <EOL> greenthread . sleep ( FILE_READ_WRITE_TASK_SLEEP_TIME ) <EOL> except Exception as excep : <EOL> self . stop ( ) <EOL> excep_msg = _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> LOG . exception ( excep_msg ) <EOL> excep = exceptions . ImageTransferException ( excep_msg , excep ) <EOL> self . _done . send_exception ( excep ) <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT:source>' : self . _input_file , <EOL> '<STR_LIT>' : self . _output_file } ) <EOL> greenthread . spawn ( _inner ) <EOL> return self . _done <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( "<STR_LIT>" ) <EOL> self . _running = False <EOL> def wait ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _done . wait ( ) <EOL> def __str__ ( self ) : <EOL> string = ( "<STR_LIT>" % <EOL> ( self . _input_file , self . _output_file ) ) <EOL> return string <EOL> def _start_transfer ( context , timeout_secs , read_file_handle , max_data_size , <EOL> write_file_handle = None , image_service = None , image_id = None , <EOL> image_meta = None ) : <EOL> """<STR_LIT>""" <EOL> blocking_queue = BlockingQueue ( BLOCKING_QUEUE_SIZE , max_data_size ) <EOL> reader = FileReadWriteTask ( read_file_handle , blocking_queue ) <EOL> if write_file_handle : <EOL> writer = FileReadWriteTask ( blocking_queue , write_file_handle ) <EOL> elif image_service and image_id : <EOL> writer = ImageWriter ( context , <EOL> blocking_queue , <EOL> image_service , <EOL> image_id , <EOL> image_meta ) <EOL> else : <EOL> excep_msg = _ ( "<STR_LIT>" ) <EOL> LOG . error ( excep_msg ) <EOL> raise ValueError ( excep_msg ) <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : reader , <EOL> '<STR_LIT>' : writer } ) <EOL> reader . start ( ) <EOL> writer . start ( ) <EOL> timer = timeout . Timeout ( timeout_secs ) <EOL> try : <EOL> reader . wait ( ) <EOL> writer . wait ( ) <EOL> except ( timeout . Timeout , exceptions . ImageTransferException ) as excep : <EOL> excep_msg = ( _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % <EOL> { '<STR_LIT>' : reader , <EOL> '<STR_LIT>' : writer } ) <EOL> LOG . exception ( excep_msg ) <EOL> reader . stop ( ) <EOL> writer . stop ( ) <EOL> if isinstance ( excep , exceptions . ImageTransferException ) : <EOL> raise <EOL> raise exceptions . ImageTransferException ( excep_msg , excep ) <EOL> finally : <EOL> timer . cancel ( ) <EOL> read_file_handle . close ( ) <EOL> if write_file_handle : <EOL> write_file_handle . close ( ) <EOL> def download_image ( image , image_meta , session , datastore , rel_path , <EOL> bypass = True , timeout_secs = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> image_size = int ( image_meta [ '<STR_LIT:size>' ] ) <EOL> method = '<STR_LIT>' <EOL> if bypass : <EOL> hosts = datastore . get_connected_hosts ( session ) <EOL> host = ds_obj . Datastore . choose_host ( hosts ) <EOL> host_name = session . invoke_api ( vim_util , '<STR_LIT>' , <EOL> session . vim , host , '<STR_LIT:name>' ) <EOL> ds_url = datastore . build_url ( session . _scheme , host_name , rel_path , <EOL> constants . ESX_DATACENTER_PATH ) <EOL> cookie = ds_url . get_transfer_ticket ( session , method ) <EOL> conn = ds_url . connect ( method , image_size , cookie ) <EOL> else : <EOL> ds_url = datastore . build_url ( session . _scheme , session . _host , rel_path ) <EOL> cookie = '<STR_LIT>' % ( constants . SOAP_COOKIE_KEY , <EOL> session . vim . get_http_cookie ( ) . strip ( "<STR_LIT>" ) ) <EOL> conn = ds_url . connect ( method , image_size , cookie ) <EOL> conn . write = conn . send <EOL> read_handle = rw_handles . ImageReadHandle ( image ) <EOL> _start_transfer ( None , timeout_secs , read_handle , image_size , <EOL> write_file_handle = conn ) <EOL> def download_flat_image ( context , timeout_secs , image_service , image_id , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( "<STR_LIT>" , <EOL> image_id ) <EOL> read_iter = image_service . download ( context , image_id ) <EOL> read_handle = rw_handles . ImageReadHandle ( read_iter ) <EOL> file_size = int ( kwargs . get ( '<STR_LIT>' ) ) <EOL> write_handle = rw_handles . FileWriteHandle ( kwargs . get ( '<STR_LIT:host>' ) , <EOL> kwargs . get ( '<STR_LIT:port>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> file_size , <EOL> cacerts = kwargs . get ( '<STR_LIT>' ) ) <EOL> _start_transfer ( context , <EOL> timeout_secs , <EOL> read_handle , <EOL> file_size , <EOL> write_file_handle = write_handle ) <EOL> LOG . debug ( "<STR_LIT>" , <EOL> image_id ) <EOL> def download_stream_optimized_data ( context , timeout_secs , read_handle , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> file_size = int ( kwargs . get ( '<STR_LIT>' ) ) <EOL> write_handle = rw_handles . VmdkWriteHandle ( kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT:host>' ) , <EOL> kwargs . get ( '<STR_LIT:port>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> file_size ) <EOL> _start_transfer ( context , <EOL> timeout_secs , <EOL> read_handle , <EOL> file_size , <EOL> write_file_handle = write_handle ) <EOL> return write_handle . get_imported_vm ( ) <EOL> def download_stream_optimized_image ( context , timeout_secs , image_service , <EOL> image_id , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> image_id ) <EOL> read_iter = image_service . download ( context , image_id ) <EOL> read_handle = rw_handles . ImageReadHandle ( read_iter ) <EOL> imported_vm = download_stream_optimized_data ( context , timeout_secs , <EOL> read_handle , ** kwargs ) <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> image_id ) <EOL> return imported_vm <EOL> def copy_stream_optimized_disk ( <EOL> context , timeout_secs , write_handle , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> vmdk_file_path = kwargs . get ( '<STR_LIT>' ) <EOL> LOG . debug ( "<STR_LIT>" , <EOL> { '<STR_LIT>' : vmdk_file_path , <EOL> '<STR_LIT>' : write_handle . name } ) <EOL> file_size = kwargs . get ( '<STR_LIT>' ) <EOL> read_handle = rw_handles . VmdkReadHandle ( kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT:host>' ) , <EOL> kwargs . get ( '<STR_LIT:port>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> file_size ) <EOL> _start_transfer ( context , timeout_secs , read_handle , file_size , <EOL> write_file_handle = write_handle ) <EOL> LOG . debug ( "<STR_LIT>" , vmdk_file_path ) <EOL> def upload_image ( context , timeout_secs , image_service , image_id , owner_id , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( "<STR_LIT>" , image_id ) <EOL> file_size = kwargs . get ( '<STR_LIT>' ) <EOL> read_handle = rw_handles . VmdkReadHandle ( kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT:host>' ) , <EOL> kwargs . get ( '<STR_LIT:port>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> file_size ) <EOL> image_metadata = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : kwargs . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT:name>' : kwargs . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:size>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : { '<STR_LIT>' : <EOL> kwargs . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : owner_id } } <EOL> _start_transfer ( context , <EOL> timeout_secs , <EOL> read_handle , <EOL> <NUM_LIT:0> , <EOL> image_service = image_service , <EOL> image_id = image_id , <EOL> image_meta = image_metadata ) <EOL> LOG . debug ( "<STR_LIT>" , image_id ) </s>
