Class: Etna::Clients::Metis::SyncMetisDataWorkflow
- Inherits:
-
Struct
- Object
- Struct
- Etna::Clients::Metis::SyncMetisDataWorkflow
- Defined in:
- lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb
Instance Attribute Summary collapse
-
#bucket_name ⇒ Object
Returns the value of attribute bucket_name.
-
#filesystem ⇒ Object
Returns the value of attribute filesystem.
-
#logger ⇒ Object
Returns the value of attribute logger.
-
#metis_client ⇒ Object
Returns the value of attribute metis_client.
-
#project_name ⇒ Object
Returns the value of attribute project_name.
-
#skip_tmpdir ⇒ Object
Returns the value of attribute skip_tmpdir.
Instance Method Summary collapse
- #bin_file_name(etag:) ⇒ Object
- #copy_directory(src, dest, root = dest, tmpdir = nil) ⇒ Object
- #copy_file(bin_root_dir:, tmpdir:, dest:, url:, stub: false) ⇒ Object
- #with_maybe_intermediate_tmp_dest(bin_file_name:, tmpdir:, dest_file_name:, &block) ⇒ Object
Instance Attribute Details
#bucket_name ⇒ Object
Returns the value of attribute bucket_name
9 10 11 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 9 def bucket_name @bucket_name end |
#filesystem ⇒ Object
Returns the value of attribute filesystem
9 10 11 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 9 def filesystem @filesystem end |
#logger ⇒ Object
Returns the value of attribute logger
9 10 11 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 9 def logger @logger end |
#metis_client ⇒ Object
Returns the value of attribute metis_client
9 10 11 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 9 def metis_client @metis_client end |
#project_name ⇒ Object
Returns the value of attribute project_name
9 10 11 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 9 def project_name @project_name end |
#skip_tmpdir ⇒ Object
Returns the value of attribute skip_tmpdir
9 10 11 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 9 def skip_tmpdir @skip_tmpdir end |
Instance Method Details
#bin_file_name(etag:) ⇒ Object
33 34 35 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 33 def bin_file_name(etag:) "bin/#{etag}" end |
#copy_directory(src, dest, root = dest, tmpdir = nil) ⇒ Object
11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 11 def copy_directory(src, dest, root = dest, tmpdir = nil) own_tmpdir = tmpdir.nil? && !skip_tmpdir if own_tmpdir tmpdir = filesystem.tmpdir end begin response = metis_client.list_folder(ListFolderRequest.new(project_name: project_name, bucket_name: bucket_name, folder_path: src)) response.files.all.each do |file| logger&.info("Copying file #{file.file_path} (#{Etna::Formatting.as_size(file.size)})") copy_file(bin_root_dir: root, tmpdir: tmpdir, dest: ::File.join(dest, file.file_name), url: file.download_url) end response.folders.all.each do |folder| copy_directory(::File.join(src, folder.folder_name), ::File.join(dest, folder.folder_name), root, tmpdir) end ensure filesystem.rm_rf(tmpdir) if own_tmpdir end end |
#copy_file(bin_root_dir:, tmpdir:, dest:, url:, stub: false) ⇒ Object
48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 48 def copy_file(bin_root_dir:, tmpdir:, dest:, url:, stub: false) = metis_client.(url) etag = [:etag] size = [:size] dest_bin_file = ::File.join(bin_root_dir, bin_file_name(etag: etag)) # Already materialized, continue if filesystem.exist?(dest_bin_file) return end with_maybe_intermediate_tmp_dest(bin_file_name: dest_bin_file, tmpdir: tmpdir, dest_file_name: dest) do |tmp_file| upload_timings = [] upload_amount = 0 last_rate = 0.00001 filesystem.with_writeable(tmp_file, "w", size_hint: size) do |io| if stub io.write("(stub) #{size} bytes") else metis_client.download_file(url) do |chunk| io.write(chunk) upload_timings << [chunk.length, Time.now.to_f] upload_amount += chunk.length if upload_timings.length > 150 s, _ = upload_timings.shift upload_amount -= s end _, start_time = upload_timings.first _, end_time = upload_timings.last if start_time == end_time next end rate = upload_amount / (end_time - start_time) if rate / last_rate > 1.3 || rate / last_rate < 0.7 logger&.info("Uploading #{Etna::Formatting.as_size(rate)} per second") if rate == 0 last_rate = 0.0001 else last_rate = rate end end end end end end filesystem.mkdir_p(::File.dirname(dest_bin_file)) filesystem.with_writeable(dest_bin_file, 'w', size_hint: 0) do |io| # empty file marking that this etag has been moved, to save a future write. end end |
#with_maybe_intermediate_tmp_dest(bin_file_name:, tmpdir:, dest_file_name:, &block) ⇒ Object
37 38 39 40 41 42 43 44 45 46 |
# File 'lib/etna/clients/metis/workflows/sync_metis_data_workflow.rb', line 37 def with_maybe_intermediate_tmp_dest(bin_file_name:, tmpdir:, dest_file_name:, &block) filesystem.mkdir_p(::File.dirname(dest_file_name)) if tmpdir.nil? yield dest_file_name else tmp_file = ::File.join(tmpdir, ::File.basename(bin_file_name)) yield tmp_file filesystem.mv(tmp_file, dest_file_name) end end |