Merge branch '431936-prevent-stuckImportJob-marking-false-imports' into 'master'

Making refresh_jid more robust for github imports

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



Merged-by: Luke Duncalfe <lduncalfe@gitlab.com>
Reviewed-by: Luke Duncalfe <lduncalfe@gitlab.com>
Reviewed-by: Max Fan <mfan@gitlab.com>
Co-authored-by: Max Fan <mfan@gitlab.com>

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

@@ -10,6 +10,8 @@ module StageMethods
       included do
         include ApplicationWorker
 
+        sidekiq_options status_expiration: Gitlab::Import::StuckImportJob::IMPORT_JOBS_EXPIRATION
+
         sidekiq_retries_exhausted do |msg, e|
           Gitlab::Import::ImportFailureService.track(
             project_id: msg['args'][0],
@@ -77,7 +79,7 @@ def perform(project_id)
       # client - An instance of Gitlab::GithubImport::Client.
       # project - An instance of Project.
       def try_import(client, project)
-        project.import_state.refresh_jid_expiration
+        RefreshImportJidWorker.perform_in_the_future(project.id, jid)
 
         import(client, project)
       rescue RateLimitError

app/workers/gitlab/github_import/refresh_import_jid_worker.rb

@@ -10,7 +10,7 @@ class RefreshImportJidWorker # rubocop:disable Scalability/IdempotentWorker
       include GithubImport::Queue
 
       # The interval to schedule new instances of this job at.
-      INTERVAL = 1.minute.to_i
+      INTERVAL = 5.minutes.to_i
 
       def self.perform_in_the_future(*args)
         perform_in(INTERVAL, *args)
@@ -23,9 +23,11 @@ def perform(project_id, check_job_id)
         return unless import_state
 
         if SidekiqStatus.running?(check_job_id)
-          # As long as the repository is being cloned we want to keep refreshing
-          # the import JID status.
-          import_state.refresh_jid_expiration
+          # As long as the worker is running we want to keep refreshing
+          # the worker's JID as well as the import's JID.
+          Gitlab::SidekiqStatus.expire(check_job_id, Gitlab::Import::StuckImportJob::IMPORT_JOBS_EXPIRATION)
+          Gitlab::SidekiqStatus.set(import_state.jid, Gitlab::Import::StuckImportJob::IMPORT_JOBS_EXPIRATION)
+
           self.class.perform_in_the_future(project_id, check_job_id)
         end
 

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

@@ -14,12 +14,6 @@ class ImportRepositoryWorker # rubocop:disable Scalability/IdempotentWorker
         # client - An instance of Gitlab::GithubImport::Client.
         # project - An instance of Project.
         def import(client, project)
-          # In extreme cases it's possible for a clone to take more than the
-          # import job expiration time. To work around this we schedule a
-          # separate job that will periodically run and refresh the import
-          # expiration time.
-          RefreshImportJidWorker.perform_in_the_future(project.id, jid)
-
           info(project.id, message: "starting importer", importer: 'Importer::RepositoryImporter')
 
           # If a user creates an issue while the import is in progress, this can lead to an import failure.

app/workers/gitlab/import/advance_stage.rb

@@ -37,6 +37,8 @@ def perform(project_id, waiters, next_stage, timeout_timer = Time.zone.now.to_s,
         if new_job_count != previous_job_count
           timeout_timer = Time.zone.now
           previous_job_count = new_job_count
+
+          import_state_jid.refresh_jid_expiration
         end
 
         if new_waiters.empty?

lib/gitlab/sidekiq_status.rb

@@ -50,6 +50,16 @@ def self.unset(jid)
       end
     end
 
+    # Refreshes the timeout on the key if it exists
+    #
+    # jid = The Sidekiq job ID
+    # expire - The expiration time of the Redis key.
+    def self.expire(jid, expire = DEFAULT_EXPIRATION)
+      with_redis do |redis|
+        redis.expire(key_for(jid), expire)
+      end
+    end
+
     # Returns true if all the given job have been completed.
     #
     # job_ids - The Sidekiq job IDs to check.

spec/lib/gitlab/sidekiq_status_spec.rb

@@ -54,6 +54,31 @@
       end
     end
 
+    describe '.expire' do
+      it 'refreshes the expiration time if key is present' do
+        described_class.set('123', 1.minute)
+        described_class.expire('123', 1.hour)
+
+        key = described_class.key_for('123')
+
+        with_redis do |redis|
+          expect(redis.exists?(key)).to eq(true)
+          expect(redis.ttl(key) > 5.minutes).to eq(true)
+        end
+      end
+
+      it 'does nothing if key is not present' do
+        described_class.expire('123', 1.minute)
+
+        key = described_class.key_for('123')
+
+        with_redis do |redis|
+          expect(redis.exists?(key)).to eq(false)
+          expect(redis.ttl(key)).to eq(-2)
+        end
+      end
+    end
+
     describe '.all_completed?' do
       it 'returns true if all jobs have been completed' do
         expect(described_class.all_completed?(%w[123])).to eq(true)

spec/support/shared_examples/lib/gitlab/import/advance_stage_shared_examples.rb

@@ -63,7 +63,7 @@
         next_worker = described_class::STAGES[next_stage.to_sym]
 
         expect_next_found_instance_of(import_state.class) do |state|
-          expect(state).to receive(:refresh_jid_expiration)
+          expect(state).to receive(:refresh_jid_expiration).twice
         end
 
         expect(next_worker).to receive(:perform_async).with(project.id)
@@ -124,7 +124,7 @@
           freeze_time do
             next_worker = described_class::STAGES[next_stage.to_sym]
 
-            expect(next_worker).not_to receive(:perform_async).with(project.id)
+            expect(next_worker).not_to receive(:perform_async)
             expect_next_instance_of(described_class) do |klass|
               expect(klass).to receive(:find_import_state).and_call_original
             end

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

@@ -138,6 +138,10 @@ def self.name
   end
 
   describe '#try_import' do
+    before do
+      allow(worker).to receive(:jid).and_return('jid')
+    end
+
     it 'imports the project' do
       client = double(:client)
 
@@ -145,7 +149,7 @@ def self.name
         .to receive(:import)
         .with(client, project)
 
-      expect(project.import_state).to receive(:refresh_jid_expiration)
+      expect(Gitlab::GithubImport::RefreshImportJidWorker).to receive(:perform_in_the_future).with(project.id, 'jid')
 
       worker.try_import(client, project)
     end
@@ -153,7 +157,7 @@ def self.name
     it 'reschedules the worker if RateLimitError was raised' do
       client = double(:client, rate_limit_resets_in: 10)
 
-      expect(project.import_state).to receive(:refresh_jid_expiration)
+      expect(Gitlab::GithubImport::RefreshImportJidWorker).to receive(:perform_in_the_future).with(project.id, 'jid')
 
       expect(worker)
         .to receive(:import)
@@ -186,7 +190,7 @@ def self.name
     end
   end
 
-  describe '.resumes_work_when_interrupted!' do
+  describe '.sidekiq_options!' do
     subject(:sidekiq_options) { worker.class.sidekiq_options }
 
     it 'does not set the `max_retries_after_interruption` if not called' do
@@ -198,5 +202,9 @@ def self.name
 
       is_expected.to include('max_retries_after_interruption' => 20)
     end
+
+    it 'sets the status_expiration' do
+      is_expected.to include('status_expiration' => Gitlab::Import::StuckImportJob::IMPORT_JOBS_EXPIRATION)
+    end
   end
 end

spec/workers/gitlab/bitbucket_import/advance_stage_worker_spec.rb

@@ -53,7 +53,7 @@
 
       it 'schedules the next stage' do
         expect(import_state)
-          .to receive(:refresh_jid_expiration)
+          .to receive(:refresh_jid_expiration).twice
 
         expect(Gitlab::BitbucketImport::Stage::FinishImportWorker)
           .to receive(:perform_async)

spec/workers/gitlab/github_import/refresh_import_jid_worker_spec.rb

@@ -9,7 +9,7 @@
     it 'schedules a job in the future' do
       expect(described_class)
         .to receive(:perform_in)
-        .with(1.minute.to_i, 10, '123')
+        .with(5.minutes.to_i, 10, '123')
 
       described_class.perform_in_the_future(10, '123')
     end
@@ -33,15 +33,20 @@
         allow(worker)
           .to receive(:find_import_state)
           .with(project.id)
-          .and_return(project)
+          .and_return(import_state)
 
         expect(Gitlab::SidekiqStatus)
           .to receive(:running?)
           .with('123')
           .and_return(true)
 
-        expect(project)
-          .to receive(:refresh_jid_expiration)
+        expect(Gitlab::SidekiqStatus)
+          .to receive(:expire)
+          .with('123', Gitlab::Import::StuckImportJob::IMPORT_JOBS_EXPIRATION)
+
+        expect(Gitlab::SidekiqStatus)
+          .to receive(:set)
+          .with(import_state.jid, Gitlab::Import::StuckImportJob::IMPORT_JOBS_EXPIRATION)
 
         expect(worker.class)
           .to receive(:perform_in_the_future)
@@ -63,8 +68,11 @@
           .with('123')
           .and_return(false)
 
-        expect(project)
-          .not_to receive(:refresh_jid_expiration)
+        expect(Gitlab::SidekiqStatus)
+          .not_to receive(:expire)
+
+        expect(Gitlab::SidekiqStatus)
+          .not_to receive(:set)
 
         worker.perform(project.id, '123')
       end

spec/workers/gitlab/github_import/stage/import_repository_worker_spec.rb

@@ -10,16 +10,6 @@
   it_behaves_like Gitlab::GithubImport::StageMethods
 
   describe '#import' do
-    before do
-      expect(Gitlab::GithubImport::RefreshImportJidWorker)
-        .to receive(:perform_in_the_future)
-        .with(project.id, '123')
-
-      expect(worker)
-        .to receive(:jid)
-        .and_return('123')
-    end
-
     context 'when the import succeeds' do
       context 'with issues' do
         it 'schedules the importing of the base data' do