X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/2540a26167950a6d80f5eaf183500f22739d17de..eb0fdee4d4fdb516dd57998ae6a34840354e6e77:/services/api/script/crunch-dispatch.rb

diff --git a/services/api/script/crunch-dispatch.rb b/services/api/script/crunch-dispatch.rb
index a9ca4bebc4..9eb1c28cb2 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
@@ -26,6 +27,7 @@ require File.dirname(__FILE__) + '/../config/environment'
 require 'open3'
 
 $redis ||= Redis.new
+LOG_BUFFER_SIZE = 2**20
 
 class Dispatcher
   include ApplicationHelper
@@ -36,6 +38,30 @@ class Dispatcher
 
   def refresh_todo
     @todo = Job.queue
+    @todo_pipelines = PipelineInstance.queue
+  end
+
+  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
@@ -44,8 +70,7 @@ class Dispatcher
       @node_state ||= {}
       node_seen = {}
       begin
-        `sinfo --noheader -o '%n:%t'`.
-          split("\n").
+        sinfo.split("\n").
           each do |line|
           re = line.match /(\S+?):+(idle|alloc|down)/
           next if !re
@@ -120,7 +145,12 @@ class Dispatcher
             api_client_id: 0)
       job_auth.save
 
-      cmd_args << (ENV['CRUNCH_JOB_BIN'] || `which crunch-job`.strip)
+      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'
@@ -152,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,
@@ -208,7 +244,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.strip}"
+            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
@@ -253,7 +298,6 @@ 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
@@ -264,12 +308,32 @@ class Dispatcher
     # Wait the thread
     j_done[:wait_thr].value
 
+    jobrecord = Job.find_by_uuid(job_done.uuid)
+    jobrecord.running = false
+    jobrecord.finished_at ||= Time.now,
+    # Don't set 'jobrecord.success = false' because if the job failed to run due to an
+    # issue with crunch-job or slurm, we want the job to stay in the queue.
+    jobrecord.save!
+
     # 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
 
+  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 ||= {}
@@ -291,7 +355,12 @@ class Dispatcher
       else
         refresh_todo unless did_recently(:refresh_todo, 1.0)
         update_node_status
-        start_jobs unless @todo.empty? or did_recently(:start_jobs, 1.0)
+        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,
@@ -312,4 +381,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