Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Feat: name scheduler threads + redirect error logging (#102)
* use a custom Scheduler sub-class - redirecting potential logs to LS logger - setting thread name for scheduler threads - any worker threads started are also named * Refactor: include cause when warning from exception * force testing against rufus-scheduler 3.0.9
- Loading branch information
Showing
7 changed files
with
177 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,111 @@ | ||
require 'rufus/scheduler' | ||
|
||
require 'logstash/util/loggable' | ||
|
||
module LogStash module PluginMixins module Jdbc | ||
class Scheduler < Rufus::Scheduler | ||
|
||
include LogStash::Util::Loggable | ||
|
||
# Rufus::Scheduler >= 3.4 moved the Time impl into a gem EoTime = ::EtOrbi::EoTime` | ||
# Rufus::Scheduler 3.1 - 3.3 using it's own Time impl `Rufus::Scheduler::ZoTime` | ||
TimeImpl = defined?(Rufus::Scheduler::EoTime) ? Rufus::Scheduler::EoTime : | ||
(defined?(Rufus::Scheduler::ZoTime) ? Rufus::Scheduler::ZoTime : ::Time) | ||
|
||
# @overload | ||
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 | ||
|
||
def debug_format_time(time) | ||
# EtOrbi::EoTime used by (newer) Rufus::Scheduler has to_debug_s https://git.io/JyiPj | ||
time.respond_to?(:to_debug_s) ? time.to_debug_s : time.strftime("%Y-%m-%dT%H:%M:%S.%L") | ||
end | ||
private :debug_format_time | ||
|
||
# @private helper used by JobDecorator | ||
def work_thread_name_prefix | ||
( @opts[:thread_name] || "#{@thread_key}_scheduler" ) + '_worker-' | ||
end | ||
|
||
protected | ||
|
||
# @overload | ||
def start | ||
ret = super() # @thread[:name] = @opts[:thread_name] || "#{@thread_key}_scheduler" | ||
|
||
# at least set thread.name for easier thread dump analysis | ||
if @thread.is_a?(Thread) && @thread.respond_to?(:name=) | ||
@thread.name = @thread[:name] if @thread[:name] | ||
end | ||
|
||
ret | ||
end | ||
|
||
# @overload | ||
def do_schedule(job_type, t, callable, opts, return_job_instance, block) | ||
job_or_id = super | ||
|
||
job_or_id.extend JobDecorator if return_job_instance | ||
|
||
job_or_id | ||
end | ||
|
||
module JobDecorator | ||
|
||
def start_work_thread | ||
prev_thread_count = @scheduler.work_threads.size | ||
|
||
ret = super() # does not return Thread instance in 3.0 | ||
|
||
work_threads = @scheduler.work_threads | ||
while prev_thread_count == work_threads.size # very unlikely | ||
Thread.pass | ||
work_threads = @scheduler.work_threads | ||
end | ||
|
||
work_thread_name_prefix = @scheduler.work_thread_name_prefix | ||
|
||
work_threads.sort! do |t1, t2| | ||
if t1[:name].nil? | ||
t2[:name].nil? ? 0 : +1 # nils at the end | ||
elsif t2[:name].nil? | ||
t1[:name].nil? ? 0 : -1 | ||
else | ||
t1[:name] <=> t2[:name] | ||
end | ||
end | ||
|
||
work_threads.each_with_index do |thread, i| | ||
unless thread[:name] | ||
thread[:name] = "#{work_thread_name_prefix}#{sprintf('%02i', i)}" | ||
thread.name = thread[:name] if thread.respond_to?(:name=) | ||
# e.g. "[oracle]<jdbc_scheduler_worker-00" | ||
end | ||
end | ||
|
||
ret | ||
end | ||
|
||
end | ||
|
||
end | ||
end end end |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,52 @@ | ||
# encoding: utf-8 | ||
require "logstash/devutils/rspec/spec_helper" | ||
require "logstash/plugin_mixins/jdbc/scheduler" | ||
|
||
describe LogStash::PluginMixins::Jdbc::Scheduler do | ||
|
||
let(:thread_name) { '[test]<jdbc_scheduler' } | ||
|
||
let(:opts) do | ||
{ :max_work_threads => 2, :thread_name => thread_name } | ||
end | ||
|
||
subject(:scheduler) { LogStash::PluginMixins::Jdbc::Scheduler.new(opts) } | ||
|
||
after { scheduler.stop(:wait) } | ||
|
||
it "sets scheduler thread name" do | ||
expect( scheduler.thread.name ).to include thread_name | ||
end | ||
|
||
context 'cron schedule' do | ||
|
||
before do | ||
scheduler.schedule_cron('* * * * * *') { sleep 1.25 } # every second | ||
end | ||
|
||
it "sets worker thread names" do | ||
sleep 3.0 | ||
threads = scheduler.work_threads | ||
threads.sort! { |t1, t2| (t1.name || '') <=> (t2.name || '') } | ||
|
||
expect( threads.size ).to eql 2 | ||
expect( threads.first.name ).to eql "#{thread_name}_worker-00" | ||
expect( threads.last.name ).to eql "#{thread_name}_worker-01" | ||
end | ||
|
||
end | ||
|
||
context 'every 1s' do | ||
|
||
before do | ||
scheduler.schedule_in('1s') { raise 'TEST' } # every second | ||
end | ||
|
||
it "logs errors handled" do | ||
expect( scheduler.logger ).to receive(:error).with /Scheduler intercepted an error/, hash_including(:message => 'TEST') | ||
sleep 1.5 | ||
end | ||
|
||
end | ||
|
||
end |