Class: Gitlab::JobWaiter

Inherits:
Object
  • Object
show all
Defined in:
lib/gitlab/job_waiter.rb

Overview

JobWaiter can be used to wait for a number of Sidekiq jobs to complete.

Its use requires the cooperation of the sidekiq jobs themselves. Set up the waiter, then start the jobs, passing them its ‘key`. Their `perform` methods should look like:

def perform(args, notify_key)
  # do work
ensure
  ::Gitlab::JobWaiter.notify(notify_key, jid)
end

The JobWaiter blocks popping items from a Redis array. All the sidekiq jobs push to that array when done. Once the waiter has popped ‘count` items, it knows all the jobs are done.

Constant Summary collapse

KEY_PREFIX = 
"gitlab:job_waiter"
STARTED_METRIC = 
:gitlab_job_waiter_started_total
TIMEOUTS_METRIC = 
:gitlab_job_waiter_timeouts_total
DEFAULT_TTL =

This TTL needs to be long enough to allow whichever Sidekiq job calls JobWaiter#wait to reach BLPOP.

6.hours.to_i

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(jobs_remaining = 0, key = "#{KEY_PREFIX}:#{SecureRandom.uuid}", worker_label: nil) ⇒ JobWaiter

jobs_remaining - the number of jobs left to wait for key - The key of this waiter.



56
57
58
59
60
61
 
# File 'lib/gitlab/job_waiter.rb', line 56

def initialize(jobs_remaining = 0, key = "#{KEY_PREFIX}:#{SecureRandom.uuid}", worker_label: nil)
  @key = key
  @jobs_remaining = jobs_remaining
  @finished = []
  @worker_label = worker_label
end

Instance Attribute Details

#finished ⇒ Object (readonly)

Returns the value of attribute finished.



51
52
53
 
# File 'lib/gitlab/job_waiter.rb', line 51

def finished
  @finished
end

#jobs_remaining ⇒ Object

Returns the value of attribute jobs_remaining.



52
53
54
 
# File 'lib/gitlab/job_waiter.rb', line 52

def jobs_remaining
  @jobs_remaining
end

#key ⇒ Object (readonly)

Returns the value of attribute key.



51
52
53
 
# File 'lib/gitlab/job_waiter.rb', line 51

def key
  @key
end

#worker_label ⇒ Object (readonly)

Returns the value of attribute worker_label.



51
52
53
 
# File 'lib/gitlab/job_waiter.rb', line 51

def worker_label
  @worker_label
end

Class Method Details

.delete_key(key) ⇒ Object 



47
48
49
 
# File 'lib/gitlab/job_waiter.rb', line 47

def self.delete_key(key)
  Gitlab::Redis::SharedState.with { |redis| redis.del(key) } if key?(key)
end

.generate_key ⇒ Object 



43
44
45
 
# File 'lib/gitlab/job_waiter.rb', line 43

def self.generate_key
  "#{KEY_PREFIX}:#{SecureRandom.uuid}"
end

.key?(key) ⇒ Boolean

Returns:

  • (Boolean) 


39
40
41
 
# File 'lib/gitlab/job_waiter.rb', line 39

def self.key?(key)
  key.is_a?(String) && key =~ /\A#{KEY_PREFIX}:\h{8}-\h{4}-\h{4}-\h{4}-\h{12}\z/o
end

.notify(key, jid, ttl: DEFAULT_TTL) ⇒ Object 



29
30
31
32
33
34
35
36
37
 
# File 'lib/gitlab/job_waiter.rb', line 29

def self.notify(key, jid, ttl: DEFAULT_TTL)
  Gitlab::Redis::SharedState.with do |redis|
    # Use a Redis MULTI transaction to ensure we always set an expiry
    redis.multi do |multi|
      multi.lpush(key, jid)
      multi.expire(key, ttl)
    end
  end
end

Instance Method Details

#wait(timeout = 10) ⇒ Object

Waits for all the jobs to be completed.

timeout - The maximum amount of seconds to block the caller for. This

ensures we don't indefinitely block a caller in case a job takes
long to process, or is never processed.



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/gitlab/job_waiter.rb', line 68

def wait(timeout = 10)
  deadline = Time.now.utc + timeout
  increment_counter(STARTED_METRIC)

  Gitlab::Redis::SharedState.with do |redis|
    while jobs_remaining > 0
      # Redis will not take fractional seconds. Prefer waiting too long over
      # not waiting long enough
      seconds_left = (deadline - Time.now.utc).ceil

      # Redis interprets 0 as "wait forever", so skip the final `blpop` call
      break if seconds_left <= 0

      list, jid = redis.blpop(key, timeout: seconds_left)

      # timed out
      unless list && jid
        increment_counter(TIMEOUTS_METRIC)
        break
      end

      @finished << jid
      @jobs_remaining -= 1
    end
  end

  finished
end