1 files changed, 53 insertions, 2 deletions
diff --git a/app/workers/bulk_import_worker.rb b/app/workers/bulk_import_worker.rb
index 6bce13c5ff0..83b881ee525 100644
--- a/app/workers/bulk_import_worker.rb
+++ b/app/workers/bulk_import_worker.rb
@@ -22,7 +22,7 @@ class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker
     @bulk_import.start! if @bulk_import.created?
 
     created_entities.first(next_batch_size).each do |entity|
-      BulkImports::CreatePipelineTrackersService.new(entity).execute!
+      create_tracker(entity)
 
       entity.start!
 
@@ -51,7 +51,7 @@ class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker
   end
 
   def all_entities_failed?
-    entities.all? { |entity| entity.failed? }
+    entities.all?(&:failed?)
   end
 
   # A new BulkImportWorker job is enqueued to either
@@ -72,4 +72,55 @@ class BulkImportWorker # rubocop:disable Scalability/IdempotentWorker
   def next_batch_size
     [DEFAULT_BATCH_SIZE - started_entities.count, 0].max
   end
+
+  def create_tracker(entity)
+    entity.class.transaction do
+      entity.pipelines.each do |pipeline|
+        status = skip_pipeline?(pipeline, entity) ? :skipped : :created
+
+        entity.trackers.create!(
+          stage: pipeline[:stage],
+          pipeline_name: pipeline[:pipeline],
+          status: BulkImports::Tracker.state_machine.states[status].value
+        )
+      end
+    end
+  end
+
+  def skip_pipeline?(pipeline, entity)
+    return false unless entity.source_version.valid?
+
+    minimum_version, maximum_version = pipeline.values_at(:minimum_source_version, :maximum_source_version)
+
+    if source_version_out_of_range?(minimum_version, maximum_version, entity.source_version.without_patch)
+      log_skipped_pipeline(pipeline, entity, minimum_version, maximum_version)
+      return true
+    end
+
+    false
+  end
+
+  def source_version_out_of_range?(minimum_version, maximum_version, non_patch_source_version)
+    (minimum_version && non_patch_source_version < Gitlab::VersionInfo.parse(minimum_version)) ||
+      (maximum_version && non_patch_source_version > Gitlab::VersionInfo.parse(maximum_version))
+  end
+
+  def log_skipped_pipeline(pipeline, entity, minimum_version, maximum_version)
+    logger.info(
+      message: 'Pipeline skipped as source instance version not compatible with pipeline',
+      bulk_import_entity_id: entity.id,
+      bulk_import_id: entity.bulk_import_id,
+      bulk_import_entity_type: entity.source_type,
+      source_full_path: entity.source_full_path,
+      pipeline_name: pipeline[:pipeline],
+      minimum_source_version: minimum_version,
+      maximum_source_version: maximum_version,
+      source_version: entity.source_version.to_s,
+      importer: 'gitlab_migration'
+    )
+  end
+
+  def logger
+    @logger ||= Gitlab::Import::Logger.build
+  end
 end