include Process
+$options = {}
+(ARGV.any? ? ARGV : ['--jobs', '--pipelines']).each do |arg|
+ case arg
+ when '--jobs'
+ $options[:jobs] = true
+ when '--pipelines'
+ $options[:pipelines] = true
+ else
+ abort "Unrecognized command line option '#{arg}'"
+ end
+end
+if not ($options[:jobs] or $options[:pipelines])
+ abort "Nothing to do. Please specify at least one of: --jobs, --pipelines."
+end
+
+ARGV.reject! { |a| a =~ /--jobs|--pipelines/ }
+
$warned = {}
$signal = {}
%w{TERM INT}.each do |sig|
require File.dirname(__FILE__) + '/../config/environment'
require 'open3'
-LOG_BUFFER_SIZE = 4096
-
class Dispatcher
include ApplicationHelper
end
def refresh_todo
- @todo = Job.queue.select do |j| j.repository end
- @todo_pipelines = PipelineInstance.queue
+ @todo = []
+ if $options[:jobs]
+ @todo = Job.queue.select(&:repository)
+ end
+ @todo_pipelines = []
+ if $options[:pipelines]
+ @todo_pipelines = PipelineInstance.queue
+ end
end
- def sinfo
+ def each_slurm_line(cmd, outfmt, max_fields=nil)
+ max_fields ||= outfmt.split(":").size
+ max_fields += 1 # To accommodate the node field we add
@@slurm_version ||= Gem::Version.new(`sinfo --version`.match(/\b[\d\.]+\b/)[0])
if Gem::Version.new('2.3') <= @@slurm_version
- `sinfo --noheader -o '%n:%t'`.strip
+ `#{cmd} --noheader -o '%n:#{outfmt}'`.each_line do |line|
+ yield line.chomp.split(":", max_fields)
+ end
else
# Expand rows with hostname ranges (like "foo[1-3,5,9-12]:idle")
# into multiple rows with one hostname each.
- `sinfo --noheader -o '%N:%t'`.split("\n").collect do |line|
- tokens = line.split ":"
+ `#{cmd} --noheader -o '%N:#{outfmt}'`.each_line do |line|
+ tokens = line.chomp.split(":", max_fields)
if (re = tokens[0].match /^(.*?)\[([-,\d]+)\]$/)
- re[2].split(",").collect do |range|
+ tokens.shift
+ re[2].split(",").each do |range|
range = range.split("-").collect(&:to_i)
- (range[0]..range[-1]).collect do |n|
- [re[1] + n.to_s, tokens[1..-1]].join ":"
+ (range[0]..range[-1]).each do |n|
+ yield [re[1] + n.to_s] + tokens
end
end
else
- tokens.join ":"
+ yield tokens
end
- end.flatten.join "\n"
+ end
end
end
+ def slurm_status
+ slurm_nodes = {}
+ each_slurm_line("sinfo", "%t") do |hostname, state|
+ state.sub!(/\W+$/, "")
+ state = "down" unless %w(idle alloc down).include?(state)
+ slurm_nodes[hostname] = {state: state, job: nil}
+ end
+ each_slurm_line("squeue", "%j") do |hostname, job_uuid|
+ slurm_nodes[hostname][:job] = job_uuid if slurm_nodes[hostname]
+ end
+ slurm_nodes
+ end
+
def update_node_status
- if Server::Application.config.crunch_job_wrapper.to_s.match /^slurm/
- @nodes_in_state = {idle: 0, alloc: 0, down: 0}
- @node_state ||= {}
- node_seen = {}
+ return unless Server::Application.config.crunch_job_wrapper.to_s.match /^slurm/
+ @node_state ||= {}
+ slurm_status.each_pair do |hostname, slurmdata|
+ next if @node_state[hostname] == slurmdata
begin
- sinfo.split("\n").
- each do |line|
- re = line.match /(\S+?):+(idle|alloc|down)/
- next if !re
-
- # sinfo tells us about a node N times if it is shared by N partitions
- next if node_seen[re[1]]
- node_seen[re[1]] = true
-
- # count nodes in each state
- @nodes_in_state[re[2].to_sym] += 1
-
- # update our database (and cache) when a node's state changes
- if @node_state[re[1]] != re[2]
- @node_state[re[1]] = re[2]
- node = Node.where('hostname=?', re[1]).first
- if node
- $stderr.puts "dispatch: update #{re[1]} state to #{re[2]}"
- node.info[:slurm_state] = re[2]
- node.save
- elsif re[2] != 'down'
- $stderr.puts "dispatch: sinfo reports '#{re[1]}' is not down, but no node has that name"
- end
+ node = Node.where('hostname=?', hostname).order(:last_ping_at).last
+ if node
+ $stderr.puts "dispatch: update #{hostname} state to #{slurmdata}"
+ node.info["slurm_state"] = slurmdata[:state]
+ node.job_uuid = slurmdata[:job]
+ if node.save
+ @node_state[hostname] = slurmdata
+ else
+ $stderr.puts "dispatch: failed to update #{node.uuid}: #{node.errors.messages}"
end
+ elsif slurmdata[:state] != 'down'
+ $stderr.puts "dispatch: SLURM reports '#{hostname}' is not down, but no node has that name"
end
- rescue
+ rescue => error
+ $stderr.puts "dispatch: error updating #{hostname} node status: #{error}"
end
end
end
- def start_jobs
- @todo.each do |job|
+ def positive_int(raw_value, default=nil)
+ value = begin raw_value.to_i rescue 0 end
+ if value > 0
+ value
+ else
+ default
+ end
+ end
- min_nodes = 1
- begin
- if job.runtime_constraints['min_nodes']
- min_nodes = begin job.runtime_constraints['min_nodes'].to_i rescue 1 end
+ NODE_CONSTRAINT_MAP = {
+ # Map Job runtime_constraints keys to the corresponding Node info key.
+ 'min_ram_mb_per_node' => 'total_ram_mb',
+ 'min_scratch_mb_per_node' => 'total_scratch_mb',
+ 'min_cores_per_node' => 'total_cpu_cores',
+ }
+
+ def nodes_available_for_job_now(job)
+ # Find Nodes that satisfy a Job's runtime constraints (by building
+ # a list of Procs and using them to test each Node). If there
+ # enough to run the Job, return an array of their names.
+ # Otherwise, return nil.
+ need_procs = NODE_CONSTRAINT_MAP.each_pair.map do |job_key, node_key|
+ Proc.new do |node|
+ positive_int(node.info[node_key], 0) >=
+ positive_int(job.runtime_constraints[job_key], 0)
+ end
+ end
+ min_node_count = positive_int(job.runtime_constraints['min_nodes'], 1)
+ usable_nodes = []
+ Node.find_each do |node|
+ good_node = (node.info['slurm_state'] == 'idle')
+ need_procs.each { |node_test| good_node &&= node_test.call(node) }
+ if good_node
+ usable_nodes << node
+ if usable_nodes.count >= min_node_count
+ return usable_nodes.map { |node| node.hostname }
end
end
+ end
+ nil
+ end
- begin
- next if @nodes_in_state[:idle] < min_nodes
- rescue
- end
+ def nodes_available_for_job(job)
+ # Check if there are enough idle nodes with the Job's minimum
+ # hardware requirements to run it. If so, return an array of
+ # their names. If not, up to once per hour, signal start_jobs to
+ # hold off launching Jobs. This delay is meant to give the Node
+ # Manager an opportunity to make new resources available for new
+ # Jobs.
+ #
+ # The exact timing parameters here might need to be adjusted for
+ # the best balance between helping the longest-waiting Jobs run,
+ # and making efficient use of immediately available resources.
+ # These are all just first efforts until we have more data to work
+ # with.
+ nodelist = nodes_available_for_job_now(job)
+ if nodelist.nil? and not did_recently(:wait_for_available_nodes, 3600)
+ $stderr.puts "dispatch: waiting for nodes for #{job.uuid}"
+ @node_wait_deadline = Time.now + 5.minutes
+ end
+ nodelist
+ end
+ def start_jobs
+ @todo.each do |job|
next if @running[job.uuid]
- next if !take(job)
cmd_args = nil
case Server::Application.config.crunch_job_wrapper
when :none
+ if @running.size > 0
+ # Don't run more than one at a time.
+ return
+ end
cmd_args = []
when :slurm_immediate
+ nodelist = nodes_available_for_job(job)
+ if nodelist.nil?
+ if Time.now < @node_wait_deadline
+ break
+ else
+ next
+ end
+ end
cmd_args = ["salloc",
"--chdir=/",
"--immediate",
"--exclusive",
"--no-kill",
"--job-name=#{job.uuid}",
- "--nodes=#{min_nodes}"]
+ "--nodelist=#{nodelist.join(',')}"]
else
raise "Unknown crunch_job_wrapper: #{Server::Application.config.crunch_job_wrapper}"
end
if Server::Application.config.crunch_job_user
cmd_args.unshift("sudo", "-E", "-u",
Server::Application.config.crunch_job_user,
- "PERLLIB=#{ENV['PERLLIB']}")
+ "PATH=#{ENV['PATH']}",
+ "PERLLIB=#{ENV['PERLLIB']}",
+ "PYTHONPATH=#{ENV['PYTHONPATH']}",
+ "RUBYLIB=#{ENV['RUBYLIB']}",
+ "GEM_PATH=#{ENV['GEM_PATH']}")
end
job_auth = ApiClientAuthorization.
new(user: User.where('uuid=?', job.modified_by_user_uuid).first,
api_client_id: 0)
- job_auth.save
+ if not job_auth.save
+ $stderr.puts "dispatch: job_auth.save failed"
+ next
+ end
crunch_job_bin = (ENV['CRUNCH_JOB_BIN'] || `which arv-crunch-job`.strip)
if crunch_job_bin == ''
$stderr.puts `mkdir -p #{arvados_internal.shellescape} && cd #{arvados_internal.shellescape} && git init --bare`
end
- src_repo = File.join(Rails.configuration.git_repositories_dir, job.repository + '.git')
- src_repo = File.join(Rails.configuration.git_repositories_dir, job.repository, '.git') unless File.exists? src_repo
+ repo_root = Rails.configuration.git_repositories_dir
+ src_repo = File.join(repo_root, job.repository + '.git')
+ if not File.exists? src_repo
+ src_repo = File.join(repo_root, job.repository, '.git')
+ if not File.exists? src_repo
+ $stderr.puts "dispatch: No #{job.repository}.git or #{job.repository}/.git at #{repo_root}"
+ sleep 1
+ next
+ end
+ end
- unless src_repo
- $stderr.puts "dispatch: #{File.join Rails.configuration.git_repositories_dir, job.repository} doesn't exist"
- sleep 1
- untake(job)
- next
+ git = "git --git-dir=#{arvados_internal.shellescape}"
+
+ # check if the commit needs to be fetched or not
+ commit_rev = `#{git} rev-list -n1 #{job.script_version.shellescape} 2>/dev/null`.chomp
+ unless $? == 0 and commit_rev == job.script_version
+ # commit does not exist in internal repository, so import the source repository using git fetch-pack
+ cmd = "#{git} fetch-pack --no-progress --all #{src_repo.shellescape}"
+ $stderr.puts cmd
+ $stderr.puts `#{cmd}`
+ unless $? == 0
+ $stderr.puts "dispatch: git fetch-pack failed"
+ sleep 1
+ next
+ end
end
- $stderr.puts `cd #{arvados_internal.shellescape} && git fetch --no-tags #{src_repo.shellescape} && git tag #{job.uuid.shellescape} #{job.script_version.shellescape}`
+ # check if the commit needs to be tagged with this job uuid
+ tag_rev = `#{git} rev-list -n1 #{job.uuid.shellescape} 2>/dev/null`.chomp
+ if $? != 0
+ # no job tag found, so create one
+ cmd = "#{git} tag #{job.uuid.shellescape} #{job.script_version.shellescape}"
+ $stderr.puts cmd
+ $stderr.puts `#{cmd}`
+ unless $? == 0
+ $stderr.puts "dispatch: git tag failed"
+ sleep 1
+ next
+ end
+ else
+ # job tag found, check that it has the expected revision
+ unless tag_rev == job.script_version
+ # Uh oh, the tag doesn't point to the revision we were expecting.
+ # Someone has been monkeying with the job record and/or git.
+ $stderr.puts "dispatch: Already a tag #{job.script_version} pointing to commit #{tag_rev} but expected commit #{job.script_version}"
+ job.state = "Failed"
+ if not job.save
+ $stderr.puts "dispatch: job.save failed"
+ next
+ end
+ next
+ end
+ end
cmd_args << crunch_job_bin
cmd_args << '--job-api-token'
rescue
$stderr.puts "dispatch: popen3: #{$!}"
sleep 1
- untake(job)
next
end
stderr: e,
wait_thr: t,
job: job,
- stderr_buf: '',
+ buf: {stderr: '', stdout: ''},
started: false,
sent_int: 0,
job_auth: job_auth,
- stderr_flushed_at: 0
+ stderr_buf_to_flush: '',
+ stderr_flushed_at: Time.new(0),
+ bytes_logged: 0,
+ events_logged: 0,
+ log_throttle_is_open: true,
+ log_throttle_reset_time: Time.now + Rails.configuration.crunch_log_throttle_period,
+ log_throttle_bytes_so_far: 0,
+ log_throttle_lines_so_far: 0,
+ log_throttle_bytes_skipped: 0,
}
i.close
+ update_node_status
end
end
- def take(job)
- # no-op -- let crunch-job take care of locking.
- true
- end
+ # Test for hard cap on total output and for log throttling. Returns whether
+ # the log line should go to output or not. Modifies "line" in place to
+ # replace it with an error if a logging limit is tripped.
+ def rate_limit running_job, line
+ message = false
+ linesize = line.size
+ if running_job[:log_throttle_is_open]
+ running_job[:log_throttle_lines_so_far] += 1
+ running_job[:log_throttle_bytes_so_far] += linesize
+ running_job[:bytes_logged] += linesize
+
+ if (running_job[:bytes_logged] >
+ Rails.configuration.crunch_limit_log_bytes_per_job)
+ message = "Exceeded log limit #{Rails.configuration.crunch_limit_log_bytes_per_job} bytes (crunch_limit_log_bytes_per_job). Log will be truncated."
+ running_job[:log_throttle_reset_time] = Time.now + 100.years
+ running_job[:log_throttle_is_open] = false
+
+ elsif (running_job[:log_throttle_bytes_so_far] >
+ Rails.configuration.crunch_log_throttle_bytes)
+ remaining_time = running_job[:log_throttle_reset_time] - Time.now
+ message = "Exceeded rate #{Rails.configuration.crunch_log_throttle_bytes} bytes per #{Rails.configuration.crunch_log_throttle_period} seconds (crunch_log_throttle_bytes). Logging will be silenced for the next #{remaining_time.round} seconds.\n"
+ running_job[:log_throttle_is_open] = false
+
+ elsif (running_job[:log_throttle_lines_so_far] >
+ Rails.configuration.crunch_log_throttle_lines)
+ remaining_time = running_job[:log_throttle_reset_time] - Time.now
+ message = "Exceeded rate #{Rails.configuration.crunch_log_throttle_lines} lines per #{Rails.configuration.crunch_log_throttle_period} seconds (crunch_log_throttle_lines), logging will be silenced for the next #{remaining_time.round} seconds.\n"
+ running_job[:log_throttle_is_open] = false
+ end
+ end
- def untake(job)
- # no-op -- let crunch-job take care of locking.
- true
+ if not running_job[:log_throttle_is_open]
+ # Don't log anything if any limit has been exceeded. Just count lossage.
+ running_job[:log_throttle_bytes_skipped] += linesize
+ end
+
+ if message
+ # Yes, write to logs, but use our "rate exceeded" message
+ # instead of the log message that exceeded the limit.
+ line.replace message
+ true
+ else
+ running_job[:log_throttle_is_open]
+ end
end
def read_pipes
@running.each do |job_uuid, j|
job = j[:job]
- # Throw away child stdout
- begin
- j[:stdout].read_nonblock(2**20)
- rescue Errno::EAGAIN, EOFError
- end
+ now = Time.now
+ if now > j[:log_throttle_reset_time]
+ # It has been more than throttle_period seconds since the last
+ # checkpoint so reset the throttle
+ if j[:log_throttle_bytes_skipped] > 0
+ message = "#{job_uuid} ! Skipped #{j[:log_throttle_bytes_skipped]} bytes of log"
+ $stderr.puts message
+ j[:stderr_buf_to_flush] << "#{Time.now.ctime.to_s} #{message}\n"
+ end
- # Read whatever is available from child stderr
- stderr_buf = false
- begin
- stderr_buf = j[:stderr].read_nonblock(2**20)
- rescue Errno::EAGAIN, EOFError
+ j[:log_throttle_reset_time] = now + Rails.configuration.crunch_log_throttle_period
+ j[:log_throttle_bytes_so_far] = 0
+ j[:log_throttle_lines_so_far] = 0
+ j[:log_throttle_bytes_skipped] = 0
+ j[:log_throttle_is_open] = true
end
- if stderr_buf
- if stderr_buf.index "\n"
- lines = stderr_buf.lines("\n").to_a
- lines.each do |line|
+ j[:buf].each do |stream, streambuf|
+ # Read some data from the child stream
+ buf = ''
+ begin
+ # It's important to use a big enough buffer here. When we're
+ # being flooded with logs, we must read and discard many
+ # bytes at once. Otherwise, we can easily peg a CPU with
+ # time-checking and other loop overhead. (Quick tests show a
+ # 1MiB buffer working 2.5x as fast as a 64 KiB buffer.)
+ #
+ # So don't reduce this buffer size!
+ buf = j[stream].read_nonblock(2**20)
+ rescue Errno::EAGAIN, EOFError
+ end
+
+ # Short circuit the counting code if we're just going to throw
+ # away the data anyway.
+ if not j[:log_throttle_is_open]
+ j[:log_throttle_bytes_skipped] += streambuf.size + buf.size
+ streambuf.replace ''
+ next
+ elsif buf == ''
+ next
+ end
+
+ # Append to incomplete line from previous read, if any
+ streambuf << buf
+
+ bufend = ''
+ streambuf.each_line do |line|
+ if not line.end_with? $/
+ if line.size > Rails.configuration.crunch_log_throttle_bytes
+ # Without a limit here, we'll use 2x an arbitrary amount
+ # of memory, and waste a lot of time copying strings
+ # around, all without providing any feedback to anyone
+ # about what's going on _or_ hitting any of our throttle
+ # limits.
+ #
+ # Here we leave "line" alone, knowing it will never be
+ # sent anywhere: rate_limit() will reach
+ # crunch_log_throttle_bytes immediately. However, we'll
+ # leave [...] in bufend: if the trailing end of the long
+ # line does end up getting sent anywhere, it will have
+ # some indication that it is incomplete.
+ bufend = "[...]"
+ else
+ # If line length is sane, we'll wait for the rest of the
+ # line to appear in the next read_pipes() call.
+ bufend = line
+ break
+ end
+ end
+ # rate_limit returns true or false as to whether to actually log
+ # the line or not. It also modifies "line" in place to replace
+ # it with an error if a logging limit is tripped.
+ if rate_limit j, line
$stderr.print "#{job_uuid} ! " unless line.index(job_uuid)
$stderr.puts line
- log_msg = "#{Time.now.ctime.to_s} #{line.strip}"
- j[:stderr_buf] << (log_msg + " \n")
- end
-
- if (LOG_BUFFER_SIZE < j[:stderr_buf].size) || ((j[:stderr_flushed_at]+1) < Time.now.to_i)
- write_log j
- j[:stderr_flushed_at] = Time.now.to_i
+ pub_msg = "#{Time.now.ctime.to_s} #{line.strip}\n"
+ j[:stderr_buf_to_flush] << pub_msg
end
end
+
+ # Leave the trailing incomplete line (if any) in streambuf for
+ # next time.
+ streambuf.replace bufend
end
+ # Flush buffered logs to the logs table, if appropriate. We have
+ # to do this even if we didn't collect any new logs this time:
+ # otherwise, buffered data older than seconds_between_events
+ # won't get flushed until new data arrives.
+ write_log j
end
end
$stderr.puts "dispatch: child #{pid_done} exit"
$stderr.puts "dispatch: job #{job_done.uuid} end"
- # Ensure every last drop of stdout and stderr is consumed
+ # 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"
+ # Reset flush timestamp to make sure log gets written.
+ j_done[:stderr_flushed_at] = Time.new(0)
+ # Write any remaining logs.
+ write_log j_done
+
+ j_done[:buf].each do |stream, streambuf|
+ if streambuf != ''
+ $stderr.puts streambuf + "\n"
+ end
end
- # Wait the thread
- j_done[:wait_thr].value
+ # Wait the thread (returns a Process::Status)
+ exit_status = j_done[:wait_thr].value.exitstatus
jobrecord = Job.find_by_uuid(job_done.uuid)
- if jobrecord.started_at
- # Clean up state fields in case crunch-job exited without
- # putting the job in a suitable "finished" state.
- jobrecord.running = false
- jobrecord.finished_at ||= Time.now
- if jobrecord.success.nil?
- jobrecord.success = false
+ if exit_status != 75 and jobrecord.state == "Running"
+ # crunch-job did not return exit code 75 (see below) and left the job in
+ # the "Running" state, which means there was an unhandled error. Fail
+ # the job.
+ jobrecord.state = "Failed"
+ if not jobrecord.save
+ $stderr.puts "dispatch: jobrecord.save failed"
end
- jobrecord.save!
else
# Don't fail the job if crunch-job didn't even get as far as
# starting it. If the job failed to run due to an infrastructure
# issue with crunch-job or slurm, we want the job to stay in the
- # queue.
+ # queue. If crunch-job exited after losing a race to another
+ # crunch-job process, it exits 75 and we should leave the job
+ # record alone so the winner of the race do its thing.
+ #
+ # There is still an unhandled race condition: If our crunch-job
+ # process is about to lose a race with another crunch-job
+ # process, but crashes before getting to its "exit 75" (for
+ # example, "cannot fork" or "cannot reach API server") then we
+ # will assume incorrectly that it's our process's fault
+ # jobrecord.started_at is non-nil, and mark the job as failed
+ # even though the winner of the race is probably still doing
+ # fine.
end
# Invalidate the per-job auth token
pipe_auth = (@pipe_auth_tokens[p.uuid] ||= ApiClientAuthorization.
create(user: User.where('uuid=?', p.modified_by_user_uuid).first,
api_client_id: 0))
- puts `export ARVADOS_API_TOKEN=#{pipe_auth.api_token} && arv-run-pipeline-instance --run-here --no-wait --instance #{p.uuid}`
+ puts `export ARVADOS_API_TOKEN=#{pipe_auth.api_token} && arv-run-pipeline-instance --run-pipeline-here --no-wait --instance #{p.uuid}`
expire_tokens.delete p.uuid
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] != '')
+ return if running_job[:stderr_buf_to_flush] == ''
+
+ # Send out to log event if buffer size exceeds the bytes per event or if
+ # it has been at least crunch_log_seconds_between_events seconds since
+ # the last flush.
+ if running_job[:stderr_buf_to_flush].size > Rails.configuration.crunch_log_bytes_per_event or
+ (Time.now - running_job[:stderr_flushed_at]) >= Rails.configuration.crunch_log_seconds_between_events
+ begin
log = Log.new(object_uuid: running_job[:job].uuid,
event_type: 'stderr',
- properties: {"text" => running_job[:stderr_buf]})
+ owner_uuid: running_job[:job].owner_uuid,
+ properties: {"text" => running_job[:stderr_buf_to_flush]})
log.save!
- running_job[:stderr_buf] = ''
- running_job[:stderr_flushed_at] = Time.now.to_i
+ running_job[:events_logged] += 1
+ rescue => exception
+ $stderr.puts "Failed to write logs"
+ $stderr.puts exception.backtrace
end
- rescue
- running_job[:stderr_buf] = "Failed to write logs \n"
- running_job[:stderr_flushed_at] = Time.now.to_i
+ running_job[:stderr_buf_to_flush] = ''
+ running_job[:stderr_flushed_at] = Time.now
end
end
-
end
# This is how crunch-job child procs know where the "refresh" trigger file is
projects / arvados.git / blobdiff