Remaining changes to transfer files from/to object storage

Each transfer code will handle local and remote in a slightly different
way.

This changes make sure we are switching from one mode to the other.

app/uploaders/file_uploader.rb

@@ -176,17 +176,6 @@ class FileUploader < GitlabUploader
     record_upload # after_store is not triggered
   end
 
-  # Used to replace an existing upload with the informed file without modifying stored metadata
-  # Use this method only to repair/replace an existing upload, or to upload to a Geo secondary node
-  #
-  # @param [CarrierWave::SanitizedFile] file
-  # @return CarrierWave::SanitizedFile
-  def replace_file_without_saving!(file)
-    raise ArgumentError, 'should be a CarrierWave::SanitizedFile' unless file.is_a? CarrierWave::SanitizedFile
-
-    storage.store!(file)
-  end
-
   private
 
   def apply_context!(uploader_context)

app/uploaders/gitlab_uploader.rb

@@ -99,6 +99,17 @@ class GitlabUploader < CarrierWave::Uploader::Base
     end
   end
 
+  # Used to replace an existing upload with the informed file without modifying stored metadata
+  # Use this method only to repair/replace an existing upload, or to upload to a Geo secondary node
+  #
+  # @param [CarrierWave::SanitizedFile] file
+  # @return CarrierWave::SanitizedFile
+  def replace_file_without_saving!(file)
+    raise ArgumentError, 'should be a CarrierWave::SanitizedFile' unless file.is_a? CarrierWave::SanitizedFile
+
+    storage.store!(file)
+  end
+
   private
 
   # Designed to be overridden by child uploaders that have a dynamic path

ee/lib/gitlab/geo/replication/file_transfer.rb

@@ -8,6 +8,10 @@ module Gitlab
       #   * Saving it in the right place on successful download
       #   * Returning a detailed Result object
       class FileTransfer < BaseTransfer
+        # Initialize a transfer service for a specified Upload
+        #
+        # @param [Symbol] file_type
+        # @param [Upload] upload
         def initialize(file_type, upload)
           if upload.local?
             super(local_file_attributes(file_type, upload))

ee/lib/gitlab/geo/replication/job_artifact_downloader.rb

@@ -14,7 +14,14 @@ module Gitlab
           return fail_before_transfer unless job_artifact.present?
 
           transfer = ::Gitlab::Geo::Replication::JobArtifactTransfer.new(job_artifact)
-          Result.from_transfer_result(transfer.download_from_primary)
+
+          result = if job_artifact.local_store?
+                     transfer.download_from_primary
+                   else
+                     transfer.stream_from_primary_to_object_storage
+                   end
+
+          Result.from_transfer_result(result)
         end
 
         private

ee/lib/gitlab/geo/replication/job_artifact_transfer.rb

@@ -8,17 +8,38 @@ module Gitlab
       #   * Saving it in the right place on successful download
       #   * Returning a detailed Result object
       class JobArtifactTransfer < BaseTransfer
+        # Initialize a transfer service for a specified Ci::JobArtifact
+        #
+        # @param [Ci::JobArtifact] job_artifact
         def initialize(job_artifact)
-          super(
+          if job_artifact.local_store?
+            super(local_job_artifact_attributes(job_artifact))
+          else
+            super(remote_job_artifact_attributes(job_artifact))
+          end
+        end
+
+        private
+
+        def local_job_artifact_attributes(job_artifact)
+          {
             file_type: :job_artifact,
             file_id: job_artifact.id,
             filename: job_artifact.file.path,
+            uploader: job_artifact.file,
             expected_checksum: job_artifact.file_sha256,
             request_data: job_artifact_request_data(job_artifact)
-          )
+          }
         end
 
-        private
+        def remote_job_artifact_attributes(job_artifact)
+          {
+            file_type: :job_artifact,
+            file_id: job_artifact.id,
+            uploader: job_artifact.file,
+            request_data: job_artifact_request_data(job_artifact)
+          }
+        end
 
         def job_artifact_request_data(job_artifact)
           {

ee/lib/gitlab/geo/replication/lfs_downloader.rb

@@ -14,7 +14,14 @@ module Gitlab
           return fail_before_transfer unless lfs_object.present?
 
           transfer = ::Gitlab::Geo::Replication::LfsTransfer.new(lfs_object)
-          Result.from_transfer_result(transfer.download_from_primary)
+
+          result = if lfs_object.local_store?
+                     transfer.download_from_primary
+                   else
+                     transfer.stream_from_primary_to_object_storage
+                   end
+
+          Result.from_transfer_result(result)
         end
 
         private

ee/lib/gitlab/geo/replication/lfs_transfer.rb

@@ -8,17 +8,39 @@ module Gitlab
       #   * Saving it in the right place on successful download
       #   * Returning a detailed Result object
       class LfsTransfer < BaseTransfer
+        # Initialize a transfer service for a specified LfsObject
+        #
+        # @param [LfsObject] lfs_object
         def initialize(lfs_object)
-          super(
+          if lfs_object.local_store?
+            super(local_lfs_object_attributes(lfs_object))
+          else
+            super(remote_lfs_object_attributes(lfs_object))
+          end
+        end
+
+        private
+
+        def local_lfs_object_attributes(lfs_object)
+          {
             file_type: :lfs,
             file_id: lfs_object.id,
             filename: lfs_object.file.path,
+            uploader: lfs_object.file,
             expected_checksum: lfs_object.oid,
             request_data: lfs_request_data(lfs_object)
-          )
+          }
         end
 
-        private
+        def remote_lfs_object_attributes(lfs_object)
+          {
+            file_type: :lfs,
+            file_id: lfs_object.id,
+            uploader: lfs_object.file,
+            expected_checksum: lfs_object.oid,
+            request_data: lfs_request_data(lfs_object)
+          }
+        end
 
         def lfs_request_data(lfs_object)
           {

ee/spec/lib/gitlab/geo/replication/job_artifact_transfer_spec.rb

@@ -42,7 +42,7 @@ describe Gitlab::Geo::Replication::JobArtifactTransfer, :geo do
 
     context 'when the destination filename is a directory' do
       it 'returns a failed result' do
-        expect(job_artifact).to receive(:file).and_return(double(path: '/tmp'))
+        allow(job_artifact).to receive(:file).and_return(double(path: '/tmp'))
 
         result = subject.download_from_primary
 
@@ -106,7 +106,7 @@ describe Gitlab::Geo::Replication::JobArtifactTransfer, :geo do
 
     context "invalid path" do
       it 'logs an error if the destination directory could not be created' do
-        expect(job_artifact).to receive(:file).and_return(double(path: '/foo/bar'))
+        allow(job_artifact).to receive(:file).and_return(double(path: '/foo/bar'))
 
         allow(FileUtils).to receive(:mkdir_p) { raise Errno::EEXIST }
 

ee/spec/lib/gitlab/geo/replication/lfs_transfer_spec.rb

@@ -20,7 +20,7 @@ describe Gitlab::Geo::Replication::LfsTransfer do
 
     context 'when the destination filename is a directory' do
       it 'returns a failed result' do
-        expect(lfs_object).to receive(:file).and_return(double(path: '/tmp'))
+        allow(lfs_object).to receive(:file).and_return(double(path: '/tmp'))
 
         result = subject.download_from_primary
 
@@ -84,7 +84,7 @@ describe Gitlab::Geo::Replication::LfsTransfer do
 
     context "invalid path" do
       it 'logs an error if the destination directory could not be created' do
-        expect(lfs_object).to receive(:file).and_return(double(path: '/foo/bar'))
+        allow(lfs_object).to receive(:file).and_return(double(path: '/foo/bar'))
 
         allow(FileUtils).to receive(:mkdir_p) { raise Errno::EEXIST }
 

spec/uploaders/gitlab_uploader_spec.rb

@@ -69,6 +69,16 @@ describe GitlabUploader do
     end
   end
 
+  describe '#replace_file_without_saving!' do
+    it 'allows file to be replaced without triggering any callbacks' do
+      new_file = CarrierWave::SanitizedFile.new(Tempfile.new)
+
+      expect(subject).not_to receive(:with_callbacks)
+
+      subject.replace_file_without_saving!(new_file)
+    end
+  end
+
   describe '#open' do
     context 'when trace is stored in File storage' do
       context 'when file exists' do