use ingest_metis_data_workflow for sftp_metis_uploader

Author: graft

etna/lib/etna/clients/metis/workflows/ingest_metis_data_workflow.rb

@@ -9,24 +9,35 @@ class IngestMetisDataWorkflow < Struct.new(:metis_filesystem, :ingest_filesystem
         # Since we are doing manual triage of files,
         #   do not automatically copy directory trees.
         # srcs must be a list of full paths to files.
-        def copy_files(srcs, &block)
-          srcs.each do |src|
+        def copy_files(files, &block)
+          files.each do |file|
+            if file.is_a?(Array)
+              src, dest = file
+            else
+              src = dest = file
+            end
+
             if !ingest_filesystem.exist?(src)
               logger&.warn("#{src} does not exist on source filesystem. Skipping.")
+              yield src, false if block_given?
               next
             end
 
             logger&.info("Copying file #{src} (#{Etna::Formatting.as_size(ingest_filesystem.stat(src).size)})")
 
             # For ingestion triage, just copy over the exact path + filename.
-            copy_file(dest: src, src: src, &block)
+            begin
+              copy_file(dest: dest, src: src, &block)
+            rescue Exception => e
+              yield src, false if block_given?
+            end
           end
         end
 
         def copy_file(dest:, src:, &block)
           ingest_filesystem.with_readable(src, "r") do |io|
             metis_filesystem.do_streaming_upload(io, dest, ingest_filesystem.stat(src).size)
-            yield src if block_given?
+            yield src, true if block_given?
           end
         end
       end

etna/lib/etna/clients/metis/workflows/metis_upload_workflow.rb

@@ -197,6 +197,7 @@ def next_blob_bytes
                 end
               end
             end
+            @last_bytes
           end
         end
       end

etna/lib/etna/filesystem.rb

@@ -438,9 +438,11 @@ def curl_cmd(path, opts=[])
       end
 
       def sftp_file_from_path(src)
-        file = ls(::File.dirname(src)).split("\n").map do |listing|
+        files = ls(::File.dirname(src)).split("\n").map do |listing|
           SftpFile.new(listing)
-        end.select do |file|
+        end
+
+        file = files.select do |file|
           file.name == ::File.basename(src)
         end
 

polyphemus/lib/data_eng/jobs/sftp_metis_uploader.rb

@@ -68,46 +68,41 @@ def pre(context)
     return true
   end
 
+  def metis_filesystem
+    Etna::Filesystem::Metis.new(
+      project_name: project_name,
+      bucket_name: bucket_name,
+      metis_client: metis_client,
+      root: metis_root_path
+    )
+  end
+
+  def ingest_filesystem
+    Etna::Filesystem::SftpFilesystem.new(
+      username: config["secrets"]["sftp_ingest_user"],
+      password: config["secrets"]["sftp_ingest_password"],
+      host: config["secrets"]["sftp_ingest_host"]
+    )
+  end
+
   def process(context)
     context[:failed_files] = []
 
-    context[:files_to_update].each do |file|
-      sftp_path = file[:path]
-      modified_time = file[:modified_time]
-
-      begin
-        file_stream = sftp_client.download_as_stream(sftp_path)
-        if file_stream.nil?
-          logger.info("Failed to download #{sftp_path}")
-          next
-        end
-      rescue StandardError => e
-        logger.info("Failed to download from sftp server, #{sftp_path}: #{e.message}")
-        context[:failed_files] << sftp_path
-        next
-      end
+    workflow = Etna::Clients::Metis::IngestMetisDataWorkflow.new(
+      metis_filesystem: metis_filesystem,
+      ingest_filesystem: ingest_filesystem,
+      logger: nil,
+    )
 
-      begin
-        uploader = Etna::Clients::Metis::MetisUploadWorkflow.new(
-          metis_client: metis_client,
-          project_name: project_name,
-          bucket_name: bucket_name,
-          metis_uid: metis_uid,
-        )
-
-        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,
-            size_hint: file_stream.size,
-          ),
-          metis_path
-        )
-      rescue StandardError => e
-        logger.warn("Failed to upload to metis: #{metis_path}. Error: #{e.message}")
-        context[:failed_files] << sftp_path 
+    workflow.copy_files(context[:files_to_update].map do |file|
+      [ file[:path], file[:path].gsub(name_regex, '') ]
+    end) do |filename, success|
+      if success
+        context[:successful_files] << filename
+      else
+        logger.warn("Failed to upload to metis: #{filename}")
+        context[:failed_files] << filename
       end
-
     end
   end
 

polyphemus/lib/etls/metis/loader.rb

@@ -133,7 +133,7 @@ def data_frame_update(update, tail, metis)
           )
 
           data.vectors.to_a.each do |name|
-            data.delete name unless column_map.has_key?(name)
+            data.delete_vector name unless column_map.has_key?(name)
           end
 
           # Blank data equaling values_to_ignore by setting as nil

polyphemus/spec/data_eng/sftp_metis_uploader_spec.rb

@@ -93,10 +93,19 @@ def stub_upload_file_with_stream(file_path, stream, force_error: false)
       stub_polyphemus_update_run(config["project_name"], run_id, captured_requests)
       stub_metis_setup
       stub_create_folder(bucket: config["config"]["bucket_name"], project: config["project_name"])
-    end 
+      
+      stub_request(
+        :get, "http://sftp//some-sftp-host/SSD/20240919_LH00416_0184_B22NF2WLT3/ACMK02/"
+      ).with(
+        headers: { 'Authorization'=>'Basic dXNlcjpwYXNzd29yZA==' }
+      ).to_return(status: 200, body: sftp_files.map { |f|
+        "drwxrwxr-x    4 eurystheus     labors        32 Aug 26  2022 #{::File.basename(f[:path])}"
+      }.join("\n"), headers: {})
+    end
 
     it 'successfully uploads files' do
       stub_upload_file_with_stream(file_to_upload, fake_stream)
+      allow_any_instance_of(Etna::Filesystem::SftpFilesystem).to receive(:with_readable).and_yield(StringIO.new("A"*32))
 
       job = create_job(config, runtime_config)
       context = job.execute
@@ -105,7 +114,8 @@ def stub_upload_file_with_stream(file_path, stream, force_error: false)
     end
 
     it 'fails to upload files' do
-      stub_upload_file_with_stream(file_to_upload, fake_stream, force_error: true)
+      allow_any_instance_of(Etna::Filesystem::SftpFilesystem).to receive(:with_readable).and_yield(StringIO.new)
+      stub_upload_file_with_stream(file_to_upload, fake_stream)
 
       allow(Polyphemus.instance.logger).to receive(:warn)
       expect(Polyphemus.instance.logger).to receive(:warn).with(/Failed to upload to metis/)

polyphemus/workflows/argo/ingestion/workflow.yaml

@@ -23,18 +23,6 @@ spec:
                   value: "{{workflow.parameters.config_id}}"
                 - name: version_number
                   value: "{{workflow.parameters.version_number}}"
-        - - name: c4-update
-            templateRef:
-              name: run-job-template
-              template: run-job
-            arguments:
-              parameters:
-                - name: job
-                  value: sftp_deposit_uploader
-                - name: config_id
-                  value: "{{workflow.parameters.config_id}}"
-                - name: version_number
-                  value: "{{workflow.parameters.version_number}}"
         - - name: metis-update
             templateRef:
               name: run-job-template