Module: Gitlab::SidekiqCluster

Defined in:
lib/gitlab/sidekiq_cluster.rb,
lib/gitlab/sidekiq_cluster/cli.rb

Defined Under Namespace

Classes: CLI

Constant Summary collapse

TERMINATE_SIGNALS =

The signals that should terminate both the master and workers.

%i(INT TERM).freeze
FORWARD_SIGNALS =

The signals that should simply be forwarded to the workers.

%i(TTIN USR1 USR2 HUP).freeze

Class Method Summary collapse

Class Method Details

.all_alive?(pids) ⇒ Boolean

Returns true if all the processes are alive.

Returns:

  • (Boolean)

143
144
145
146
147
148
149
# File 'lib/gitlab/sidekiq_cluster.rb', line 143

def self.all_alive?(pids)
  pids.each do |pid|
    return false unless process_alive?(pid)
  end

  true
end

.any_alive?(pids) ⇒ Boolean

Returns:

  • (Boolean)

151
152
153
# File 'lib/gitlab/sidekiq_cluster.rb', line 151

def self.any_alive?(pids)
  pids_alive(pids).any?
end

.concurrency(queues, min_concurrency, max_concurrency) ⇒ Object


127
128
129
130
131
132
133
# File 'lib/gitlab/sidekiq_cluster.rb', line 127

def self.concurrency(queues, min_concurrency, max_concurrency)
  concurrency_from_queues = queues.length + 1
  max = max_concurrency > 0 ? max_concurrency : concurrency_from_queues
  min = [min_concurrency, max].min

  concurrency_from_queues.clamp(min, max)
end

.count_by_queue(queues) ⇒ Object


113
114
115
# File 'lib/gitlab/sidekiq_cluster.rb', line 113

def self.count_by_queue(queues)
  queues.each_with_object(Hash.new(0)) { |element, hash| hash[element] += 1 }
end

.pids_alive(pids) ⇒ Object


155
156
157
# File 'lib/gitlab/sidekiq_cluster.rb', line 155

def self.pids_alive(pids)
  pids.select { |pid| process_alive?(pid) }
end

.proc_details(counts) ⇒ Object


117
118
119
120
121
122
123
124
125
# File 'lib/gitlab/sidekiq_cluster.rb', line 117

def self.proc_details(counts)
  counts.map do |queue, count|
    if count == 1
      queue
    else
      "#{queue} (#{count})"
    end
  end.join(',')
end

.process_alive?(pid) ⇒ Boolean

Returns:

  • (Boolean)

159
160
161
162
163
# File 'lib/gitlab/sidekiq_cluster.rb', line 159

def self.process_alive?(pid)
  # Signal 0 tests whether the process exists and we have access to send signals
  # but is otherwise a noop (doesn't actually send a signal to the process)
  signal(pid, 0)
end

.signal(pid, signal) ⇒ Object


39
40
41
42
43
44
# File 'lib/gitlab/sidekiq_cluster.rb', line 39

def self.signal(pid, signal)
  Process.kill(signal, pid)
  true
rescue Errno::ESRCH
  false
end

.signal_processes(pids, signal) ⇒ Object


46
47
48
# File 'lib/gitlab/sidekiq_cluster.rb', line 46

def self.signal_processes(pids, signal)
  pids.each { |pid| signal(pid, signal) }
end

.start(queues, env: :development, directory: Dir.pwd, max_concurrency: 50, min_concurrency: 0, timeout: CLI::DEFAULT_SOFT_TIMEOUT_SECONDS, dryrun: false) ⇒ Object

Starts Sidekiq workers for the pairs of processes.

Example:

start([ ['foo'], ['bar', 'baz'] ], :production)

This would start two Sidekiq processes: one processing “foo”, and one processing “bar” and “baz”. Each one is placed in its own process group.

queues - An Array containing Arrays. Each sub Array should specify the

queues to use for a single process.

directory - The directory of the Rails application.

Returns an Array containing the PIDs of the started processes.


65
66
67
68
69
70
71
72
73
74
75
# File 'lib/gitlab/sidekiq_cluster.rb', line 65

def self.start(queues, env: :development, directory: Dir.pwd, max_concurrency: 50, min_concurrency: 0, timeout: CLI::DEFAULT_SOFT_TIMEOUT_SECONDS, dryrun: false)
  queues.map.with_index do |pair, index|
    start_sidekiq(pair, env: env,
                        directory: directory,
                        max_concurrency: max_concurrency,
                        min_concurrency: min_concurrency,
                        worker_id: index,
                        timeout: timeout,
                        dryrun: dryrun)
  end
end

.start_sidekiq(queues, env:, directory:, max_concurrency:, min_concurrency:, worker_id:, timeout:, dryrun:) ⇒ Object

Starts a Sidekiq process that processes only the given queues.

Returns the PID of the started process.


80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
# File 'lib/gitlab/sidekiq_cluster.rb', line 80

def self.start_sidekiq(queues, env:, directory:, max_concurrency:, min_concurrency:, worker_id:, timeout:, dryrun:)
  counts = count_by_queue(queues)

  cmd = %w[bundle exec sidekiq]
  cmd << "-c#{self.concurrency(queues, min_concurrency, max_concurrency)}"
  cmd << "-e#{env}"
  cmd << "-t#{timeout}"
  cmd << "-gqueues:#{proc_details(counts)}"
  cmd << "-r#{directory}"

  counts.each do |queue, count|
    cmd << "-q#{queue},#{count}"
  end

  if dryrun
    puts Shellwords.join(cmd) # rubocop:disable Rails/Output
    return
  end

  pid = Process.spawn(
    { 'ENABLE_SIDEKIQ_CLUSTER' => '1',
      'SIDEKIQ_WORKER_ID' => worker_id.to_s },
    *cmd,
    pgroup: true,
    err: $stderr,
    out: $stdout
  )

  wait_async(pid)

  pid
end

.trap_forward(&block) ⇒ Object


35
36
37
# File 'lib/gitlab/sidekiq_cluster.rb', line 35

def self.trap_forward(&block)
  trap_signals(FORWARD_SIGNALS, &block)
end

.trap_signals(signals) ⇒ Object

Traps the given signals and yields the block whenever these signals are received.

The block is passed the name of the signal.

Example:

trap_signals(%i(HUP TERM)) do |signal|
  ...
end

23
24
25
26
27
28
29
# File 'lib/gitlab/sidekiq_cluster.rb', line 23

def self.trap_signals(signals)
  signals.each do |signal|
    trap(signal) do
      yield signal
    end
  end
end

.trap_terminate(&block) ⇒ Object


31
32
33
# File 'lib/gitlab/sidekiq_cluster.rb', line 31

def self.trap_terminate(&block)
  trap_signals(TERMINATE_SIGNALS, &block)
end

.wait_async(pid) ⇒ Object

Waits for the given process to complete using a separate thread.


136
137
138
139
140
# File 'lib/gitlab/sidekiq_cluster.rb', line 136

def self.wait_async(pid)
  Thread.new do
    Process.wait(pid) rescue Errno::ECHILD
  end
end

.write_pid(path) ⇒ Object


165
166
167
168
169
# File 'lib/gitlab/sidekiq_cluster.rb', line 165

def self.write_pid(path)
  File.open(path, 'w') do |handle|
    handle.write(Process.pid.to_s)
  end
end