X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/f4cc8a680d7d9d0ea1447a8cbf208daebabfa503..cf7e30873ef4b92cc8ec099b2bb344391a070e93:/services/api/script/crunch-dispatch.rb?ds=sidebyside diff --git a/services/api/script/crunch-dispatch.rb b/services/api/script/crunch-dispatch.rb index 24595ac6af..693dfbdb7f 100755 --- a/services/api/script/crunch-dispatch.rb +++ b/services/api/script/crunch-dispatch.rb @@ -2,6 +2,7 @@ include Process +$warned = {} $signal = {} %w{TERM INT}.each do |sig| signame = sig @@ -11,12 +12,23 @@ $signal = {} end end +if ENV["CRUNCH_DISPATCH_LOCKFILE"] + lockfilename = ENV.delete "CRUNCH_DISPATCH_LOCKFILE" + lockfile = File.open(lockfilename, File::RDWR|File::CREAT, 0644) + unless lockfile.flock File::LOCK_EX|File::LOCK_NB + abort "Lock unavailable on #{lockfilename} - exit" + end +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 + class Dispatcher include ApplicationHelper @@ -26,31 +38,82 @@ class Dispatcher def refresh_todo @todo = Job.queue + @todo_pipelines = PipelineInstance.queue end - def start_jobs + def sinfo + @@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 + 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 ":" + if (re = tokens[0].match /^(.*?)\[([-,\d]+)\]$/) + re[2].split(",").collect do |range| + range = range.split("-").collect(&:to_i) + (range[0]..range[-1]).collect do |n| + [re[1] + n.to_s, tokens[1..-1]].join ":" + end + end + else + tokens.join ":" + end + end.flatten.join "\n" + end + end + + def update_node_status if Server::Application.config.crunch_job_wrapper.to_s.match /^slurm/ - @idle_slurm_nodes = 0 + @nodes_in_state = {idle: 0, alloc: 0, down: 0} + @node_state ||= {} + node_seen = {} begin - `sinfo`. - split("\n"). - collect { |line| line.match /(\d+) +idle/ }. - each do |re| - @idle_slurm_nodes = re[1].to_i if re + 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 + end end rescue end end + end + def start_jobs @todo.each do |job| min_nodes = 1 begin - if job.resource_limits['min_nodes'] - min_nodes = begin job.resource_limits['min_nodes'].to_i rescue 1 end + if job.runtime_constraints['min_nodes'] + min_nodes = begin job.runtime_constraints['min_nodes'].to_i rescue 1 end end end - next if @idle_slurm_nodes and @idle_slurm_nodes < min_nodes + + begin + next if @nodes_in_state[:idle] < min_nodes + rescue + end next if @running[job.uuid] next if !take(job) @@ -61,6 +124,7 @@ class Dispatcher cmd_args = [] when :slurm_immediate cmd_args = ["salloc", + "--chdir=/", "--immediate", "--exclusive", "--no-kill", @@ -77,11 +141,16 @@ class Dispatcher end job_auth = ApiClientAuthorization. - new(user: User.where('uuid=?', job.modified_by_user).first, + new(user: User.where('uuid=?', job.modified_by_user_uuid).first, api_client_id: 0) job_auth.save - cmd_args << ENV['CRUNCH_JOB_BIN'] || 'crunch-job' + crunch_job_bin = (ENV['CRUNCH_JOB_BIN'] || `which arv-crunch-job`.strip) + if crunch_job_bin == '' + raise "No CRUNCH_JOB_BIN env var, and crunch-job not in path." + end + + cmd_args << crunch_job_bin cmd_args << '--job-api-token' cmd_args << job_auth.api_token cmd_args << '--job' @@ -90,10 +159,17 @@ class Dispatcher commit = Commit.where(sha1: job.script_version).first if commit cmd_args << '--git-dir' - cmd_args << File. - join(Rails.configuration.git_repositories_dir, - commit.repository_name, - '.git') + if File.exists?(File. + join(Rails.configuration.git_repositories_dir, + commit.repository_name + '.git')) + cmd_args << File. + join(Rails.configuration.git_repositories_dir, + commit.repository_name + '.git') + else + cmd_args << File. + join(Rails.configuration.git_repositories_dir, + commit.repository_name, '.git') + end end $stderr.puts "dispatch: #{cmd_args.join ' '}" @@ -106,8 +182,14 @@ class Dispatcher untake(job) next end - $stderr.puts "dispatch: job #{job.uuid} start" - $stderr.puts "dispatch: child #{t.pid} start" + + $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, stdout: o, @@ -162,6 +244,16 @@ class Dispatcher 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 end end end @@ -206,6 +298,7 @@ class Dispatcher job_done = j_done[:job] $stderr.puts "dispatch: child #{pid_done} exit" $stderr.puts "dispatch: job #{job_done.uuid} end" + $redis.publish job_done.uuid, "end" # Ensure every last drop of stdout and stderr is consumed read_pipes @@ -222,6 +315,17 @@ class Dispatcher @running.delete job_done.uuid end + def update_pipelines + @todo_pipelines.each do |p| + pipe_auth = ApiClientAuthorization. + new(user: User.where('uuid=?', p.modified_by_user_uuid).first, + api_client_id: 0) + pipe_auth.save + + puts `export ARVADOS_API_TOKEN=#{pipe_auth.api_token} && arv-run-pipeline-instance --run-here --no-wait --instance #{p.uuid}` + end + end + def run act_as_system_user @running ||= {} @@ -242,7 +346,13 @@ class Dispatcher end else refresh_todo unless did_recently(:refresh_todo, 1.0) - start_jobs unless @todo.empty? or did_recently(:start_jobs, 1.0) + update_node_status + unless @todo.empty? or did_recently(:start_jobs, 1.0) or $signal[:term] + start_jobs + end + unless @todo_pipelines.empty? or did_recently(:update_pipelines, 5.0) + update_pipelines + end end reap_children select(@running.values.collect { |j| [j[:stdout], j[:stderr]] }.flatten, @@ -250,6 +360,8 @@ class Dispatcher end end + + protected def did_recently(thing, min_interval) @@ -263,4 +375,7 @@ class Dispatcher end end +# This is how crunch-job child procs know where the "refresh" trigger file is +ENV["CRUNCH_REFRESH_TRIGGER"] = Rails.configuration.crunch_refresh_trigger + Dispatcher.new.run