123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175 |
- class Scheduler < ApplicationModel
-
- @@jobs_started = {}
-
-
- def self.threads
- Thread.abort_on_exception = true
-
- worker
-
- loop do
- logger.info 'Scheduler running...'
-
- begin
- ActiveRecord::Base.connection.reconnect!
- rescue => e
- logger.error "Can't reconnect to database #{ e.inspect }"
- end
-
- jobs = Scheduler.where( 'active = ?', true )
- jobs.each {|job|
-
- next if @@jobs_started[ job.id ]
-
- next if job.last_run && job.period && job.last_run > ( Time.zone.now - job.period )
-
- @@jobs_started[ job.id ] = true
- start_job( job )
- }
- sleep 60
- end
- end
- def self.start_job( job )
- sleep 4
- Thread.new {
- logger.info "Started job thread for '#{job.name}' (#{job.method})..."
-
- if job.period && job.period <= 300
- loop do
- _start_job( job )
- job = Scheduler.lookup( id: job.id )
-
- break if !job
-
- break if !job.active
-
- break if !job.period
-
- sleep job.period
- end
- else
- _start_job( job )
- end
- job.pid = ''
- job.save
- logger.info " ...stopped thread for '#{job.method}'"
- ActiveRecord::Base.connection.close
-
- @@jobs_started[ job.id ] = false
- }
- end
- def self._start_job( job, try_count = 0, try_run_time = Time.zone.now )
- sleep 5
- begin
- job.last_run = Time.zone.now
- job.pid = Thread.current.object_id
- job.save
- logger.info "execute #{job.method} (try_count #{try_count})..."
- eval job.method()
- rescue => e
- logger.error "execute #{job.method} (try_count #{try_count}) exited with error #{ e.inspect }"
-
- begin
- ActiveRecord::Base.connection.reconnect!
- rescue => e
- logger.error "Can't reconnect to database #{ e.inspect }"
- end
- try_run_max = 10
- try_count += 1
-
- if try_run_time + ( 60 * 5 ) < Time.zone.now
- try_count = 0
- end
- try_run_time = Time.zone.now
-
- if try_run_max > try_count
- _start_job( job, try_count, try_run_time)
- else
- raise "STOP thread for #{job.method} after #{try_count} tries"
- end
- end
- end
- def self.worker
- wait = 8
- Thread.new {
- sleep wait
- logger.info "Starting worker thread #{Delayed::Job}"
- loop do
- result = nil
- realtime = Benchmark.realtime do
- result = Delayed::Worker.new.work_off
- end
- count = result.sum
- if count.zero?
- sleep wait
- logger.debug '*** worker thread loop'
- else
- format "*** #{count} jobs processed at %.4f j/s, %d failed ...\n", count / realtime, result.last
- end
- end
- logger.info ' ...stopped worker thread'
- ActiveRecord::Base.connection.close
- }
- end
- def self.check( name, time_warning = 10, time_critical = 20 )
- time_warning_time = Time.zone.now - time_warning.minutes
- time_critical_time = Time.zone.now - time_critical.minutes
- scheduler = Scheduler.find_by( name: name )
- if !scheduler
- puts "CRITICAL - no such scheduler jobs '#{name}'"
- return true
- end
- logger.debug scheduler.inspect
- if !scheduler.last_run
- puts "CRITICAL - scheduler jobs never started '#{name}'"
- exit 2
- end
- if scheduler.last_run < time_critical_time
- puts "CRITICAL - scheduler jobs was not running in last '#{time_critical}' minutes - last run at '#{scheduler.last_run}' '#{name}'"
- exit 2
- end
- if scheduler.last_run < time_warning_time
- puts "CRITICAL - scheduler jobs was not running in last '#{time_warning}' minutes - last run at '#{scheduler.last_run}' '#{name}'"
- exit 2
- end
- puts "ok - scheduler jobs was running at '#{scheduler.last_run}' '#{name}'"
- exit 0
- end
- end
|