require 'iron_mq' require 'fanforce/api' require 'timeout' class Fanforce::PluginWorker::Runner include Fanforce::PluginWorker::Utils MAX_EXECUTION_TIME = 3300 class Timeout < RuntimeError; end def initialize(worker_data, min_execution_time=300, &code_block) raise "min_execution_time was set to #{min_execution_time}, which is #{min_execution_time - MAX_EXECUTION_TIME} seconds too long" if min_execution_time > MAX_EXECUTION_TIME log.debug '------------------------------------------------------------------------------------' log.debug 'LOADING WORKER ENV' @queue_id = worker_data['queue_id'] || (raise 'worker_data must contain queue_id') @worker_env = worker_data['env_vars'] || {} @min_execution_time = min_execution_time @code_block = code_block load_env_from_server load_env_from_worker(@worker_env) ENV.each {|k,v| log.debug "#{k} = #{v}" } load_jobs end def load_jobs log.debug '------------------------------------------------------------------------------------' log.debug 'PROCESSING JOBS...' log.debug '------------------------------------------------------------------------------------' job_num = 0 job_data = nil while job_has_enough_time_to_run and (job = Fanforce::PluginWorker.iron_mq.queue(iron_queue_id(@queue_id)).get(timeout: 3600)) do log.debug "- JOB #{job_num+=1}: #{job.body}" timeout(worker_time_remaining, Timeout) do job_data = nil job_data = MultiJson.load(job.body, symbolize_keys: true) run_job(job, job_data, &@code_block) end delete_job log.debug '------------------------------------------------------------------------------------' end delete_job log.debug 'WINDING DOWN WORKER!' rescue Exception => e handle_job_loading_error(e, job, job_data) end def load_env_from_server if File.exists?('.developmentenv.rb') require '.developmentenv' elsif File.exists?('.stagingenv.rb') require '.stagingenv' elsif File.exists?('.productionenv.rb') require '.productionenv' end end def load_env_from_worker(vars) vars.each {|k,v| ENV[k.to_s]=v } end def run_job(job, job_data, &code_block) @current_job = job @current_params = job_data[:params] @current_retries = job_data[:retries] code_block.call(job_data[:params].clone, retries: job_data[:retries], queue_id: @queue_id) delete_job(job) rescue Exception => e handle_job_error(e, job, job_data) end def handle_job_loading_error(e, job, job_data) raise($!, "#{$!}: THERE IS NO JOB", $!.backtrace) if job.nil? delete_job(job) log.debug 'REMOVED JOB FROM QUEUE, BUT COULD NOT SAVE TO ERROR CACHE...' raise($!, "#{$!}: #{job_data.to_json}", $!.backtrace) end def handle_job_error(e, job, job_data) raise($!, "#{$!}: THERE IS NO JOB", $!.backtrace) if job.nil? delete_job(job) require_relative 'errors' log.debug 'REMOVED JOB FROM QUEUE, AND SAVING TO ERROR CACHE...' Fanforce::PluginWorker::Errors.add(@queue_id, e, job_data, @worker_env) end def worker_time_remaining time_since_load = Time.now - Fanforce::PluginWorker::LOADED_AT MAX_EXECUTION_TIME - time_since_load end def job_has_enough_time_to_run time_since_load = Time.now - Fanforce::PluginWorker::LOADED_AT return false if time_since_load > MAX_EXECUTION_TIME return false if worker_time_remaining < @min_execution_time return true end def delete_job(job=nil) return if job.nil? and @current_job.nil? (job || @current_job).delete rescue Exception => e log.debug "Job could not be deleted: #{e.message}" ensure @current_job = nil end end