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

   1 class Arvados::V1::JobsController < ApplicationController
   2   accept_attribute_as_json :components, Hash
   3   accept_attribute_as_json :script_parameters, Hash
   4   accept_attribute_as_json :runtime_constraints, Hash
   5   accept_attribute_as_json :tasks_summary, Hash
   6   skip_before_filter :find_object_by_uuid, :only => [:queue, :queue_size]
   7   skip_before_filter :render_404_if_no_object, :only => [:queue, :queue_size]
   8
   9   include DbCurrentTime
  10
  11   def create
  12     [:repository, :script, :script_version, :script_parameters].each do |r|
  13       if !resource_attrs[r]
  14         return send_error("#{r} attribute must be specified",
  15                           status: :unprocessable_entity)
  16       end
  17     end
  18
  19     # We used to ask for the minimum_, exclude_, and no_reuse params
  20     # in the job resource. Now we advertise them as flags that alter
  21     # the behavior of the create action.
  22     [:minimum_script_version, :exclude_script_versions].each do |attr|
  23       if resource_attrs.has_key? attr
  24         params[attr] = resource_attrs.delete attr
  25       end
  26     end
  27     if resource_attrs.has_key? :no_reuse
  28       params[:find_or_create] = !resource_attrs.delete(:no_reuse)
  29     end
  30
  31     return super if !params[:find_or_create]
  32     return if !load_filters_param
  33
  34     begin
  35       @object = Job.find_reusable(resource_attrs, params, @filters, @read_users)
  36     rescue ArgumentError => error
  37       return send_error(error.message)
  38     end
  39
  40     if @object
  41       show
  42     else
  43       super
  44     end
  45   end
  46
  47   def cancel
  48     reload_object_before_update
  49     @object.cancel cascade: params[:cascade]
  50     show
  51   end
  52
  53   def lock
  54     @object.lock current_user.uuid
  55     show
  56   end
  57
  58   class LogStreamer
  59     Q_UPDATE_INTERVAL = 12
  60     def initialize(job, opts={})
  61       @job = job
  62       @opts = opts
  63     end
  64     def each
  65       if @job.finished_at
  66         yield "#{@job.uuid} finished at #{@job.finished_at}\n"
  67         return
  68       end
  69       while not @job.started_at
  70         # send a summary (job queue + available nodes) to the client
  71         # every few seconds while waiting for the job to start
  72         current_time = db_current_time
  73         last_ack_at ||= current_time - Q_UPDATE_INTERVAL - 1
  74         if current_time - last_ack_at >= Q_UPDATE_INTERVAL
  75           nodes_in_state = {idle: 0, alloc: 0}
  76           ActiveRecord::Base.uncached do
  77             Node.where('hostname is not ?', nil).collect do |n|
  78               if n.info[:slurm_state]
  79                 nodes_in_state[n.info[:slurm_state]] ||= 0
  80                 nodes_in_state[n.info[:slurm_state]] += 1
  81               end
  82             end
  83           end
  84           job_queue = Job.queue.select(:uuid)
  85           n_queued_before_me = 0
  86           job_queue.each do |j|
  87             break if j.uuid == @job.uuid
  88             n_queued_before_me += 1
  89           end
  90           yield "#{db_current_time}" \
  91             " job #{@job.uuid}" \
  92             " queue_position #{n_queued_before_me}" \
  93             " queue_size #{job_queue.count}" \
  94             " nodes_idle #{nodes_in_state[:idle]}" \
  95             " nodes_alloc #{nodes_in_state[:alloc]}\n"
  96           last_ack_at = db_current_time
  97         end
  98         sleep 3
  99         ActiveRecord::Base.uncached do
 100           @job.reload
 101         end
 102       end
 103     end
 104   end
 105
 106   def queue
 107     params[:order] ||= ['priority desc', 'created_at']
 108     load_limit_offset_order_params
 109     load_where_param
 110     @where.merge!({state: Job::Queued})
 111     return if !load_filters_param
 112     find_objects_for_index
 113     index
 114   end
 115
 116   def queue_size
 117     # Users may not be allowed to see all the jobs in the queue, so provide a
 118     # method to get just the queue size in order to get a gist of how busy the
 119     # cluster is.
 120     render :json => {:queue_size => Job.queue.size}
 121   end
 122
 123   def self._create_requires_parameters
 124     (super rescue {}).
 125       merge({
 126               find_or_create: {
 127                 type: 'boolean', required: false, default: false
 128               },
 129               filters: {
 130                 type: 'array', required: false
 131               },
 132               minimum_script_version: {
 133                 type: 'string', required: false
 134               },
 135               exclude_script_versions: {
 136                 type: 'array', required: false
 137               },
 138             })
 139   end
 140
 141   def self._queue_requires_parameters
 142     self._index_requires_parameters
 143   end
 144
 145   protected
 146
 147   def load_filters_param
 148     begin
 149       super
 150       attrs = resource_attrs rescue {}
 151       @filters = Job.load_job_specific_filters attrs, @filters, @read_users
 152     rescue ArgumentError => error
 153       send_error(error.message)
 154       false
 155     else
 156       true
 157     end
 158   end
 159 end