X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/bf7687829f5222375441cf75c7d82d82a4f508c4..96fe9fc822c2270cb6e4212798d61e61481ec2e0:/services/api/script/crunch-dispatch.rb diff --git a/services/api/script/crunch-dispatch.rb b/services/api/script/crunch-dispatch.rb index 87acb651a4..c39c8ea992 100755 --- a/services/api/script/crunch-dispatch.rb +++ b/services/api/script/crunch-dispatch.rb @@ -1,7 +1,5 @@ #!/usr/bin/env ruby -require 'trollop' - include Process $warned = {} @@ -22,18 +20,12 @@ if ENV["CRUNCH_DISPATCH_LOCKFILE"] 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' -LOG_BUFFER_SIZE = 4096 - class Dispatcher include ApplicationHelper @@ -71,7 +63,6 @@ class Dispatcher 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 = {} begin @@ -84,79 +75,132 @@ class Dispatcher 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 + node.info['slurm_state'] = re[2] + if not node.save + $stderr.puts "dispatch: failed to update #{node.uuid}: #{node.errors.messages}" + end elsif re[2] != 'down' $stderr.puts "dispatch: sinfo reports '#{re[1]}' is not down, but no node has that name" end end end - rescue + rescue => error + $stderr.puts "dispatch: error updating 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) - 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 + next if !take(job) - 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']}" + if Server::Application.config.crunch_job_user + cmd_args.unshift("sudo", "-E", "-u", + Server::Application.config.crunch_job_user, + "PATH=#{ENV['PATH']}", + "PERLLIB=#{ENV['PERLLIB']}", + "PYTHONPATH=#{ENV['PYTHONPATH']}", + "RUBYLIB=#{ENV['RUBYLIB']}", + "GEM_PATH=#{ENV['GEM_PATH']}") end job_auth = ApiClientAuthorization. @@ -186,7 +230,7 @@ class Dispatcher next end - $stderr.puts `cd #{arvados_internal.shellescape} && git fetch --no-tags #{src_repo.shellescape} && git tag #{job.uuid.shellescape} #{job.script_version.shellescape}` + $stderr.puts `cd #{arvados_internal.shellescape} && git fetch-pack --all #{src_repo.shellescape} && git tag #{job.uuid.shellescape} #{job.script_version.shellescape}` cmd_args << crunch_job_bin cmd_args << '--job-api-token' @@ -196,10 +240,10 @@ class Dispatcher cmd_args << '--git-dir' cmd_args << arvados_internal - $stderr.puts "dispatch: #{cmd_args}" + $stderr.puts "dispatch: #{cmd_args.join ' '}" begin - i, o, e, t = Open3.popen3({}, *cmd_args, { :unsetenv_others => true}) + i, o, e, t = Open3.popen3(*cmd_args) rescue $stderr.puts "dispatch: popen3: #{$!}" sleep 1 @@ -222,9 +266,13 @@ class Dispatcher sent_int: 0, job_auth: job_auth, stderr_buf_to_flush: '', - stderr_flushed_at: 0 + stderr_flushed_at: 0, + bytes_logged: 0, + events_logged: 0, + log_truncated: false } i.close + update_node_status end end @@ -271,7 +319,8 @@ class Dispatcher 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) + if (Rails.configuration.crunch_log_bytes_per_event < j[:stderr_buf_to_flush].size or + (j[:stderr_flushed_at] + Rails.configuration.crunch_log_seconds_between_events < Time.now.to_i)) write_log j end end @@ -405,6 +454,15 @@ class Dispatcher protected + def too_many_bytes_logged_for_job(j) + return (j[:bytes_logged] + j[:stderr_buf_to_flush].size > + Rails.configuration.crunch_limit_log_event_bytes_per_job) + end + + def too_many_events_logged_for_job(j) + return (j[:events_logged] >= Rails.configuration.crunch_limit_log_events_per_job) + end + def did_recently(thing, min_interval) @did_recently ||= {} if !@did_recently[thing] or @did_recently[thing] < Time.now - min_interval @@ -419,11 +477,26 @@ class Dispatcher def write_log running_job begin if (running_job && running_job[:stderr_buf_to_flush] != '') + # Truncate logs if they exceed crunch_limit_log_event_bytes_per_job + # or crunch_limit_log_events_per_job. + if (too_many_bytes_logged_for_job(running_job)) + return if running_job[:log_truncated] + running_job[:log_truncated] = true + running_job[:stderr_buf_to_flush] = + "Server configured limit reached (crunch_limit_log_event_bytes_per_job: #{Rails.configuration.crunch_limit_log_event_bytes_per_job}). Subsequent logs truncated" + elsif (too_many_events_logged_for_job(running_job)) + return if running_job[:log_truncated] + running_job[:log_truncated] = true + running_job[:stderr_buf_to_flush] = + "Server configured limit reached (crunch_limit_log_events_per_job: #{Rails.configuration.crunch_limit_log_events_per_job}). Subsequent logs truncated" + end 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[:bytes_logged] += running_job[:stderr_buf_to_flush].size + running_job[:events_logged] += 1 running_job[:stderr_buf_to_flush] = '' running_job[:stderr_flushed_at] = Time.now.to_i end