Module: Executor::CloudStorage

Defined in:
lib/hyperflow-amqp-executor/cloud_storage.rb

Instance Method Summary collapse

Instance Method Details

#stage_in ⇒ Object 



9
10
11
12
13
14
15
16
17
18
19
20
21
 
# File 'lib/hyperflow-amqp-executor/cloud_storage.rb', line 9

def stage_in
  @bucket = @provider.directories.get(@job.options.bucket)

  @job.inputs.each do |file|
    Executor::logger.debug "[#{@id}] Downloading #{file.name}"
    File.open(@workdir+"/"+file.name, File::RDWR|File::CREAT) do |local_file|
      @bucket.files.get(@job.options.prefix+file.name) do |chunk, remaining_bytes, total_bytes|
        local_file.write(chunk)
        # print "\rDownloading #{file.name}: #{100*(total_bytes-remaining_bytes)/total_bytes}%"
      end
    end
  end
end

#stage_out ⇒ Object 



23
24
25
26
27
28
 
# File 'lib/hyperflow-amqp-executor/cloud_storage.rb', line 23

def stage_out
  @job.outputs.each do |file|
    Executor::logger.debug "[#{@id}] Uploading #{file.name}"
    @bucket.files.create(key: @job.options.prefix+file.name, body: File.open(@workdir+"/"+file.name))
  end
end

#storage_init ⇒ Object 



5
6
7
 
# File 'lib/hyperflow-amqp-executor/cloud_storage.rb', line 5

def storage_init
  @provider = Fog::Storage.new(@job.options.cloud_storage || Executor::settings.cloud_storage.to_h)
end

#workdir(&block) ⇒ Object 



30
31
32
 
# File 'lib/hyperflow-amqp-executor/cloud_storage.rb', line 30

def workdir(&block)
  Dir::mktmpdir(&block)
end