X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/023517001bf8efd7d44ab33d2a79d92d2ff30a30..c6fe632f972e4610ff7f35f83a5d7dcd2d6e7ecb:/services/api/script/crunch-dispatch.rb

diff --git a/services/api/script/crunch-dispatch.rb b/services/api/script/crunch-dispatch.rb
index 81e1ef96e3..1002f91347 100755
--- a/services/api/script/crunch-dispatch.rb
+++ b/services/api/script/crunch-dispatch.rb
@@ -44,19 +44,50 @@ require File.dirname(__FILE__) + '/../config/boot'
 require File.dirname(__FILE__) + '/../config/environment'
 require 'open3'
 
+class LogTime < Time
+  def to_s
+    self.utc.strftime "%Y-%m-%d_%H:%M:%S"
+  end
+end
+
 class Dispatcher
   include ApplicationHelper
 
+  EXIT_TEMPFAIL = 75
+
+  def initialize
+    @crunch_job_bin = (ENV['CRUNCH_JOB_BIN'] || `which arv-crunch-job`.strip)
+    if @crunch_job_bin.empty?
+      raise "No CRUNCH_JOB_BIN env var, and crunch-job not in path."
+    end
+
+    @arvados_internal = Rails.configuration.git_internal_dir
+    if not File.exists? @arvados_internal
+      $stderr.puts `mkdir -p #{@arvados_internal.shellescape} && git init --bare #{@arvados_internal.shellescape}`
+      raise "No internal git repository available" unless ($? == 0)
+    end
+
+    @repo_root = Rails.configuration.git_repositories_dir
+    @arvados_repo_path = Repository.where(name: "arvados").first.server_path
+    @authorizations = {}
+    @did_recently = {}
+    @fetched_commits = {}
+    @git_tags = {}
+    @node_state = {}
+    @pipe_auth_tokens = {}
+    @running = {}
+    @todo = []
+    @todo_pipelines = []
+  end
+
   def sysuser
     return act_as_system_user
   end
 
   def refresh_todo
-    @todo = []
     if $options[:jobs]
       @todo = Job.queue.select(&:repository)
     end
-    @todo_pipelines = []
     if $options[:pipelines]
       @todo_pipelines = PipelineInstance.queue
     end
@@ -108,7 +139,6 @@ class Dispatcher
 
   def update_node_status
     return unless Server::Application.config.crunch_job_wrapper.to_s.match /^slurm/
-    @node_state ||= {}
     slurm_status.each_pair do |hostname, slurmdata|
       next if @node_state[hostname] == slurmdata
       begin
@@ -205,10 +235,102 @@ class Dispatcher
     rescue
       $stderr.puts "dispatch: log.create failed"
     end
-    job.state = "Failed"
-    if not job.save
-      $stderr.puts "dispatch: job.save failed"
+
+    begin
+      job.lock @authorizations[job.uuid].user.uuid
+      job.state = "Failed"
+      if not job.save
+        $stderr.puts "dispatch: save failed setting job #{job.uuid} to failed"
+      end
+    rescue ArvadosModel::AlreadyLockedError
+      $stderr.puts "dispatch: tried to mark job #{job.uuid} as failed but it was already locked by someone else"
+    end
+  end
+
+  def stdout_s(cmd_a, opts={})
+    IO.popen(cmd_a, "r", opts) do |pipe|
+      return pipe.read.chomp
+    end
+  end
+
+  def git_cmd(*cmd_a)
+    ["git", "--git-dir=#{@arvados_internal}"] + cmd_a
+  end
+
+  def get_authorization(job)
+    if @authorizations[job.uuid] and
+        @authorizations[job.uuid].user.uuid != job.modified_by_user_uuid
+      # We already made a token for this job, but we need a new one
+      # because modified_by_user_uuid has changed (the job will run
+      # as a different user).
+      @authorizations[job.uuid].update_attributes expires_at: Time.now
+      @authorizations[job.uuid] = nil
+    end
+    if not @authorizations[job.uuid]
+      auth = ApiClientAuthorization.
+        new(user: User.where('uuid=?', job.modified_by_user_uuid).first,
+            api_client_id: 0)
+      if not auth.save
+        $stderr.puts "dispatch: auth.save failed for #{job.uuid}"
+      else
+        @authorizations[job.uuid] = auth
+      end
+    end
+    @authorizations[job.uuid]
+  end
+
+  def internal_repo_has_commit? sha1
+    if (not @fetched_commits[sha1] and
+        sha1 == stdout_s(git_cmd("rev-list", "-n1", sha1), err: "/dev/null") and
+        $? == 0)
+      @fetched_commits[sha1] = true
+    end
+    return @fetched_commits[sha1]
+  end
+
+  def get_commit src_repo, sha1
+    return true if internal_repo_has_commit? sha1
+
+    # commit does not exist in internal repository, so import the
+    # source repository using git fetch-pack
+    cmd = git_cmd("fetch-pack", "--no-progress", "--all", src_repo)
+    $stderr.puts "dispatch: #{cmd}"
+    $stderr.puts(stdout_s(cmd))
+    @fetched_commits[sha1] = ($? == 0)
+  end
+
+  def tag_commit(commit_hash, tag_name)
+    # @git_tags[T]==V if we know commit V has been tagged T in the
+    # arvados_internal repository.
+    if not @git_tags[tag_name]
+      cmd = git_cmd("tag", tag_name, commit_hash)
+      $stderr.puts "dispatch: #{cmd}"
+      $stderr.puts(stdout_s(cmd, err: "/dev/null"))
+      unless $? == 0
+        # git tag failed.  This may be because the tag already exists, so check for that.
+        tag_rev = stdout_s(git_cmd("rev-list", "-n1", tag_name))
+        if $? == 0
+          # We got a revision back
+          if tag_rev != commit_hash
+            # Uh oh, the tag doesn't point to the revision we were expecting.
+            # Someone has been monkeying with the job record and/or git.
+            fail_job job, "Existing tag #{tag_name} points to commit #{tag_rev} but expected commit #{commit_hash}"
+            return nil
+          end
+          # we're okay (fall through to setting @git_tags below)
+        else
+          # git rev-list failed for some reason.
+          fail_job job, "'git tag' for #{tag_name} failed but did not find any existing tag using 'git rev-list'"
+          return nil
+        end
+      end
+      # 'git tag' was successful, or there is an existing tag that points to the same revision.
+      @git_tags[tag_name] = commit_hash
+    elsif @git_tags[tag_name] != commit_hash
+      fail_job job, "Existing tag #{tag_name} points to commit #{@git_tags[tag_name]} but this job uses commit #{commit_hash}"
+      return nil
     end
+    @git_tags[tag_name]
   end
 
   def start_jobs
@@ -253,108 +375,47 @@ class Dispatcher
                          "GEM_PATH=#{ENV['GEM_PATH']}")
       end
 
-      @authorizations ||= {}
-      if @authorizations[job.uuid] and
-          @authorizations[job.uuid].user.uuid != job.modified_by_user_uuid
-        # We already made a token for this job, but we need a new one
-        # because modified_by_user_uuid has changed (the job will run
-        # as a different user).
-        @authorizations[job.uuid].update_attributes expires_at: Time.now
-        @authorizations[job.uuid] = nil
-      end
-      if not @authorizations[job.uuid]
-        auth = ApiClientAuthorization.
-          new(user: User.where('uuid=?', job.modified_by_user_uuid).first,
-              api_client_id: 0)
-        if not auth.save
-          $stderr.puts "dispatch: auth.save failed"
-          next
-        end
-        @authorizations[job.uuid] = auth
-      end
-
-      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
+      next unless get_authorization job
 
-      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
+      ready = internal_repo_has_commit? job.script_version
 
-      git = "git --git-dir=#{arvados_internal.shellescape}"
-
-      # @fetched_commits[V]==true if we know commit V exists in the
-      # arvados_internal git repository.
-      @fetched_commits ||= {}
-      if !@fetched_commits[job.script_version]
-
-        repo_root = Rails.configuration.git_repositories_dir
-        src_repo = File.join(repo_root, job.repository + '.git')
-        if not File.exists? src_repo
-          src_repo = File.join(repo_root, job.repository, '.git')
-          if not File.exists? src_repo
-            fail_job job, "No #{job.repository}.git or #{job.repository}/.git at #{repo_root}"
-            next
-          end
+      if not ready
+        # Import the commit from the specified repository into the
+        # internal repository. This should have been done already when
+        # the job was created/updated; this code is obsolete except to
+        # avoid deployment races. Failing the job would be a
+        # reasonable thing to do at this point.
+        repo = Repository.where(name: job.repository).first
+        if repo.nil? or repo.server_path.nil?
+          fail_job "Repository #{job.repository} not found under #{@repo_root}"
+          next
         end
+        ready &&= get_commit repo.server_path, job.script_version
+        ready &&= tag_commit job.script_version, job.uuid
+      end
 
-        # check if the commit needs to be fetched or not
-        commit_rev = `#{git} rev-list -n1 #{job.script_version.shellescape} 2>/dev/null`.chomp
-        unless $? == 0 and commit_rev == job.script_version
-          # commit does not exist in internal repository, so import the source repository using git fetch-pack
-          cmd = "#{git} fetch-pack --no-progress --all #{src_repo.shellescape}"
-          $stderr.puts "dispatch: #{cmd}"
-          $stderr.puts `#{cmd}`
-          unless $? == 0
-            fail_job job, "git fetch-pack failed"
-            next
-          end
-        end
-        @fetched_commits[job.script_version] = true
+      # This should be unnecessary, because API server does it during
+      # job create/update, but it's still not a bad idea to verify the
+      # tag is correct before starting the job:
+      ready &&= tag_commit job.script_version, job.uuid
+
+      # The arvados_sdk_version doesn't support use of arbitrary
+      # remote URLs, so the requested version isn't necessarily copied
+      # into the internal repository yet.
+      if job.arvados_sdk_version
+        ready &&= get_commit @arvados_repo_path, job.arvados_sdk_version
+        ready &&= tag_commit job.arvados_sdk_version, "#{job.uuid}-arvados-sdk"
       end
 
-      # @job_tags[J]==V if we know commit V has been tagged J in the
-      # arvados_internal repository. (J is a job UUID, V is a commit
-      # sha1.)
-      @job_tags ||= {}
-      if not @job_tags[job.uuid]
-        cmd = "#{git} tag #{job.uuid.shellescape} #{job.script_version.shellescape} 2>/dev/null"
-        $stderr.puts "dispatch: #{cmd}"
-        $stderr.puts `#{cmd}`
-        unless $? == 0
-          # git tag failed.  This may be because the tag already exists, so check for that.
-          tag_rev = `#{git} rev-list -n1 #{job.uuid.shellescape}`.chomp
-          if $? == 0
-            # We got a revision back
-            if tag_rev != job.script_version
-              # Uh oh, the tag doesn't point to the revision we were expecting.
-              # Someone has been monkeying with the job record and/or git.
-              fail_job job, "Existing tag #{job.uuid} points to commit #{tag_rev} but expected commit #{job.script_version}"
-              next
-            end
-            # we're okay (fall through to setting @job_tags below)
-          else
-            # git rev-list failed for some reason.
-            fail_job job, "'git tag' for #{job.uuid} failed but did not find any existing tag using 'git rev-list'"
-            next
-          end
-        end
-        # 'git tag' was successful, or there is an existing tag that points to the same revision.
-        @job_tags[job.uuid] = job.script_version
-      elsif @job_tags[job.uuid] != job.script_version
-        fail_job job, "Existing tag #{job.uuid} points to commit #{@job_tags[job.uuid]} but this job uses commit #{job.script_version}"
+      if not ready
+        fail_job job, "commit not present in internal repository"
         next
       end
 
-      cmd_args << crunch_job_bin
-      cmd_args << '--job-api-token'
-      cmd_args << @authorizations[job.uuid].api_token
-      cmd_args << '--job'
-      cmd_args << job.uuid
-      cmd_args << '--git-dir'
-      cmd_args << arvados_internal
+      cmd_args += [@crunch_job_bin,
+                   '--job-api-token', @authorizations[job.uuid].api_token,
+                   '--job', job.uuid,
+                   '--git-dir', @arvados_internal]
 
       $stderr.puts "dispatch: #{cmd_args.join ' '}"
 
@@ -367,7 +428,7 @@ class Dispatcher
       end
 
       $stderr.puts "dispatch: job #{job.uuid}"
-      start_banner = "dispatch: child #{t.pid} start #{Time.now.ctime.to_s}"
+      start_banner = "dispatch: child #{t.pid} start #{LogTime.now}"
       $stderr.puts start_banner
 
       @running[job.uuid] = {
@@ -452,7 +513,7 @@ class Dispatcher
         if j[:log_throttle_bytes_skipped] > 0
           message = "#{job_uuid} ! Skipped #{j[:log_throttle_bytes_skipped]} bytes of log"
           $stderr.puts message
-          j[:stderr_buf_to_flush] << "#{Time.now.ctime.to_s} #{message}\n"
+          j[:stderr_buf_to_flush] << "#{LogTime.now} #{message}\n"
         end
 
         j[:log_throttle_reset_time] = now + Rails.configuration.crunch_log_throttle_period
@@ -520,7 +581,7 @@ class Dispatcher
           if rate_limit j, line
             $stderr.print "#{job_uuid} ! " unless line.index(job_uuid)
             $stderr.puts line
-            pub_msg = "#{Time.now.ctime.to_s} #{line.strip}\n"
+            pub_msg = "#{LogTime.now} #{line.strip}\n"
             j[:stderr_buf_to_flush] << pub_msg
           end
         end
@@ -593,7 +654,7 @@ class Dispatcher
     exit_status = j_done[:wait_thr].value.exitstatus
 
     jobrecord = Job.find_by_uuid(job_done.uuid)
-    if exit_status != 75 and jobrecord.state == "Running"
+    if exit_status != EXIT_TEMPFAIL and jobrecord.state == "Running"
       # crunch-job did not return exit code 75 (see below) and left the job in
       # the "Running" state, which means there was an unhandled error.  Fail
       # the job.
@@ -619,8 +680,11 @@ class Dispatcher
       # fine.
     end
 
-    # Invalidate the per-job auth token
-    j_done[:job_auth].update_attributes expires_at: Time.now
+    # Invalidate the per-job auth token, unless the job is still queued and we
+    # might want to try it again.
+    if jobrecord.state != "Queued"
+      j_done[:job_auth].update_attributes expires_at: Time.now
+    end
 
     @running.delete job_done.uuid
   end
@@ -643,8 +707,6 @@ class Dispatcher
 
   def run
     act_as_system_user
-    @running ||= {}
-    @pipe_auth_tokens ||= { }
     $stderr.puts "dispatch: ready"
     while !$signal[:term] or @running.size > 0
       read_pipes
@@ -679,7 +741,6 @@ class Dispatcher
   protected
 
   def did_recently(thing, min_interval)
-    @did_recently ||= {}
     if !@did_recently[thing] or @did_recently[thing] < Time.now - min_interval
       @did_recently[thing] = Time.now
       false
@@ -717,4 +778,10 @@ end
 # This is how crunch-job child procs know where the "refresh" trigger file is
 ENV["CRUNCH_REFRESH_TRIGGER"] = Rails.configuration.crunch_refresh_trigger
 
+# If salloc can't allocate resources immediately, make it use our temporary
+# failure exit code.  This ensures crunch-dispatch won't mark a job failed
+# because of an issue with node allocation.  This often happens when
+# another dispatcher wins the race to allocate nodes.
+ENV["SLURM_EXIT_IMMEDIATE"] = Dispatcher::EXIT_TEMPFAIL.to_s
+
 Dispatcher.new.run