Class: LogStash::PluginMixins::Scheduler::RufusImpl::SchedulerImpl

Inherits:
Rufus::Scheduler
  • Object
show all
Defined in:
lib/logstash/plugin_mixins/scheduler/rufus_impl.rb

Constant Summary collapse

TimeImpl =

Rufus::Scheduler >= 3.4 moved the Time impl into a gem EoTime = ::EtOrbi::EoTimeRufus::Scheduler 3.1 - 3.3 using it's own Time implRufus::Scheduler::ZoTime`

defined?(::Rufus::Scheduler::EoTime) ? ::Rufus::Scheduler::EoTime :
(defined?(::Rufus::Scheduler::ZoTime) ? ::Rufus::Scheduler::ZoTime : ::Time)

Instance Method Summary collapse

Constructor Details

#initialize(opts, logger) ⇒ SchedulerImpl

Returns a new instance of SchedulerImpl.



78
79
80
81
 
# File 'lib/logstash/plugin_mixins/scheduler/rufus_impl.rb', line 78

def initialize(opts, logger)
  super(opts)
  @_logger = logger
end

Instance Method Details

#join ⇒ Object 



84
85
86
87
88
 
# File 'lib/logstash/plugin_mixins/scheduler/rufus_impl.rb', line 84

def join
  fail NotRunningError.new('cannot join scheduler that is not running') unless @thread
  fail ThreadError.new('scheduler thread cannot join itself') if @thread == Thread.current
  @thread.join # makes scheduler.join behavior consistent across 3.x versions
end

#on_error(job, err) ⇒ Object 



137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/logstash/plugin_mixins/scheduler/rufus_impl.rb', line 137

def on_error(job, err)
  details = { exception: err.class, message: err.message, backtrace: err.backtrace }
  details[:cause] = err.cause if err.cause

  details[:now] = debug_format_time(TimeImpl.now)
  details[:last_time] = (debug_format_time(job.last_time) rescue nil)
  details[:next_time] = (debug_format_time(job.next_time) rescue nil)
  details[:job] = job

  details[:opts] = @opts
  details[:started_at] = started_at
  details[:thread] = thread.inspect
  details[:jobs_size] = @jobs.size
  details[:work_threads_size] = work_threads.size
  details[:work_queue_size] = work_queue.size

  logger.error("Scheduler intercepted an error:", details)

rescue => e
  logger.error("Scheduler failed in #on_error #{e.inspect}")
end

#shutdown(opt = nil) ⇒ Object 



91
92
93
94
95
96
97
98
 
# File 'lib/logstash/plugin_mixins/scheduler/rufus_impl.rb', line 91

def shutdown(opt=nil)
  if @thread # do not fail when scheduler thread failed to start
    super(opt)
  else
    @started_at = nil # bare minimum to look like the scheduler is down
    # when the scheduler thread fails `@started_at = ...` might still be set
  end
end

#timeout_jobs ⇒ Object 



101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/logstash/plugin_mixins/scheduler/rufus_impl.rb', line 101

def timeout_jobs
  # Rufus relies on `Thread.list` which is a blocking operation and with many schedulers
  # (and threads) within LS will have a negative impact on performance as scheduler
  # threads will end up waiting to obtain the `Thread.list` lock.
  #
  # However, this isn't necessary we can easily detect whether there are any jobs
  # that might need to timeout: only when `@opts[:timeout]` is set causes worker thread(s)
  # to have a `Thread.current[:rufus_scheduler_timeout]` that is not nil
  return unless @opts[:timeout]
  super
end

#work_thread_name_prefix ⇒ Object 



166
167
168
 
# File 'lib/logstash/plugin_mixins/scheduler/rufus_impl.rb', line 166

def work_thread_name_prefix
  ( @opts[:thread_name] || "#{@thread_key}_scheduler" ) + '_worker-'
end

#work_threads(query = :all) ⇒ Object 



114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/logstash/plugin_mixins/scheduler/rufus_impl.rb', line 114

def work_threads(query = :all)
  if query == :__all_no_cache__ # special case from JobDecorator#start_work_thread
    @_work_threads = nil # when a new worker thread is being added reset
    return super(:all)
  end

  # Gets executed every time a job is triggered, we're going to cache the
  # worker threads for this scheduler (to avoid `Thread.list`) - they only
  # change when a new thread is being started from #start_work_thread ...
  work_threads = @_work_threads
  if work_threads.nil?
    work_threads = threads.select { |t| t[:rufus_scheduler_work_thread] }
    @_work_threads = work_threads
  end

  case query
  when :active then work_threads.select { |t| t[:rufus_scheduler_job] }
  when :vacant then work_threads.reject { |t| t[:rufus_scheduler_job] }
  else work_threads
  end
end