Read about our upcoming Code of Conduct on this issue

stuck_import_jobs_worker.rb 2.38 KB
Newer Older
1
2
# frozen_string_literal: true

3
class StuckImportJobsWorker
4
  include ApplicationWorker
5
6
  include CronjobQueue

7
  IMPORT_JOBS_EXPIRATION = 15.hours.to_i
8
9

  def perform
10
11
    import_state_without_jid_count = mark_import_states_without_jid_as_failed!
    import_state_with_jid_count = mark_import_states_with_jid_as_failed!
12
13

    Gitlab::Metrics.add_event(:stuck_import_jobs,
14
15
                             projects_without_jid_count: import_state_without_jid_count,
                             projects_with_jid_count: import_state_with_jid_count)
16
17
18
19
  end

  private

20
21
22
  def mark_import_states_without_jid_as_failed!
    enqueued_import_states_without_jid.each do |import_state|
      import_state.mark_as_failed(error_message)
23
24
25
    end.count
  end

26
  # rubocop: disable CodeReuse/ActiveRecord
27
28
  def mark_import_states_with_jid_as_failed!
    jids_and_ids = enqueued_import_states_with_jid.pluck(:jid, :id).to_h
29

30
31
32
    # Find the jobs that aren't currently running or that exceeded the threshold.
    completed_jids = Gitlab::SidekiqStatus.completed_jids(jids_and_ids.keys)
    return unless completed_jids.any?
33

34
    completed_import_state_ids = jids_and_ids.values_at(*completed_jids)
35

36
    # We select the import states again, because they may have transitioned from
37
    # scheduled/started to finished/failed while we were looking up their Sidekiq status.
38
    completed_import_states = enqueued_import_states_with_jid.where(id: completed_import_state_ids)
39

40
41
    completed_import_state_jids = completed_import_states.map { |import_state| import_state.jid }.join(', ')
    Rails.logger.info("Marked stuck import jobs as failed. JIDs: #{completed_import_state_jids}")
42

43
44
    completed_import_states.each do |import_state|
      import_state.mark_as_failed(error_message)
45
    end.count
46
  end
47
  # rubocop: enable CodeReuse/ActiveRecord
48

49
50
  def enqueued_import_states
    ProjectImportState.with_status([:scheduled, :started])
51
52
  end

53
  # rubocop: disable CodeReuse/ActiveRecord
54
55
  def enqueued_import_states_with_jid
    enqueued_import_states.where.not(jid: nil)
56
  end
57
  # rubocop: enable CodeReuse/ActiveRecord
58

59
  # rubocop: disable CodeReuse/ActiveRecord
60
61
  def enqueued_import_states_without_jid
    enqueued_import_states.where(jid: nil)
62
  end
63
  # rubocop: enable CodeReuse/ActiveRecord
64
65

  def error_message
66
    _("Import timed out. Import took longer than %{import_jobs_expiration} seconds") % { import_jobs_expiration: IMPORT_JOBS_EXPIRATION }
67
68
  end
end