Merge branch 'georgekoltsov/fix-bulk-import-undefined-error' into 'master'

Update BulkImports Export to handle unexpected failure

See merge request gitlab-org/gitlab!82221

app/models/bulk_imports/entity.rb

@@ -20,6 +20,8 @@
 class BulkImports::Entity < ApplicationRecord
   self.table_name = 'bulk_import_entities'
 
+  FailedError = Class.new(StandardError)
+
   belongs_to :bulk_import, optional: false
   belongs_to :parent, class_name: 'BulkImports::Entity', optional: true
 

app/models/bulk_imports/export_status.rb

@@ -30,7 +30,12 @@ module BulkImports
 
     def export_status
       strong_memoize(:export_status) do
-        fetch_export_status.find { |item| item['relation'] == relation }
+        status = fetch_export_status
+
+        # Consider empty response as failed export
+        raise StandardError, 'Empty export status response' unless status&.present?
+
+        status.find { |item| item['relation'] == relation }
       end
     rescue StandardError => e
       { 'status' => Export::FAILED, 'error' => e.message }

app/workers/bulk_imports/export_request_worker.rb

@@ -14,6 +14,10 @@ module BulkImports
       entity = BulkImports::Entity.find(entity_id)
 
       request_export(entity)
+    rescue BulkImports::NetworkError => e
+      log_export_failure(e, entity)
+
+      entity.fail_op!
     end
 
     private
@@ -28,5 +32,24 @@ module BulkImports
         token: configuration.access_token
       )
     end
+
+    def log_export_failure(exception, entity)
+      attributes = {
+        bulk_import_entity_id: entity.id,
+        pipeline_class: 'ExportRequestWorker',
+        exception_class: exception.class.to_s,
+        exception_message: exception.message.truncate(255),
+        correlation_id_value: Labkit::Correlation::CorrelationId.current_or_new_id
+      }
+
+      Gitlab::Import::Logger.warn(
+        attributes.merge(
+          bulk_import_id: entity.bulk_import.id,
+          bulk_import_entity_type: entity.source_type
+        )
+      )
+
+      BulkImports::Failure.create(attributes)
+    end
   end
 end

app/workers/bulk_imports/pipeline_worker.rb

@@ -43,6 +43,10 @@ module BulkImports
     private
 
     def run(pipeline_tracker)
+      if pipeline_tracker.entity.failed?
+        raise(Entity::FailedError, 'Failed entity status')
+      end
+
       if ndjson_pipeline?(pipeline_tracker)
         status = ExportStatus.new(pipeline_tracker, pipeline_tracker.pipeline_class.relation)
 

lib/bulk_imports/clients/http.rb

@@ -123,7 +123,7 @@ module BulkImports
       def with_error_handling
         response = yield
 
-        raise ::BulkImports::NetworkError.new("Unsuccessful response #{response.code} from #{response.request.path.path}", response: response) unless response.success?
+        raise ::BulkImports::NetworkError.new("Unsuccessful response #{response.code} from #{response.request.path.path}. Body: #{response.parsed_response}", response: response) unless response.success?
 
         response
       rescue *Gitlab::HTTP::HTTP_ERRORS => e

spec/lib/bulk_imports/clients/http_spec.rb

@@ -38,11 +38,11 @@ RSpec.describe BulkImports::Clients::HTTP do
 
       context 'when response is not success' do
         it 'raises BulkImports::Error' do
-          response_double = double(code: 503, success?: false, request: double(path: double(path: '/test')))
+          response_double = double(code: 503, success?: false, parsed_response: 'Error', request: double(path: double(path: '/test')))
 
           allow(Gitlab::HTTP).to receive(method).and_return(response_double)
 
-          expect { subject.public_send(method, resource) }.to raise_exception(BulkImports::NetworkError, 'Unsuccessful response 503 from /test')
+          expect { subject.public_send(method, resource) }.to raise_exception(BulkImports::NetworkError, 'Unsuccessful response 503 from /test. Body: Error')
         end
       end
     end

spec/models/bulk_imports/export_status_spec.rb

@@ -55,6 +55,17 @@ RSpec.describe BulkImports::ExportStatus do
         expect(subject.failed?).to eq(false)
       end
     end
+
+    context 'when export status is not present' do
+      let(:response_double) do
+        double(parsed_response: [])
+      end
+
+      it 'returns true' do
+        expect(subject.failed?).to eq(true)
+        expect(subject.error).to eq('Empty export status response')
+      end
+    end
   end
 
   describe '#error' do

spec/workers/bulk_imports/export_request_worker_spec.rb

@@ -28,6 +28,31 @@ RSpec.describe BulkImports::ExportRequestWorker do
 
           perform_multiple(job_args)
         end
+
+        context 'when network error is raised' do
+          it 'logs export failure and marks entity as failed' do
+            expect_next_instance_of(BulkImports::Clients::HTTP) do |client|
+              expect(client).to receive(:post).and_raise(BulkImports::NetworkError, 'Export error').twice
+            end
+
+            expect(Gitlab::Import::Logger).to receive(:warn).with(
+              bulk_import_entity_id: entity.id,
+              pipeline_class: 'ExportRequestWorker',
+              exception_class: 'BulkImports::NetworkError',
+              exception_message: 'Export error',
+              correlation_id_value: anything,
+              bulk_import_id: bulk_import.id,
+              bulk_import_entity_type: entity.source_type
+            ).twice
+
+            perform_multiple(job_args)
+
+            failure = entity.failures.last
+
+            expect(failure.pipeline_class).to eq('ExportRequestWorker')
+            expect(failure.exception_message).to eq('Export error')
+          end
+        end
       end
     end
 

spec/workers/bulk_imports/pipeline_worker_spec.rb

@@ -136,6 +136,34 @@ RSpec.describe BulkImports::PipelineWorker do
       expect(pipeline_tracker.jid).to eq('jid')
     end
 
+    context 'when entity is failed' do
+      it 'marks tracker as failed and logs the error' do
+        pipeline_tracker = create(
+          :bulk_import_tracker,
+          entity: entity,
+          pipeline_name: 'Pipeline',
+          status_event: 'enqueue'
+        )
+
+        entity.update!(status: -1)
+
+        expect_next_instance_of(Gitlab::Import::Logger) do |logger|
+          expect(logger)
+            .to receive(:error)
+            .with(
+              worker: described_class.name,
+              pipeline_name: 'Pipeline',
+              entity_id: entity.id,
+              message: 'Failed entity status'
+            )
+        end
+
+        subject.perform(pipeline_tracker.id, pipeline_tracker.stage, entity.id)
+
+        expect(pipeline_tracker.reload.status_name).to eq(:failed)
+      end
+    end
+
     context 'when it is a network error' do
       it 'reenqueue on retriable network errors' do
         pipeline_tracker = create(