X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e346e5d3bd09f4b9fe66538a825cf866faeb308b..64e7ea7910c797f0339ddd1d5df676b3e53fa9de:/services/api/script/crunch-dispatch.rb

diff --git a/services/api/script/crunch-dispatch.rb b/services/api/script/crunch-dispatch.rb
index 0bcb5d56c0..a23b02b15c 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,6 +12,14 @@ $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'
@@ -18,6 +27,7 @@ require File.dirname(__FILE__) + '/../config/environment'
 require 'open3'
 
 $redis ||= Redis.new
+LOG_BUFFER_SIZE = 2**20
 
 class Dispatcher
   include ApplicationHelper
@@ -30,29 +40,57 @@ class Dispatcher
     @todo = Job.queue
   end
 
-  def start_jobs
+  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`.
+        `sinfo --noheader -o '%n:%t'`.
           split("\n").
-          collect { |line| line.match /(\d+) +idle/ }.
-          each do |re|
-          @idle_slurm_nodes = re[1].to_i if re
+          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)
@@ -90,6 +128,10 @@ class Dispatcher
       cmd_args << '--job'
       cmd_args << job.uuid
 
+      if cmd_args[0] == ''
+        raise "No CRUNCH_JOB_BIN env var, and crunch-job not in path."
+      end
+
       commit = Commit.where(sha1: job.script_version).first
       if commit
         cmd_args << '--git-dir'
@@ -116,8 +158,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,
@@ -172,7 +220,16 @@ class Dispatcher
           lines.each do |line|
             $stderr.print "#{job_uuid} ! " unless line.index(job_uuid)
             $stderr.puts line
-            $redis.publish job_uuid, "#{Time.now.ctime.to_s} #{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
@@ -254,7 +311,10 @@ 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
       end
       reap_children
       select(@running.values.collect { |j| [j[:stdout], j[:stderr]] }.flatten,
@@ -275,4 +335,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