services/api/app/controllers/arvados/v1/jobs_controller.rb

   1 class Arvados::V1::JobsController < ApplicationController
   2   accept_attribute_as_json :script_parameters, Hash
   3   accept_attribute_as_json :runtime_constraints, Hash
   4   accept_attribute_as_json :tasks_summary, Hash
   5   skip_before_filter :find_object_by_uuid, :only => :queue
   6
   7   def index
   8     want_ancestor = @where[:script_version_descends_from]
   9     if want_ancestor
  10       # Check for missing commit_ancestor rows, and create them if
  11       # possible.
  12       @objects.
  13         dup.
  14         includes(:commit_ancestors). # I wish Rails would let me
  15                                      # specify here which
  16                                      # commit_ancestors I am
  17                                      # interested in.
  18         each do |o|
  19         if o.commit_ancestors.
  20             select { |ca| ca.ancestor == want_ancestor }.
  21             empty? and !o.script_version.nil?
  22           begin
  23             o.commit_ancestors << CommitAncestor.find_or_create_by_descendant_and_ancestor(o.script_version, want_ancestor)
  24           rescue
  25           end
  26         end
  27         o.commit_ancestors.
  28           select { |ca| ca.ancestor == want_ancestor }.
  29           select(&:is).
  30           first
  31       end
  32       # Now it is safe to do an .includes().where() because we are no
  33       # longer interested in jobs that have other ancestors but not
  34       # want_ancestor.
  35       @objects = @objects.
  36         includes(:commit_ancestors).
  37         where('commit_ancestors.ancestor = ? and commit_ancestors.is = ?',
  38               want_ancestor, true)
  39     end
  40     super
  41   end
  42
  43   def cancel
  44     @object.update_attributes cancelled_at: Time.now
  45     show
  46   end
  47
  48   class LogStreamer
  49     Q_UPDATE_INTERVAL = 12
  50     def initialize(job, opts={})
  51       @job = job
  52       @opts = opts
  53     end
  54     def each
  55       if @job.finished_at
  56         yield "#{@job.uuid} finished at #{@job.finished_at}\n"
  57         return
  58       end
  59       while not @job.started_at
  60         # send a summary (job queue + available nodes) to the client
  61         # every few seconds while waiting for the job to start
  62         last_ack_at ||= Time.now - Q_UPDATE_INTERVAL - 1
  63         if Time.now - last_ack_at >= Q_UPDATE_INTERVAL
  64           nodes_in_state = {idle: 0, alloc: 0}
  65           ActiveRecord::Base.uncached do
  66             Node.where('hostname is not ?', nil).collect do |n|
  67               if n.info[:slurm_state]
  68                 nodes_in_state[n.info[:slurm_state]] ||= 0
  69                 nodes_in_state[n.info[:slurm_state]] += 1
  70               end
  71             end
  72           end
  73           job_queue = Job.queue
  74           n_queued_before_me = 0
  75           job_queue.each do |j|
  76             break if j.uuid == @job.uuid
  77             n_queued_before_me += 1
  78           end
  79           yield "#{Time.now}" \
  80             " job #{@job.uuid}" \
  81             " queue_position #{n_queued_before_me}" \
  82             " queue_size #{job_queue.size}" \
  83             " nodes_idle #{nodes_in_state[:idle]}" \
  84             " nodes_alloc #{nodes_in_state[:alloc]}\n"
  85           last_ack_at = Time.now
  86         end
  87         sleep 3
  88         ActiveRecord::Base.uncached do
  89           @job.reload
  90         end
  91       end
  92       @redis = Redis.new(:timeout => 0)
  93       if @redis.exists @job.uuid
  94         # A log buffer exists. Start by showing the last few KB.
  95         @redis.
  96           getrange(@job.uuid, 0 - [@opts[:buffer_size], 1].max, -1).
  97           sub(/^[^\n]*\n?/, '').
  98           split("\n").
  99           each do |line|
 100           yield "#{line}\n"
 101         end
 102       end
 103       # TODO: avoid missing log entries between getrange() above and
 104       # subscribe() below.
 105       @redis.subscribe(@job.uuid) do |event|
 106         event.message do |channel, msg|
 107           if msg == "end"
 108             @redis.unsubscribe @job.uuid
 109           else
 110             yield "#{msg}\n"
 111           end
 112         end
 113       end
 114     end
 115   end
 116
 117   def self._log_tail_follow_requires_parameters
 118     {
 119       buffer_size: {type: 'integer', required: false, default: 2**13}
 120     }
 121   end
 122   def log_tail_follow
 123     if !@object.andand.uuid
 124       return render_not_found
 125     end
 126     if client_accepts_plain_text_stream
 127       self.response.headers['Last-Modified'] = Time.now.ctime.to_s
 128       self.response_body = LogStreamer.new @object, {
 129         buffer_size: (params[:buffer_size].to_i rescue 2**13)
 130       }
 131     else
 132       render json: {
 133         href: url_for(uuid: @object.uuid),
 134         comment: ('To retrieve the log stream as plain text, ' +
 135                   'use a request header like "Accept: text/plain"')
 136       }
 137     end
 138   end
 139
 140   def queue
 141     load_where_param
 142     @where.merge!({
 143                     started_at: nil,
 144                     is_locked_by_uuid: nil,
 145                     cancelled_at: nil
 146                   })
 147     params[:order] ||= 'priority desc, created_at'
 148     find_objects_for_index
 149     index
 150   end
 151
 152   def self._queue_requires_parameters
 153     self._index_requires_parameters
 154   end
 155 end