1 files changed, 22 insertions, 19 deletions

diff --git a/lib/bulk_imports/pipeline/runner.rb b/lib/bulk_imports/pipeline/runner.rb
index d39f4121b51..e3535e585cc 100644
--- a/lib/bulk_imports/pipeline/runner.rb
+++ b/lib/bulk_imports/pipeline/runner.rb
@@ -8,7 +8,7 @@ module BulkImports
       MarkedAsFailedError = Class.new(StandardError)
 
       def run
-        raise MarkedAsFailedError if marked_as_failed?
+        raise MarkedAsFailedError if context.entity.failed?
 
         info(message: 'Pipeline started')
 
@@ -40,7 +40,7 @@ module BulkImports
       private # rubocop:disable Lint/UselessAccessModifier
 
       def run_pipeline_step(step, class_name = nil)
-        raise MarkedAsFailedError if marked_as_failed?
+        raise MarkedAsFailedError if context.entity.failed?
 
         info(pipeline_step: step, step_class: class_name)
 
@@ -62,24 +62,13 @@ module BulkImports
       end
 
       def mark_as_failed
-        warn(message: 'Pipeline failed', pipeline_class: pipeline)
+        warn(message: 'Pipeline failed')
 
         context.entity.fail_op!
       end
 
-      def marked_as_failed?
-        return true if context.entity.failed?
-
-        false
-      end
-
       def log_skip(extra = {})
-        log = {
-          message: 'Skipping due to failed pipeline status',
-          pipeline_class: pipeline
-        }.merge(extra)
-
-        info(log)
+        info({ message: 'Skipping due to failed pipeline status' }.merge(extra))
       end
 
       def log_import_failure(exception, step)
@@ -92,25 +81,39 @@ module BulkImports
           correlation_id_value: Labkit::Correlation::CorrelationId.current_or_new_id
         }
 
+        error(
+          pipeline_step: step,
+          exception_class: exception.class.to_s,
+          exception_message: exception.message
+        )
+
         BulkImports::Failure.create(attributes)
       end
 
+      def info(extra = {})
+        logger.info(log_params(extra))
+      end
+
       def warn(extra = {})
         logger.warn(log_params(extra))
       end
 
-      def info(extra = {})
-        logger.info(log_params(extra))
+      def error(extra = {})
+        logger.error(log_params(extra))
       end
 
       def log_params(extra)
         defaults = {
+          bulk_import_id: context.bulk_import.id,
           bulk_import_entity_id: context.entity.id,
           bulk_import_entity_type: context.entity.source_type,
-          pipeline_class: pipeline
+          pipeline_class: pipeline,
+          context_extra: context.extra
         }
 
-        defaults.merge(extra).compact
+        defaults
+          .merge(extra)
+          .reject { |_key, value| value.blank? }
       end
 
       def logger