Backports EE mirror stuck handling feature (https://gitlab.com/gitlab-org/gitlab-ee/merge_requests/2628) to CEee-2628-backport-to-ce

4 files changed, 82 insertions, 33 deletions

diff --git a/app/models/project.rb b/app/models/project.rb
index 0de7da0ddaa..6955a8a161b 100644
--- a/app/models/project.rb
+++ b/app/models/project.rb
@@ -369,7 +369,10 @@ class Project < ActiveRecord::Base
     state :failed
 
     after_transition [:none, :finished, :failed] => :scheduled do |project, _|
-      project.run_after_commit { add_import_job }
+      project.run_after_commit do
+        job_id = add_import_job
+        update(import_jid: job_id) if job_id
+      end
     end
 
     after_transition started: :finished do |project, _|
@@ -524,17 +527,26 @@ class Project < ActiveRecord::Base
   def add_import_job
     job_id =
       if forked?
-        RepositoryForkWorker.perform_async(id, forked_from_project.repository_storage_path,
-          forked_from_project.full_path,
-          self.namespace.full_path)
+        RepositoryForkWorker.perform_async(id,
+                                           forked_from_project.repository_storage_path,
+                                           forked_from_project.full_path,
+                                           self.namespace.full_path)
       else
         RepositoryImportWorker.perform_async(self.id)
       end
 
+    log_import_activity(job_id)
+
+    job_id
+  end
+
+  def log_import_activity(job_id, type: :import)
+    job_type = type.to_s.capitalize
+
     if job_id
-      Rails.logger.info "Import job started for #{full_path} with job ID #{job_id}"
+      Rails.logger.info("#{job_type} job scheduled for #{full_path} with job ID #{job_id}.")
     else
-      Rails.logger.error "Import job failed to start for #{full_path}"
+      Rails.logger.error("#{job_type} job failed to create for #{full_path}.")
     end
   end
 
@@ -543,6 +555,7 @@ class Project < ActiveRecord::Base
       ProjectCacheWorker.perform_async(self.id)
     end
 
+    update(import_error: nil)
     remove_import_data
   end
 
diff --git a/app/workers/repository_fork_worker.rb b/app/workers/repository_fork_worker.rb
index a338523dc6b..cde5b45ad41 100644
--- a/app/workers/repository_fork_worker.rb
+++ b/app/workers/repository_fork_worker.rb
@@ -5,14 +5,17 @@ class RepositoryForkWorker
   include Gitlab::ShellAdapter
   include DedicatedSidekiqQueue
 
+  sidekiq_options status_expiration: StuckImportJobsWorker::IMPORT_JOBS_EXPIRATION
+
   def perform(project_id, forked_from_repository_storage_path, source_path, target_path)
+    project = Project.find(project_id)
+
+    return unless start_fork(project)
+
     Gitlab::Metrics.add_event(:fork_repository,
                               source_path: source_path,
                               target_path: target_path)
 
-    project = Project.find(project_id)
-    project.import_start
-
     result = gitlab_shell.fork_repository(forked_from_repository_storage_path, source_path,
                                           project.repository_storage_path, target_path)
     raise ForkError, "Unable to fork project #{project_id} for repository #{source_path} -> #{target_path}" unless result
@@ -33,6 +36,13 @@ class RepositoryForkWorker
 
   private
 
+  def start_fork(project)
+    return true if project.import_start
+
+    Rails.logger.info("Project #{project.full_path} was in inconsistent state (#{project.import_status}) while forking.")
+    false
+  end
+
   def fail_fork(project, message)
     Rails.logger.error(message)
     project.mark_import_as_failed(message)
diff --git a/app/workers/repository_import_worker.rb b/app/workers/repository_import_worker.rb
index 6be541abd3e..2c2d1e8b91f 100644
--- a/app/workers/repository_import_worker.rb
+++ b/app/workers/repository_import_worker.rb
@@ -4,23 +4,18 @@ class RepositoryImportWorker
   include Sidekiq::Worker
   include DedicatedSidekiqQueue
 
-  sidekiq_options status_expiration: StuckImportJobsWorker::IMPORT_EXPIRATION
-
-  attr_accessor :project, :current_user
+  sidekiq_options status_expiration: StuckImportJobsWorker::IMPORT_JOBS_EXPIRATION
 
   def perform(project_id)
-    @project = Project.find(project_id)
-    @current_user = @project.creator
+    project = Project.find(project_id)
 
-    project.import_start
+    return unless start_import(project)
 
     Gitlab::Metrics.add_event(:import_repository,
-                              import_url: @project.import_url,
-                              path: @project.full_path)
-
-    project.update_columns(import_jid: self.jid, import_error: nil)
+                              import_url: project.import_url,
+                              path: project.full_path)
 
-    result = Projects::ImportService.new(project, current_user).execute
+    result = Projects::ImportService.new(project, project.creator).execute
     raise ImportError, result[:message] if result[:status] == :error
 
     project.repository.after_import
@@ -37,6 +32,13 @@ class RepositoryImportWorker
 
   private
 
+  def start_import(project)
+    return true if project.import_start
+
+    Rails.logger.info("Project #{project.full_path} was in inconsistent state (#{project.import_status}) while importing.")
+    false
+  end
+
   def fail_import(project, message)
     project.mark_import_as_failed(message)
   end
diff --git a/app/workers/stuck_import_jobs_worker.rb b/app/workers/stuck_import_jobs_worker.rb
index bfc5e667bb6..f850e459cd9 100644
--- a/app/workers/stuck_import_jobs_worker.rb
+++ b/app/workers/stuck_import_jobs_worker.rb
@@ -2,36 +2,60 @@ class StuckImportJobsWorker
   include Sidekiq::Worker
   include CronjobQueue
 
-  IMPORT_EXPIRATION = 15.hours.to_i
+  IMPORT_JOBS_EXPIRATION = 15.hours.to_i
 
   def perform
-    stuck_projects.find_in_batches(batch_size: 500) do |group|
+    projects_without_jid_count = mark_projects_without_jid_as_failed!
+    projects_with_jid_count = mark_projects_with_jid_as_failed!
+
+    Gitlab::Metrics.add_event(:stuck_import_jobs,
+                             projects_without_jid_count: projects_without_jid_count,
+                             projects_with_jid_count: projects_with_jid_count)
+  end
+
+  private
+
+  def mark_projects_without_jid_as_failed!
+    started_projects_without_jid.each do |project|
+      project.mark_import_as_failed(error_message)
+    end.count
+  end
+
+  def mark_projects_with_jid_as_failed!
+    completed_jids_count = 0
+
+    started_projects_with_jid.find_in_batches(batch_size: 500) do |group|
       jids = group.map(&:import_jid)
 
       # Find the jobs that aren't currently running or that exceeded the threshold.
-      completed_jids = Gitlab::SidekiqStatus.completed_jids(jids)
+      completed_jids = Gitlab::SidekiqStatus.completed_jids(jids).to_set
 
       if completed_jids.any?
-        completed_ids = group.select { |project| completed_jids.include?(project.import_jid) }.map(&:id)
+        completed_jids_count += completed_jids.count
+        group.each do |project|
+          project.mark_import_as_failed(error_message) if completed_jids.include?(project.import_jid)
+        end
 
-        fail_batch!(completed_jids, completed_ids)
+        Rails.logger.info("Marked stuck import jobs as failed. JIDs: #{completed_jids.to_a.join(', ')}")
       end
     end
-  end
 
-  private
+    completed_jids_count
+  end
 
-  def stuck_projects
-    Project.select('id, import_jid').with_import_status(:started).where.not(import_jid: nil)
+  def started_projects
+    Project.with_import_status(:started)
   end
 
-  def fail_batch!(completed_jids, completed_ids)
-    Project.where(id: completed_ids).update_all(import_status: 'failed', import_error: error_message)
+  def started_projects_with_jid
+    started_projects.where.not(import_jid: nil)
+  end
 
-    Rails.logger.info("Marked stuck import jobs as failed. JIDs: #{completed_jids.join(', ')}")
+  def started_projects_without_jid
+    started_projects.where(import_jid: nil)
   end
 
   def error_message
-    "Import timed out. Import took longer than #{IMPORT_EXPIRATION} seconds"
+    "Import timed out. Import took longer than #{IMPORT_JOBS_EXPIRATION} seconds"
   end
 end