Class: Gitlab::SidekiqDaemon::Monitor

Inherits:
Daemon
  • Object
show all
Extended by:
Utils::Override
Includes:
Utils::StrongMemoize
Defined in:
lib/gitlab/sidekiq_daemon/monitor.rb

Constant Summary collapse

NOTIFICATION_CHANNEL = 
'sidekiq:cancel:notifications'
CANCEL_DEADLINE = 
24.hours.seconds
RECONNECT_TIME = 
3.seconds
CancelledError =

We use exception derived from `Exception` to consider this as an very low-level exception that should not be caught by application

Class.new(Exception)

Instance Attribute Summary collapse

Attributes inherited from Daemon

#thread

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Utils::Override

extended, extensions, included, method_added, override, prepended, queue_verification, verify!

Methods included from Utils::StrongMemoize

#clear_memoization, #strong_memoize, #strong_memoized?

Methods inherited from Daemon

#enabled?, initialize_instance, instance, #start, #stop, #thread?

Constructor Details

#initialize ⇒ Monitor

Returns a new instance of Monitor.


21
22
23
24
25
26
 
# File 'lib/gitlab/sidekiq_daemon/monitor.rb', line 21

def initialize
  super

  @jobs = {}
  @jobs_mutex = Mutex.new
end

Instance Attribute Details

#jobs ⇒ Object (readonly)

Returns the value of attribute jobs.


18
19
20
 
# File 'lib/gitlab/sidekiq_daemon/monitor.rb', line 18

def jobs
  @jobs
end

#jobs_mutex ⇒ Object (readonly)

Returns the value of attribute jobs_mutex.


19
20
21
 
# File 'lib/gitlab/sidekiq_daemon/monitor.rb', line 19

def jobs_mutex
  @jobs_mutex
end

Class Method Details

.cancel_job(jid) ⇒ Object 


56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/gitlab/sidekiq_daemon/monitor.rb', line 56

def self.cancel_job(jid)
  payload = {
    action: 'cancel',
    jid: jid
  }.to_json

  ::Gitlab::Redis::SharedState.with do |redis|
    redis.setex(cancel_job_key(jid), CANCEL_DEADLINE, 1)
    redis.publish(NOTIFICATION_CHANNEL, payload)
  end
end

Instance Method Details

#thread_name ⇒ Object 


29
30
31
 
# File 'lib/gitlab/sidekiq_daemon/monitor.rb', line 29

def thread_name
  "job_monitor"
end

#within_job(worker_class, jid, queue) ⇒ Object 


33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/gitlab/sidekiq_daemon/monitor.rb', line 33

def within_job(worker_class, jid, queue)
  jobs_mutex.synchronize do
    jobs[jid] = { worker_class: worker_class, thread: Thread.current, started_at: Gitlab::Metrics::System.monotonic_time }
  end

  if cancelled?(jid)
    Sidekiq.logger.warn(
      class: self.class.to_s,
      action: 'run',
      queue: queue,
      jid: jid,
      canceled: true
    )
    raise CancelledError
  end

  yield
ensure
  jobs_mutex.synchronize do
    jobs.delete(jid)
  end
end