use state instead of files for sftp jobs

Author: graft

etna/lib/etna/remote.rb

@@ -13,6 +13,7 @@ def initialize(host:, username:, password: nil, port: 22, root:, **args)
       @host = host
       @port = port
       @root = root
+      raise "RemoteSSH must have host, username and password" unless @host && @username && @password
     end
 
     def ssh

polyphemus/lib/data_eng/clients/sftp_client.rb

@@ -6,6 +6,9 @@ def initialize(host, username, password, port: 22)
     @username = username
     @password = password
     @port = port
+
+    raise "SFTPClient must have host, username and password" unless @host && @username && @password
+
     begin
       Net::SFTP.start(@host, @username, password: @password, port: @port) do |sftp|
         sftp.dir.entries('.')

polyphemus/lib/data_eng/jobs/common.rb

@@ -1,9 +1,3 @@
-def build_pipeline_state_dir(path_to_write_files, run_id)
-  dir_path = File.join(path_to_write_files, run_id)
-  FileUtils.mkdir_p(dir_path)
-  dir_path
-end
-
 def remove_dscolab_prefix(path)
-    path.gsub('DSCOLAB_', '')
-end
+  path.gsub('DSCOLAB_', '')
+end

polyphemus/lib/data_eng/jobs/sftp_deposit_uploader.rb

@@ -20,8 +20,12 @@ def workflow_version
     config['version_number']
   end
 
-  def path_to_write_files
-    config['config']['path_to_write_files']
+  def magic_string
+    config['config']['magic_string']
+  end
+
+  def name_regex
+    Regexp.new("#{magic_string}(-|_)")
   end
 
   def deposit_root_path
@@ -57,9 +61,7 @@ def pre(context)
       raise "Run #{run_id} not found"
     end
 
-    context[:files_to_update] = CSV.foreach(run["state"]["files_to_update_path"], headers: true).map do |row|
-      { path: row['path'], modified_time: row['modified_time'].to_i }
-    end
+    context[:files_to_update] = run["state"]["files_to_update"].map(&:symbolize_keys)
 
     if context[:files_to_update].empty?
       logger.info("No new files to upload...")
@@ -89,7 +91,7 @@ def process(context)
       end
 
       begin
-        deposit_path = File.join(deposit_root_path, remove_dscolab_prefix(sftp_path))
+        deposit_path = File.join(deposit_root_path, sftp_path.gsub(name_regex,''))
         remote_ssh.file_upload(deposit_path, file_stream)
       rescue Etna::RemoteSSH::RemoteSSHError => e
         logger.warn("Failed to upload to deposit host #{deposit_host}: #{deposit_path}. Error: #{e.message}")
@@ -100,23 +102,15 @@ def process(context)
 
   def post(context)
     if context[:failed_files].any?
-      writable_dir = build_pipeline_state_dir(path_to_write_files, run_id)
-      context[:failed_files_path] = File.join(writable_dir, DEPOSIT_FAILED_FILES_CSV)
-
-      logger.warn("Writing failed files to #{context[:failed_files_path]}")
       logger.warn("Found #{context[:failed_files].size} failed files")
 
-      CSV.open(context[:failed_files_path], "wb") do |csv|
-        csv << ["path"]
-        context[:failed_files].each do |path|
-          csv << [path]
-        end
+      context[:failed_files].each do |file_path|
+        logger.warn(file_path)
       end
 
       polyphemus_client.update_run(project_name, run_id, {
         state: {
-          deposit_num_failed_files: context[:failed_files].size,
-          deposit_failed_files_path: context[:failed_files_path],
+          deposit_num_failed_files: context[:failed_files].size
         },
       })
     else

polyphemus/lib/data_eng/jobs/sftp_file_discovery.rb

@@ -6,8 +6,6 @@ class SftpFileDiscoveryJob < Polyphemus::ETLJob
   include WithEtnaClients
   include WithLogger
 
-  SFTP_FILES_TO_UPDATE_CSV = "sftp_files_to_update.csv"
-
   private
 
   def project_name
@@ -22,16 +20,16 @@ def workflow_version
     config['version_number']
   end
 
-  def file_regex
-    config['config']['file_regex']
+  def magic_string
+    config['config']['magic_string']
   end
 
-  def sftp_root_dir
-    config['config']['sftp_root_dir']
+  def file_regex
+    Regexp.new("#{magic_string}(-|_).*")
   end
 
-  def path_to_write_files
-    config['config']['path_to_write_files']
+  def ingest_root_path
+    config['config']['ingest_root_path']
   end
 
   def restart_scan?
@@ -43,7 +41,7 @@ def initial_start_scan_time
   end
 
   def override_interval
-    runtime_config['config']['override_interval'] || nil
+    runtime_config['config']['override_interval']
   end
 
   def sftp_client
@@ -69,27 +67,20 @@ def pre(context)
   def process(context)
     logger.info("Searching for files from #{context[:start_time]} to #{context[:end_time]}")
     files_to_update = sftp_client.search_files(
-      sftp_root_dir,
+      ingest_root_path,
       file_regex,
       context[:start_time],
       context[:end_time ]
     )
-    if files_to_update.empty?
-      context[:num_files_to_update] = 0
-      logger.info("No files found to update...")
-    else
-      context[:num_files_to_update] = files_to_update.size
-      writable_dir = build_pipeline_state_dir(path_to_write_files, run_id)
-      context[:files_to_update_path] = write_csv(writable_dir, files_to_update)
-      logger.info("Found #{files_to_update.size} files to update...")
-    end
+    logger.info("Found #{files_to_update.size} files to update...")
+    context[:files_to_update] = files_to_update
   end
 
   def post(context)
     polyphemus_client.update_run(project_name, run_id, {
       state: {
         num_files_to_update: context[:num_files_to_update],
-        files_to_update_path: context[:files_to_update_path],
+        files_to_update: context[:files_to_update],
         start_time: context[:start_time],
         end_time: context[:end_time]
       }
@@ -117,15 +108,4 @@ def fetch_last_scan
       response["end_time"]
     end
   end
-
-  def write_csv(dir_path, files_to_update)
-    filepath = File.join(dir_path, SFTP_FILES_TO_UPDATE_CSV)
-    CSV.open(filepath, "wb") do |csv|
-      csv << ["path", "modified_time"]
-      files_to_update.each do |file|
-        csv << [file[:path], file[:modified_time]]
-      end
-    end
-    filepath
-  end
 end

polyphemus/lib/data_eng/jobs/sftp_metis_uploader.rb

@@ -23,6 +23,14 @@ def workflow_version
     config['version_number']
   end
 
+  def magic_string
+    config['config']['magic_string']
+  end
+
+  def name_regex
+    Regexp.new("#{magic_string}(-|_)")
+  end
+
   def bucket_name
     config['config']['bucket_name']
   end
@@ -35,10 +43,6 @@ def metis_root_path
     config['config']['metis_root_path']
   end
 
-  def path_to_write_files
-    config['config']['path_to_write_files']
-  end
-
   def sftp_client
     @sftp_client ||= SFTPClient.new(
       config["secrets"]["sftp_ingest_host"],
@@ -51,19 +55,17 @@ def sftp_client
 
   def pre(context)
     run = polyphemus_client.get_run(project_name, run_id)
-    unless run
-        raise "Run #{run_id} not found"
-    end
-    context[:files_to_update] = CSV.foreach(run["state"]["files_to_update_path"], headers: true).map do |row|
-      { path: row['path'], modified_time: row['modified_time'].to_i }
-    end
+
+    raise "Run #{run_id} not found" unless run
+
+    context[:files_to_update] = run["state"]["files_to_update"].map(&:symbolize_keys)
 
     if context[:files_to_update].empty?
       logger.info("No new files to upload...")
       return false
-    else
-      return true
     end
+
+    return true
   end
 
   def process(context)
@@ -93,7 +95,7 @@ def process(context)
           metis_uid: metis_uid,
         )
 
-        metis_path = File.join(metis_root_path, remove_dscolab_prefix(sftp_path))
+        metis_path = File.join(metis_root_path, sftp_path.gsub(name_regex,''))
         uploader.do_upload(
           Etna::Clients::Metis::MetisUploadWorkflow::StreamingIOUpload.new(
             readable_io: file_stream,
@@ -111,21 +113,15 @@ def process(context)
 
   def post(context)
     if context[:failed_files].any?
-      writable_dir = build_pipeline_state_dir(path_to_write_files, run_id)
-      context[:failed_files_path] = File.join(writable_dir, METIS_FAILED_FILES_CSV)
-
-      logger.warn("Writing failed files to #{context[:failed_files_path]}")
       logger.warn("Found #{context[:failed_files].size} failed files")
-      CSV.open(context[:failed_files_path], "wb") do |csv|
-        csv << ["path"]
-        context[:failed_files].each do |path|
-        csv << [path]
-        end
+
+      context[:failed_files].each do |file_path|
+        logger.warn(file_path)
       end
+
       polyphemus_client.update_run(project_name, run_id, {
         state: {
-          metis_num_failed_files: context[:failed_files].size,
-          metis_failed_files_path: context[:failed_files_path],
+          metis_num_failed_files: context[:failed_files].size
         },
       })
     else

polyphemus/lib/data_eng/workflow_manifests/ingestion.rb

@@ -10,20 +10,22 @@ def as_json
             description: "This loader downloads data from a remote host to Metis or another remote host",
 	    title: "Ingestion Loader",
             properties: {
-              root_dir: { type: 'string' },
-              file_regex: { type: 'string' },
-              sftp_root_dir: { type: 'string' },
-              path_to_write_files: { type: 'string' },
               bucket_name: { type: 'string' },
+              magic_string: { type: 'string' },
+              ingest_root_path: { type: 'string' },
               metis_root_path: { type: 'string' },
               deposit_root_path: { type: 'string' }
             }
           },
           runtime_params: {
-            commit: 'boolean',
             initial_start_scan_time: 'integer', # unix timestamp
+            override_interval: 'integer',
+            restart_scan: 'boolean'
           },
-          secrets: [:sftp_host, :sftp_user, :sftp_password, :sftp_port],
+          secrets: [
+            :sftp_deposit_host, :sftp_deposit_password, :sftp_deposit_user,
+            :sftp_ingest_host, :sftp_ingest_password, :sftp_ingest_user
+          ],
           workflow_path: '/app/workflows/argo/ingestion/workflow.yaml'
         }
       end

polyphemus/spec/data_eng/sftp_deposit_uploader_spec.rb

@@ -7,14 +7,16 @@ def create_job(config, runtime_config)
       config = {
         "project_name" => "labors",
         "secrets" => {
-          "sftp_host" => "some-sftp-host", 
-          "sftp_user" => "user",
-          "sftp_password" => "password",
-          "sftp_port" => "22"
+          "sftp_ingest_host" => "some-sftp-host", 
+          "sftp_ingest_user" => "user",
+          "sftp_ingest_password" => "password",
+          "sftp_deposit_host" => "other-sftp-host", 
+          "sftp_deposit_user" => "user",
+          "sftp_deposit_password" => "password"
         },
         "config" => {
-          "file_regex" => "LABORS(-|_).*",
-          "sftp_root_dir" => "SSD",
+          "magic_string" => "LABORS",
+          "ingest_root_path" => "SSD",
           "path_to_write_files" => "/tmp/",
           "bucket_name" => "deposit",
           "metis_root_path" => "files/some-sftp-host",
@@ -49,24 +51,16 @@ def create_job(config, runtime_config)
         run_id: run_id,
         config_id: config["config_id"],
         version_number: config["version_number"],
-        state: {files_to_update_path: "/tmp/#{run_id}/#{SftpFileDiscoveryJob::SFTP_FILES_TO_UPDATE_CSV}"},
+        state: {
+          files_to_update: sftp_files
+        },
         orchestrator_metadata: {},
         output: nil,
         created_at: Time.now,
         updated_at: Time.now
       }
     }
 
-    def create_files_to_update_csv
-      FileUtils.mkdir_p("/tmp/#{run_id}")
-      CSV.open(run_record[:state][:files_to_update_path], "wb") do |csv|
-        csv << ["path", "modified_time"]
-        sftp_files.each do |file|
-          csv << [file[:path], file[:modified_time]]
-        end
-      end
-    end
-    
     before do
       ENV['TOKEN'] = TEST_TOKEN
       ENV['KUBE_ID'] = run_id
@@ -78,7 +72,6 @@ def create_files_to_update_csv
       let(:captured_requests) { [] }
 
       before do
-        create_files_to_update_csv
         stub_initial_sftp_connection
         stub_initial_ssh_connection
         stub_polyphemus_get_run(config["project_name"], run_id, run_record)
@@ -100,15 +93,16 @@ def create_files_to_update_csv
       end
 
       it 'fails to upload files' do
+        allow(Polyphemus.instance.logger).to receive(:warn)
+        expect(Polyphemus.instance.logger).to receive(:warn).with(/Failed to upload to deposit host other-sftp-host/)
+        expect(Polyphemus.instance.logger).to receive(:warn).with(/Found 1 failed files/)
+        expect(Polyphemus.instance.logger).to receive(:warn).with(/SSD.*LABORS_S1.fastq.gz/)
         stub_remote_ssh_file_upload(success: false)
 
         job = create_job(config, runtime_config)
         context = job.execute
 
-        failed_files_path = "/tmp/1234567890/#{SftpDepositUploaderJob::DEPOSIT_FAILED_FILES_CSV}"
-        expect(File.exist?(failed_files_path)).to be_truthy
         expect(captured_requests[0][:state][:deposit_num_failed_files]).to eq(1)
-        expect(captured_requests[0][:state][:deposit_failed_files_path]).to eq(failed_files_path)
       end
 
     end