Read about our upcoming Code of Conduct on this issue

This instance was upgraded to Heptapod 0.28.0 today

advance_stage_worker.rb 2.59 KB
Newer Older
1
2
3
4
5
6
7
8
9
# frozen_string_literal: true

module Gitlab
  module GithubImport
    # AdvanceStageWorker is a worker used by the GitHub importer to wait for a
    # number of jobs to complete, without blocking a thread. Once all jobs have
    # been completed this worker will advance the import process to the next
    # stage.
    class AdvanceStageWorker
10
      include ApplicationWorker
11

12
      sidekiq_options dead: false
13
14
15
16

      INTERVAL = 30.seconds.to_i

      # The number of seconds to wait (while blocking the thread) before
17
      # continuing to the next waiter.
18
19
20
21
22
23
      BLOCKING_WAIT_TIME = 5

      # The known importer stages and their corresponding Sidekiq workers.
      STAGES = {
        issues_and_diff_notes: Stage::ImportIssuesAndDiffNotesWorker,
        notes: Stage::ImportNotesWorker,
24
        lfs_objects: Stage::ImportLfsObjectsWorker,
25
26
27
28
29
30
31
32
33
        finish: Stage::FinishImportWorker
      }.freeze

      # project_id - The ID of the project being imported.
      # waiters - A Hash mapping Gitlab::JobWaiter keys to the number of
      #           remaining jobs.
      # next_stage - The name of the next stage to start when all jobs have been
      #              completed.
      def perform(project_id, waiters, next_stage)
34
        return unless import_state = find_import_state(project_id)
35
36
37
38
39
40
41
42
43

        new_waiters = wait_for_jobs(waiters)

        if new_waiters.empty?
          # We refresh the import JID here so workers importing individual
          # resources (e.g. notes) don't have to do this all the time, reducing
          # the pressure on Redis. We _only_ do this once all jobs are done so
          # we don't get stuck forever if one or more jobs failed to notify the
          # JobWaiter.
44
          import_state.refresh_jid_expiration
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

          STAGES.fetch(next_stage.to_sym).perform_async(project_id)
        else
          self.class.perform_in(INTERVAL, project_id, new_waiters, next_stage)
        end
      end

      def wait_for_jobs(waiters)
        waiters.each_with_object({}) do |(key, remaining), new_waiters|
          waiter = JobWaiter.new(remaining, key)

          # We wait for a brief moment of time so we don't reschedule if we can
          # complete the work fast enough.
          waiter.wait(BLOCKING_WAIT_TIME)

          next unless waiter.jobs_remaining.positive?

          new_waiters[waiter.key] = waiter.jobs_remaining
        end
      end

66
      # rubocop: disable CodeReuse/ActiveRecord
67
68
      def find_import_state(project_id)
        ProjectImportState.select(:jid).with_status(:started).find_by(project_id: project_id)
69
      end
70
      # rubocop: enable CodeReuse/ActiveRecord
71
72
73
    end
  end
end