Merge branch...

Merge branch '416777-increase-max_retries_after_interruption-for-github-import-stages' into 'master' 

Increase interruption retries for GitHub workers

See merge request https://gitlab.com/gitlab-org/gitlab/-/merge_requests/134949



Merged-by: Luke Duncalfe <lduncalfe@gitlab.com>
Approved-by: Madelein van Niekerk <mvanniekerk@gitlab.com>
Reviewed-by: George Koltsov <gkoltsov@gitlab.com>
Co-authored-by: Madelein van Niekerk <mvanniekerk@gitlab.com>
Co-authored-by: Luke Duncalfe <lduncalfe@eml.cc>

Gemfile.lock

@@ -152,7 +152,7 @@ PATH
 PATH
   remote: vendor/gems/sidekiq-reliable-fetch
   specs:
-    gitlab-sidekiq-fetcher (0.9.0)
+    gitlab-sidekiq-fetcher (0.10.0)
       json (>= 2.5)
       sidekiq (~> 6.1)
 

app/workers/concerns/gitlab/github_import/stage_methods.rb

@@ -5,6 +5,8 @@ module GithubImport
     module StageMethods
       extend ActiveSupport::Concern
 
+      MAX_RETRIES_AFTER_INTERRUPTION = 20
+
       included do
         include ApplicationWorker
 
@@ -18,6 +20,29 @@ module StageMethods
         end
       end
 
+      class_methods do
+        # We can increase the number of times a GitHubImport::Stage worker is retried
+        # after being interrupted if the importer it executes can restart exactly
+        # from where it left off.
+        #
+        # It is not safe to call this method if the importer loops over its data from
+        # the beginning when restarted, even if it skips data that is already imported
+        # inside the loop, as there is a possibility the importer will never reach
+        # the end of the loop.
+        #
+        # Examples of stage workers that call this method are ones that execute services that:
+        #
+        # - Continue paging an endpoint from where it left off:
+        #   https://gitlab.com/gitlab-org/gitlab/-/blob/487521cc/lib/gitlab/github_import/parallel_scheduling.rb#L114-117
+        # - Continue their loop from where it left off:
+        #   https://gitlab.com/gitlab-org/gitlab/-/blob/024235ec/lib/gitlab/github_import/importer/pull_requests/review_requests_importer.rb#L15
+        def resumes_work_when_interrupted!
+          return unless Feature.enabled?(:github_importer_raise_max_interruptions)
+
+          sidekiq_options max_retries_after_interruption: MAX_RETRIES_AFTER_INTERRUPTION
+        end
+      end
+
       # project_id - The ID of the GitLab project to import the data into.
       def perform(project_id)
         info(project_id, message: 'starting stage')

app/workers/gitlab/github_import/stage/import_attachments_worker.rb

@@ -11,6 +11,8 @@ class ImportAttachmentsWorker # rubocop:disable Scalability/IdempotentWorker
         include GithubImport::Queue
         include StageMethods
 
+        resumes_work_when_interrupted!
+
         # client - An instance of Gitlab::GithubImport::Client.
         # project - An instance of Project.
         def import(client, project)

app/workers/gitlab/github_import/stage/import_issue_events_worker.rb

@@ -11,6 +11,8 @@ class ImportIssueEventsWorker # rubocop:disable Scalability/IdempotentWorker
         include GithubImport::Queue
         include StageMethods
 
+        resumes_work_when_interrupted!
+
         # client - An instance of Gitlab::GithubImport::Client.
         # project - An instance of Project.
         def import(client, project)

app/workers/gitlab/github_import/stage/import_issues_and_diff_notes_worker.rb

@@ -11,6 +11,8 @@ class ImportIssuesAndDiffNotesWorker # rubocop:disable Scalability/IdempotentWor
         include GithubImport::Queue
         include StageMethods
 
+        resumes_work_when_interrupted!
+
         # client - An instance of Gitlab::GithubImport::Client.
         # project - An instance of Project.
         def import(client, project)

app/workers/gitlab/github_import/stage/import_lfs_objects_worker.rb

@@ -11,6 +11,11 @@ class ImportLfsObjectsWorker # rubocop:disable Scalability/IdempotentWorker
         include GithubImport::Queue
         include StageMethods
 
+        # Importer::LfsObjectsImporter can resume work when interrupted as
+        # it uses Projects::LfsPointers::LfsObjectDownloadListService which excludes LFS objects that already exist.
+        # https://gitlab.com/gitlab-org/gitlab/-/blob/eabf0800/app/services/projects/lfs_pointers/lfs_object_download_list_service.rb#L69-71
+        resumes_work_when_interrupted!
+
         def perform(project_id)
           return unless (project = find_project(project_id))
 

app/workers/gitlab/github_import/stage/import_notes_worker.rb

@@ -11,6 +11,8 @@ class ImportNotesWorker # rubocop:disable Scalability/IdempotentWorker
         include GithubImport::Queue
         include StageMethods
 
+        resumes_work_when_interrupted!
+
         # client - An instance of Gitlab::GithubImport::Client.
         # project - An instance of Project.
         def import(client, project)

app/workers/gitlab/github_import/stage/import_pull_requests_merged_by_worker.rb

@@ -11,6 +11,8 @@ class ImportPullRequestsMergedByWorker # rubocop:disable Scalability/IdempotentW
         include GithubImport::Queue
         include StageMethods
 
+        resumes_work_when_interrupted!
+
         # client - An instance of Gitlab::GithubImport::Client.
         # project - An instance of Project.
         def import(client, project)

app/workers/gitlab/github_import/stage/import_pull_requests_review_requests_worker.rb

@@ -11,6 +11,8 @@ class ImportPullRequestsReviewRequestsWorker # rubocop:disable Scalability/Idemp
         include GithubImport::Queue
         include StageMethods
 
+        resumes_work_when_interrupted!
+
         # client - An instance of Gitlab::GithubImport::Client.
         # project - An instance of Project.
         def import(client, project)

app/workers/gitlab/github_import/stage/import_pull_requests_reviews_worker.rb

@@ -11,6 +11,8 @@ class ImportPullRequestsReviewsWorker # rubocop:disable Scalability/IdempotentWo
         include GithubImport::Queue
         include StageMethods
 
+        resumes_work_when_interrupted!
+
         # client - An instance of Gitlab::GithubImport::Client.
         # project - An instance of Project.
         def import(client, project)

app/workers/gitlab/github_import/stage/import_pull_requests_worker.rb

@@ -11,6 +11,8 @@ class ImportPullRequestsWorker # rubocop:disable Scalability/IdempotentWorker
         include GithubImport::Queue
         include StageMethods
 
+        resumes_work_when_interrupted!
+
         # client - An instance of Gitlab::GithubImport::Client.
         # project - An instance of Project.
         def import(client, project)

config/feature_flags/development/github_importer_raise_max_interruptions.yml

+---
+name: github_importer_raise_max_interruptions
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/134949
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/429306
+milestone: '16.6'
+type: development
+group: group::import and integrate
+default_enabled: false

spec/workers/concerns/gitlab/github_import/stage_methods_spec.rb

@@ -185,4 +185,30 @@ def self.name
       expect(worker.find_project(-1)).to be_nil
     end
   end
+
+  describe '.resumes_work_when_interrupted!' do
+    subject(:sidekiq_options) { worker.class.sidekiq_options }
+
+    it 'does not set the `max_retries_after_interruption` if not called' do
+      is_expected.not_to have_key('max_retries_after_interruption')
+    end
+
+    it 'sets the `max_retries_after_interruption`' do
+      worker.class.resumes_work_when_interrupted!
+
+      is_expected.to include('max_retries_after_interruption' => 20)
+    end
+
+    context 'when the flag is disabled' do
+      before do
+        stub_feature_flags(github_importer_raise_max_interruptions: false)
+      end
+
+      it 'does not set `max_retries_after_interruption`' do
+        worker.class.resumes_work_when_interrupted!
+
+        is_expected.not_to have_key('max_retries_after_interruption')
+      end
+    end
+  end
 end

vendor/gems/sidekiq-reliable-fetch/Gemfile.lock

 PATH
   remote: .
   specs:
-    gitlab-sidekiq-fetcher (0.9.0)
+    gitlab-sidekiq-fetcher (0.10.0)
       json (>= 2.5)
       sidekiq (~> 6.1)
 

vendor/gems/sidekiq-reliable-fetch/gitlab-sidekiq-fetcher.gemspec

 Gem::Specification.new do |s|
   s.name          = 'gitlab-sidekiq-fetcher'
-  s.version       = '0.9.0'
+  s.version       = '0.10.0'
   s.authors       = ['TEA', 'GitLab']
   s.email         = 'valery@gitlab.com'
   s.license       = 'LGPL-3.0'

vendor/gems/sidekiq-reliable-fetch/lib/sidekiq/base_reliable_fetch.rb

@@ -230,7 +230,7 @@ def max_retries_after_interruption(worker_class)
       max_retries_after_interruption = nil
 
       max_retries_after_interruption ||= begin
-        Object.const_get(worker_class).sidekiq_options[:max_retries_after_interruption]
+        Object.const_get(worker_class).sidekiq_options['max_retries_after_interruption']
       rescue NameError
       end
 

vendor/gems/sidekiq-reliable-fetch/spec/base_reliable_fetch_spec.rb

@@ -76,6 +76,19 @@
       expect(queue2.size).to eq 1
       expect(Sidekiq::InterruptedSet.new.size).to eq 0
     end
+
+    it 'does not put jobs into interrupted queue if it is disabled on the worker' do
+      stub_const('Bob', double(sidekiq_options: { 'max_retries_after_interruption' => -1 }))
+
+      uow = described_class::UnitOfWork
+      interrupted_job = Sidekiq.dump_json(class: 'Bob', args: [1, 2, 'foo'], interrupted_count: 3)
+      jobs = [ uow.new('queue:foo', interrupted_job), uow.new('queue:foo', job), uow.new('queue:bar', job) ]
+      described_class.new(options).bulk_requeue(jobs, nil)
+
+      expect(queue1.size).to eq 2
+      expect(queue2.size).to eq 1
+      expect(Sidekiq::InterruptedSet.new.size).to eq 0
+    end
   end
 
   it 'sets heartbeat' do