fluke/devops-service/workers/worker.rb
2015-11-17 16:21:28 +03:00

144 lines
3.3 KiB
Ruby

root = File.join(File.dirname(__FILE__), "..")
$LOAD_PATH.push root unless $LOAD_PATH.include? root
require "sidekiq"
require "sidekiq/api"
require "fileutils"
require "core/devops-service"
require "core/devops-config"
require "core/devops-logger"
require "core/devops-db"
require "providers/provider_factory"
require "lib/knife/knife_factory"
# All options keys MUST be a symbol!!!
class Worker
include Sidekiq::Worker
attr_accessor :out
module STATUS
INIT = "init"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
IN_QUEUE = "queued"
end
def self.start_async(worker_class, job_options)
jid = worker_class.perform_async(job_options.dup)
Worker.set_status jid, Worker::STATUS::IN_QUEUE
DevopsLogger.logger.info "Job '#{jid}' has been queued"
jid
=begin
uri = URI.parse(request.url)
uri.path = "#{DevopsConfig[:url_prefix]}/v2.0/report/#{jid}"
uri.to_s
=end
end
def self.start_sync(worker_class, request, job_options, out)
stringified_options = job_options
=begin
job_options.each do |key, value|
stringified_options[key.to_s] = value
end
=end
w = worker_class.new
w.out = out
w.perform(stringified_options)
end
def self.set_status id, status
Sidekiq.redis {|con| con.hset "devops", id, status}
end
def call &block
begin
initialize_devops()
if jid
call_async(&block)
else
call_sync(&block)
end
rescue StandardError => e
DevopsLogger.logger.error "#{e.message}:\n#{e.backtrace.join("\n")}"
end
end
private
def initialize_devops()
DevopsLogger.logger = logger
DevopsConfig.read
DevopsService.init
::Provider::ProviderFactory.init(DevopsConfig.config)
end
# outputs to file
def call_async()
dir = DevopsConfig[:report_dir_v2]
# directory is created on server start in config.ru
file = File.join(dir, jid)
update_job_status(STATUS::INIT, nil)
File.open(file, "w") do |out|
begin
update_job_status(STATUS::RUNNING, nil)
self.out = out
job_result = yield(out, file)
out.puts "TODO: job_result: #{job_result}"
canonical_status = (job_result == 0 ? STATUS::COMPLETED : STATUS::FAILED)
out.puts "TODO: status: #{canonical_status}"
out.flush
DevopsLogger.logger.info "TODO: res: #{job_result} status: #{canonical_status}"
update_job_status(canonical_status, job_result)
rescue StandardError, RecordNotFound => e
out << "\n #{e.class}\n #{e.message}\n"
out << e.backtrace.join("\n")
update_job_status(STATUS::FAILED, -100)
end
end
end
# outputs to STDOUT
def call_sync()
out = STDOUT
begin
yield(out, '')
rescue StandardError, RecordNotFound => e
out << "\n"
out << e.message
out << "\n"
out << e.backtrace.join("\n")
end
end
def mongo
::Devops::Db.connector
end
def update_job_status(status, job_result_code)
Worker.set_status(jid, status)
mongo.set_report_status(jid, status, job_result_code)
status
end
def convert_config conf
config = {}
conf.each {|k,v| config[k.is_a?(String) ? k.to_sym : k] = v}
config
end
=begin
def set_status id, status
self.class.set_status(id, status)
end
=end
end