Merge branch 'master' into 1971-show-image-thumbnails

[arvados.git] / services / api / script / crunch-dispatch.rb

diff --git a/services/api/script/crunch-dispatch.rb b/services/api/script/crunch-dispatch.rb
index 59ea162c0e44a8fda87f70d9c2257ec6a3eac7c3..d9db69f6502a87ca22c5565c149806b6e5da6198 100755 (executable)
--- a/services/api/script/crunch-dispatch.rb
+++ b/services/api/script/crunch-dispatch.rb
@@ -38,6 +38,7 @@ class Dispatcher
 
   def refresh_todo
     @todo = Job.queue
+    @todo_pipelines = PipelineInstance.queue
   end
 
   def sinfo
@@ -144,32 +145,38 @@ class Dispatcher
             api_client_id: 0)
       job_auth.save
 
-      cmd_args << (ENV['CRUNCH_JOB_BIN'] || `which crunch-job`.strip)
-      cmd_args << '--job-api-token'
-      cmd_args << job_auth.api_token
-      cmd_args << '--job'
-      cmd_args << job.uuid
-
-      if cmd_args[0] == ''
+      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
 
-      commit = Commit.where(sha1: job.script_version).first
-      if commit
-        cmd_args << '--git-dir'
-        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
+      require 'shellwords'
+
+      arvados_internal = Rails.configuration.git_internal_dir
+      if not File.exists? arvados_internal
+        $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
+
+      unless src_repo
+        $stderr.puts "dispatch: #{File.join Rails.configuration.git_repositories_dir, job.repository} doesn't exist"
+        sleep 1
+        untake(job)
+        next
+      end
+
+      $stderr.puts `cd #{arvados_internal.shellescape} && git fetch --no-tags #{src_repo.shellescape} && git tag #{job.uuid.shellescape} #{job.script_version.shellescape}`
+
+      cmd_args << crunch_job_bin
+      cmd_args << '--job-api-token'
+      cmd_args << job_auth.api_token
+      cmd_args << '--job'
+      cmd_args << job.uuid
+      cmd_args << '--git-dir'
+      cmd_args << arvados_internal
+
       $stderr.puts "dispatch: #{cmd_args.join ' '}"
 
       begin
@@ -296,7 +303,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
@@ -307,15 +313,41 @@ 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
+    expire_tokens = @pipe_auth_tokens.dup
+    @todo_pipelines.each do |p|
+      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}`
+      expire_tokens.delete p.uuid
+    end
+
+    expire_tokens.each do |k, v|
+      v.update_attributes expires_at: Time.now
+      @pipe_auth_tokens.delete k
+    end
+  end
+
   def run
     act_as_system_user
     @running ||= {}
+    @pipe_auth_tokens ||= { }
     $stderr.puts "dispatch: ready"
     while !$signal[:term] or @running.size > 0
       read_pipes
@@ -337,6 +369,9 @@ class Dispatcher
         unless @todo.empty? or did_recently(:start_jobs, 1.0) or $signal[:term]
           start_jobs
         end
+        unless (@todo_pipelines.empty? and @pipe_auth_tokens.empty?) or did_recently(:update_pipelines, 5.0)
+          update_pipelines
+        end
       end
       reap_children
       select(@running.values.collect { |j| [j[:stdout], j[:stderr]] }.flatten,