Add relations export request when Bulk Import is initiated

  - When Bulk Import is initiated make sure to request
  relations export for each of the entities that need to be
  imported

Changelog: added

app/models/bulk_imports/entity.rb

@@ -68,6 +68,10 @@ class BulkImports::Entity < ApplicationRecord
     end
   end
 
+  def encoded_source_full_path
+    ERB::Util.url_encode(source_full_path)
+  end
+
   private
 
   def validate_parent_is_a_group

app/workers/all_queues.yml

@@ -1836,6 +1836,15 @@
   :idempotent:
   :tags:
   - :exclude_from_kubernetes
+- :name: bulk_imports_export_request
+  :worker_name: BulkImports::ExportRequestWorker
+  :feature_category: :importers
+  :has_external_dependencies: true
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: bulk_imports_pipeline
   :worker_name: BulkImports::PipelineWorker
   :feature_category: :importers

app/workers/bulk_import_worker.rb

@@ -24,6 +24,7 @@ class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker
     created_entities.first(next_batch_size).each do |entity|
       create_pipeline_tracker_for(entity)
 
+      BulkImports::ExportRequestWorker.perform_async(entity.id)
       BulkImports::EntityWorker.perform_async(entity.id)
 
       entity.start!

app/workers/bulk_imports/export_request_worker.rb

+# frozen_string_literal: true
+
+module BulkImports
+  class ExportRequestWorker
+    include ApplicationWorker
+
+    idempotent!
+    worker_has_external_dependencies!
+    feature_category :importers
+
+    GROUP_EXPORTED_URL_PATH = "/groups/%s/export_relations"
+
+    def perform(entity_id)
+      entity = BulkImports::Entity.find(entity_id)
+
+      request_export(entity)
+    end
+
+    private
+
+    def request_export(entity)
+      http_client(entity.bulk_import.configuration)
+        .post(GROUP_EXPORTED_URL_PATH % entity.encoded_source_full_path)
+    end
+
+    def http_client(configuration)
+      @client ||= Clients::Http.new(
+        uri: configuration.url,
+        token: configuration.access_token
+      )
+    end
+  end
+end

changelogs/unreleased/georgekoltsov-add-export-request-worker.yml

+---
+title: Add relations export request when Bulk Import is initiated
+merge_request: 61365
+author:
+type: changed

config/sidekiq_queues.yml

@@ -56,6 +56,8 @@
   - 1
 - - bulk_imports_entity
   - 1
+- - bulk_imports_export_request
+  - 1
 - - bulk_imports_pipeline
   - 1
 - - bulk_imports_relation_export

lib/bulk_imports/clients/http.rb

@@ -28,6 +28,17 @@ module BulkImports
         end
       end
 
+      def post(resource, body = {})
+        with_error_handling do
+          Gitlab::HTTP.post(
+            resource_url(resource),
+            headers: request_headers,
+            follow_redirects: false,
+            body: body
+          )
+        end
+      end
+
       def each_page(method, resource, query = {}, &block)
         return to_enum(__method__, method, resource, query) unless block_given?
 

spec/lib/bulk_imports/clients/http_spec.rb

@@ -8,42 +8,50 @@ RSpec.describe BulkImports::Clients::Http do
   let(:uri) { 'http://gitlab.example' }
   let(:token) { 'token' }
   let(:resource) { 'resource' }
+  let(:response_double) { double(code: 200, success?: true, parsed_response: {}) }
 
   subject { described_class.new(uri: uri, token: token) }
 
-  describe '#get' do
-    let(:response_double) { double(code: 200, success?: true, parsed_response: {}) }
-
   shared_examples 'performs network request' do
     it 'performs network request' do
-        expect(Gitlab::HTTP).to receive(:get).with(*expected_args).and_return(response_double)
+      expect(Gitlab::HTTP).to receive(method).with(*expected_args).and_return(response_double)
+
+      subject.public_send(method, resource)
+    end
+
+    context 'error handling' do
+      context 'when error occurred' do
+        it 'raises ConnectionError' do
+          allow(Gitlab::HTTP).to receive(method).and_raise(Errno::ECONNREFUSED)
 
-        subject.get(resource)
+          expect { subject.public_send(method, resource) }.to raise_exception(described_class::ConnectionError)
         end
       end
 
-    describe 'request query' do
-      include_examples 'performs network request' do
-        let(:expected_args) do
-          [
-            anything,
-            hash_including(
-              query: {
-                page: described_class::DEFAULT_PAGE,
-                per_page: described_class::DEFAULT_PER_PAGE
-              }
-            )
-          ]
+      context 'when response is not success' do
+        it 'raises ConnectionError' do
+          response_double = double(code: 503, success?: false)
+
+          allow(Gitlab::HTTP).to receive(method).and_return(response_double)
+
+          expect { subject.public_send(method, resource) }.to raise_exception(described_class::ConnectionError)
+        end
       end
     end
   end
 
-    describe 'request headers' do
+  describe '#get' do
+    let(:method) { :get }
+
     include_examples 'performs network request' do
       let(:expected_args) do
         [
-            anything,
+          'http://gitlab.example:80/api/v4/resource',
           hash_including(
+            query: {
+              page: described_class::DEFAULT_PAGE,
+              per_page: described_class::DEFAULT_PER_PAGE
+            },
             headers: {
               'Content-Type' => 'application/json',
               'Authorization' => "Bearer #{token}"
@@ -52,35 +60,6 @@ RSpec.describe BulkImports::Clients::Http do
         ]
       end
     end
-    end
-
-    describe 'request uri' do
-      include_examples 'performs network request' do
-        let(:expected_args) do
-          ['http://gitlab.example:80/api/v4/resource', anything]
-        end
-      end
-    end
-
-    context 'error handling' do
-      context 'when error occurred' do
-        it 'raises ConnectionError' do
-          allow(Gitlab::HTTP).to receive(:get).and_raise(Errno::ECONNREFUSED)
-
-          expect { subject.get(resource) }.to raise_exception(described_class::ConnectionError)
-        end
-      end
-
-      context 'when response is not success' do
-        it 'raises ConnectionError' do
-          response_double = double(code: 503, success?: false)
-
-          allow(Gitlab::HTTP).to receive(:get).and_return(response_double)
-
-          expect { subject.get(resource) }.to raise_exception(described_class::ConnectionError)
-        end
-      end
-    end
 
     describe '#each_page' do
       let(:objects1) { [{ object: 1 }, { object: 2 }] }
@@ -129,4 +108,23 @@ RSpec.describe BulkImports::Clients::Http do
       end
     end
   end
+
+  describe '#post' do
+    let(:method) { :post }
+
+    include_examples 'performs network request' do
+      let(:expected_args) do
+        [
+          'http://gitlab.example:80/api/v4/resource',
+          hash_including(
+            body: {},
+            headers: {
+              'Content-Type' => 'application/json',
+              'Authorization' => "Bearer #{token}"
+            }
+          )
+        ]
+      end
+    end
+  end
 end

spec/models/bulk_imports/entity_spec.rb

@@ -125,4 +125,13 @@ RSpec.describe BulkImports::Entity, type: :model do
       end
     end
   end
+
+  describe '#encoded_source_full_path' do
+    it 'encodes entity source full path' do
+      expected = 'foo%2Fbar'
+      entity = build(:bulk_import_entity, source_full_path: 'foo/bar')
+
+      expect(entity.encoded_source_full_path).to eq(expected)
+    end
+  end
 end

spec/workers/bulk_import_worker_spec.rb

@@ -69,7 +69,7 @@ RSpec.describe BulkImportWorker do
       end
 
       context 'when there are created entities to process' do
-        it 'marks a batch of entities as started, enqueues BulkImports::EntityWorker and reenqueues' do
+        it 'marks a batch of entities as started, enqueues EntityWorker, ExportRequestWorker and reenqueues' do
           stub_const("#{described_class}::DEFAULT_BATCH_SIZE", 1)
 
           bulk_import = create(:bulk_import, :created)
@@ -78,6 +78,7 @@ RSpec.describe BulkImportWorker do
 
           expect(described_class).to receive(:perform_in).with(described_class::PERFORM_DELAY, bulk_import.id)
           expect(BulkImports::EntityWorker).to receive(:perform_async)
+          expect(BulkImports::ExportRequestWorker).to receive(:perform_async)
 
           subject.perform(bulk_import.id)
 

spec/workers/bulk_imports/export_request_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe BulkImports::ExportRequestWorker do
+  let_it_be(:bulk_import) { create(:bulk_import) }
+  let_it_be(:config) { create(:bulk_import_configuration, bulk_import: bulk_import) }
+  let_it_be(:entity) { create(:bulk_import_entity, source_full_path: 'foo/bar', bulk_import: bulk_import) }
+
+  let(:response_double) { double(code: 200, success?: true, parsed_response: {}) }
+  let(:job_args) { [entity.id] }
+
+  describe '#perform' do
+    before do
+      allow(Gitlab::HTTP).to receive(:post).and_return(response_double)
+    end
+
+    include_examples 'an idempotent worker' do
+      it 'requests relations export' do
+        expected = "/groups/foo%2Fbar/export_relations"
+
+        expect_next_instance_of(BulkImports::Clients::Http) do |client|
+          expect(client).to receive(:post).with(expected).twice
+        end
+
+        perform_multiple(job_args)
+      end
+    end
+  end
+end