#!/usr/bin/env ruby
+require 'trollop'
+
include Process
$warned = {}
end
end
+$trollopts = Trollop::options do
+ opt :use_env, "Pass selected environment variables (PATH, PYTHONPATH, RUBYLIB, GEM_PATH, PERLLIB) to crunch-job"
+end
+
ENV["RAILS_ENV"] = ARGV[0] || ENV["RAILS_ENV"] || "development"
require File.dirname(__FILE__) + '/../config/boot'
require File.dirname(__FILE__) + '/../config/environment'
require 'open3'
-$redis ||= Redis.new
-LOG_BUFFER_SIZE = 2**20
+LOG_BUFFER_SIZE = 4096
class Dispatcher
include ApplicationHelper
return act_as_system_user
end
+ def refresh_running
+ Job.running.each do |jobrecord|
+ if !@running[jobrecord.uuid]
+ f = Log.where("object_uuid=?", jobrecord.uuid).limit(1).order("created_at desc").first
+ age = (Time.now - f.created_at)
+ if age > 300
+ $stderr.puts "dispatch: failing orphan job #{jobrecord.uuid}, last log is #{age} seconds old"
+ # job is marked running, but not known to crunch-dispatcher, and
+ # hasn't produced any log entries for 5 minutes, so mark it as failed.
+ jobrecord.running = false
+ jobrecord.finished_at ||= Time.now
+ if jobrecord.success.nil?
+ jobrecord.success = false
+ end
+ jobrecord.save!
+ end
+ end
+ end
+ end
+
def refresh_todo
@todo = Job.queue.select do |j| j.repository end
@todo_pipelines = PipelineInstance.queue
end
if Server::Application.config.crunch_job_user
- cmd_args.unshift("sudo", "-E", "-u",
- Server::Application.config.crunch_job_user,
- "PERLLIB=#{ENV['PERLLIB']}")
+ cmd_args.unshift("sudo", "-E", "-u", Server::Application.config.crunch_job_user)
+ end
+
+ cmd_args << "HOME=/dev/null"
+ cmd_args << "ARVADOS_API_HOST=#{ENV['ARVADOS_API_HOST']}"
+ cmd_args << "ARVADOS_API_HOST_INSECURE=#{ENV['ARVADOS_API_HOST_INSECURE']}" if ENV['ARVADOS_API_HOST_INSECURE']
+
+ ENV.each do |k, v|
+ cmd_args << "#{k}=#{v}" if k.starts_with? "CRUNCH_"
+ end
+
+ if $trollopts.use_env
+ cmd_args << "PATH=#{ENV['PATH']}"
+ cmd_args << "PYTHONPATH=#{ENV['PYTHONPATH']}"
+ cmd_args << "PERLLIB=#{ENV['PERLLIB']}"
+ cmd_args << "RUBYLIB=#{ENV['RUBYLIB']}"
+ cmd_args << "GEM_PATH=#{ENV['GEM_PATH']}"
end
job_auth = ApiClientAuthorization.
cmd_args << '--git-dir'
cmd_args << arvados_internal
- $stderr.puts "dispatch: #{cmd_args.join ' '}"
+ $stderr.puts "dispatch: #{cmd_args}"
begin
- i, o, e, t = Open3.popen3(*cmd_args)
+ i, o, e, t = Open3.popen3({}, *cmd_args, { :unsetenv_others => true})
rescue
$stderr.puts "dispatch: popen3: #{$!}"
sleep 1
$stderr.puts "dispatch: job #{job.uuid}"
start_banner = "dispatch: child #{t.pid} start #{Time.now.ctime.to_s}"
$stderr.puts start_banner
- $redis.set job.uuid, start_banner + "\n"
- $redis.publish job.uuid, start_banner
- $redis.publish job.owner_uuid, start_banner
@running[job.uuid] = {
stdin: i,
stderr_buf: '',
started: false,
sent_int: 0,
- job_auth: job_auth
+ job_auth: job_auth,
+ stderr_buf_to_flush: '',
+ stderr_flushed_at: 0
}
i.close
end
lines.each do |line|
$stderr.print "#{job_uuid} ! " unless line.index(job_uuid)
$stderr.puts line
- pub_msg = "#{Time.now.ctime.to_s} #{line.strip}"
- $redis.publish job.owner_uuid, pub_msg
- $redis.publish job_uuid, pub_msg
- $redis.append job_uuid, pub_msg + "\n"
- if LOG_BUFFER_SIZE < $redis.strlen(job_uuid)
- $redis.set(job_uuid,
- $redis
- .getrange(job_uuid, (LOG_BUFFER_SIZE >> 1), -1)
- .sub(/^.*?\n/, ''))
- end
+ pub_msg = "#{Time.now.ctime.to_s} #{line.strip} \n"
+ j[:stderr_buf_to_flush] << pub_msg
+ end
+
+ if (LOG_BUFFER_SIZE < j[:stderr_buf_to_flush].size) || ((j[:stderr_flushed_at]+1) < Time.now.to_i)
+ write_log j
end
end
end
# Ensure every last drop of stdout and stderr is consumed
read_pipes
+ write_log j_done # write any remaining logs
+
if j_done[:stderr_buf] and j_done[:stderr_buf] != ''
$stderr.puts j_done[:stderr_buf] + "\n"
end
# Invalidate the per-job auth token
j_done[:job_auth].update_attributes expires_at: Time.now
- $redis.publish job_done.uuid, "end"
-
@running.delete job_done.uuid
end
end
end
else
+ refresh_running unless did_recently(:refresh_running, 60.0)
refresh_todo unless did_recently(:refresh_todo, 1.0)
update_node_status
unless @todo.empty? or did_recently(:start_jobs, 1.0) or $signal[:term]
true
end
end
+
+ # send message to log table. we want these records to be transient
+ def write_log running_job
+ begin
+ if (running_job && running_job[:stderr_buf_to_flush] != '')
+ log = Log.new(object_uuid: running_job[:job].uuid,
+ event_type: 'stderr',
+ owner_uuid: running_job[:job].owner_uuid,
+ properties: {"text" => running_job[:stderr_buf_to_flush]})
+ log.save!
+ running_job[:stderr_buf_to_flush] = ''
+ running_job[:stderr_flushed_at] = Time.now.to_i
+ end
+ rescue
+ running_job[:stderr_buf] = "Failed to write logs \n"
+ running_job[:stderr_buf_to_flush] = ''
+ running_job[:stderr_flushed_at] = Time.now.to_i
+ end
+ end
+
end
# This is how crunch-job child procs know where the "refresh" trigger file is