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     if @filters.empty?  # Translate older creation parameters into filters.
  35       @filters =
  36         [["repository", "=", resource_attrs[:repository]],
  37          ["script", "=", resource_attrs[:script]],
  38          ["script_version", "not in git", params[:exclude_script_versions]],
  39         ].reject { |filter| filter.last.nil? or filter.last.empty? }
  40       if !params[:minimum_script_version].blank?
  41         @filters << ["script_version", "in git",
  42                      params[:minimum_script_version]]
  43       else
  44         add_default_git_filter("script_version", resource_attrs[:repository],
  45                                resource_attrs[:script_version])
  46       end
  47       if image_search = resource_attrs[:runtime_constraints].andand["docker_image"]
  48         if image_tag = resource_attrs[:runtime_constraints]["docker_image_tag"]
  49           image_search += ":#{image_tag}"
  50         end
  51         image_locator = Collection.
  52           for_latest_docker_image(image_search).andand.portable_data_hash
  53       else
  54         image_locator = nil
  55       end
  56       @filters << ["docker_image_locator", "=", image_locator]
  57       if sdk_version = resource_attrs[:runtime_constraints].andand["arvados_sdk_version"]
  58         add_default_git_filter("arvados_sdk_version", "arvados", sdk_version)
  59       end
  60       begin
  61         load_job_specific_filters
  62       rescue ArgumentError => error
  63         return send_error(error.message)
  64       end
  65     end
  66
  67     # Check specified filters for some reasonableness.
  68     filter_names = @filters.map { |f| f.first }.uniq
  69     ["repository", "script"].each do |req_filter|
  70       if not filter_names.include?(req_filter)
  71         return send_error("#{req_filter} filter required")
  72       end
  73     end
  74
  75     # Search for a reusable Job, and return it if found.
  76     @objects = Job.
  77       readable_by(current_user).
  78       where('state = ? or (owner_uuid = ? and state in (?))',
  79             Job::Complete, current_user.uuid, [Job::Queued, Job::Running]).
  80       where('script_parameters_digest = ?', Job.sorted_hash_digest(resource_attrs[:script_parameters])).
  81       where('nondeterministic is distinct from ?', true).
  82       order('state desc, created_at') # prefer Running jobs over Queued
  83     apply_filters
  84     @object = nil
  85     incomplete_job = nil
  86     @objects.each do |j|
  87       if j.state != Job::Complete
  88         # We'll use this if we don't find a job that has completed
  89         incomplete_job ||= j
  90         next
  91       end
  92
  93       if @object
  94         if @object.output != j.output
  95           # If two matching jobs produced different outputs, just run
  96           # a new job instead of choosing one arbitrarily.
  97           @object = nil
  98           return super
  99         end
 100         # ...and that's the only thing we need to do once we've chosen
 101         # an @object to reuse.
 102       elsif !Collection.readable_by(current_user).find_by_portable_data_hash(j.output)
 103         # As soon as the output we will end up returning (if any) is
 104         # decided, check whether it will be visible to the user; if
 105         # not, any further investigation of reusable jobs is futile.
 106         return super
 107       else
 108         @object = j
 109       end
 110     end
 111
 112     @object ||= incomplete_job
 113     if @object
 114       show
 115     else
 116       super
 117     end
 118   end
 119
 120   def cancel
 121     reload_object_before_update
 122     @object.update_attributes! state: Job::Cancelled
 123     show
 124   end
 125
 126   def lock
 127     @object.lock current_user.uuid
 128     show
 129   end
 130
 131   class LogStreamer
 132     Q_UPDATE_INTERVAL = 12
 133     def initialize(job, opts={})
 134       @job = job
 135       @opts = opts
 136     end
 137     def each
 138       if @job.finished_at
 139         yield "#{@job.uuid} finished at #{@job.finished_at}\n"
 140         return
 141       end
 142       while not @job.started_at
 143         # send a summary (job queue + available nodes) to the client
 144         # every few seconds while waiting for the job to start
 145         current_time = db_current_time
 146         last_ack_at ||= current_time - Q_UPDATE_INTERVAL - 1
 147         if current_time - last_ack_at >= Q_UPDATE_INTERVAL
 148           nodes_in_state = {idle: 0, alloc: 0}
 149           ActiveRecord::Base.uncached do
 150             Node.where('hostname is not ?', nil).collect do |n|
 151               if n.info[:slurm_state]
 152                 nodes_in_state[n.info[:slurm_state]] ||= 0
 153                 nodes_in_state[n.info[:slurm_state]] += 1
 154               end
 155             end
 156           end
 157           job_queue = Job.queue.select(:uuid)
 158           n_queued_before_me = 0
 159           job_queue.each do |j|
 160             break if j.uuid == @job.uuid
 161             n_queued_before_me += 1
 162           end
 163           yield "#{db_current_time}" \
 164             " job #{@job.uuid}" \
 165             " queue_position #{n_queued_before_me}" \
 166             " queue_size #{job_queue.count}" \
 167             " nodes_idle #{nodes_in_state[:idle]}" \
 168             " nodes_alloc #{nodes_in_state[:alloc]}\n"
 169           last_ack_at = db_current_time
 170         end
 171         sleep 3
 172         ActiveRecord::Base.uncached do
 173           @job.reload
 174         end
 175       end
 176     end
 177   end
 178
 179   def queue
 180     params[:order] ||= ['priority desc', 'created_at']
 181     load_limit_offset_order_params
 182     load_where_param
 183     @where.merge!({state: Job::Queued})
 184     return if !load_filters_param
 185     find_objects_for_index
 186     index
 187   end
 188
 189   def queue_size
 190     # Users may not be allowed to see all the jobs in the queue, so provide a
 191     # method to get just the queue size in order to get a gist of how busy the
 192     # cluster is.
 193     render :json => {:queue_size => Job.queue.size}
 194   end
 195
 196   def self._create_requires_parameters
 197     (super rescue {}).
 198       merge({
 199               find_or_create: {
 200                 type: 'boolean', required: false, default: false
 201               },
 202               filters: {
 203                 type: 'array', required: false
 204               },
 205               minimum_script_version: {
 206                 type: 'string', required: false
 207               },
 208               exclude_script_versions: {
 209                 type: 'array', required: false
 210               },
 211             })
 212   end
 213
 214   def self._queue_requires_parameters
 215     self._index_requires_parameters
 216   end
 217
 218   protected
 219
 220   def add_default_git_filter(attr_name, repo_name, refspec)
 221     # Add a filter to @filters for `attr_name` = the latest commit available
 222     # in `repo_name` at `refspec`.  No filter is added if refspec can't be
 223     # resolved.
 224     commits = Commit.find_commit_range(repo_name, nil, refspec, nil)
 225     if commit_hash = commits.first
 226       @filters << [attr_name, "=", commit_hash]
 227     end
 228   end
 229
 230   def load_job_specific_filters
 231     # Convert Job-specific @filters entries into general SQL filters.
 232     script_info = {"repository" => nil, "script" => nil}
 233     git_filters = Hash.new do |hash, key|
 234       hash[key] = {"max_version" => "HEAD", "exclude_versions" => []}
 235     end
 236     @filters.select! do |(attr, operator, operand)|
 237       if (script_info.has_key? attr) and (operator == "=")
 238         if script_info[attr].nil?
 239           script_info[attr] = operand
 240         elsif script_info[attr] != operand
 241           raise ArgumentError.new("incompatible #{attr} filters")
 242         end
 243       end
 244       case operator
 245       when "in git"
 246         git_filters[attr]["min_version"] = operand
 247         false
 248       when "not in git"
 249         git_filters[attr]["exclude_versions"] += Array.wrap(operand)
 250         false
 251       when "in docker", "not in docker"
 252         image_hashes = Array.wrap(operand).flat_map do |search_term|
 253           image_search, image_tag = search_term.split(':', 2)
 254           Collection.
 255             find_all_for_docker_image(image_search, image_tag, @read_users).
 256             map(&:portable_data_hash)
 257         end
 258         @filters << [attr, operator.sub(/ docker$/, ""), image_hashes]
 259         false
 260       else
 261         true
 262       end
 263     end
 264
 265     # Build a real script_version filter from any "not? in git" filters.
 266     git_filters.each_pair do |attr, filter|
 267       case attr
 268       when "script_version"
 269         script_info.each_pair do |key, value|
 270           if value.nil?
 271             raise ArgumentError.new("script_version filter needs #{key} filter")
 272           end
 273         end
 274         filter["repository"] = script_info["repository"]
 275         begin
 276           filter["max_version"] = resource_attrs[:script_version]
 277         rescue
 278           # Using HEAD, set earlier by the hash default, is fine.
 279         end
 280       when "arvados_sdk_version"
 281         filter["repository"] = "arvados"
 282       else
 283         raise ArgumentError.new("unknown attribute for git filter: #{attr}")
 284       end
 285       revisions = Commit.find_commit_range(filter["repository"],
 286                                            filter["min_version"],
 287                                            filter["max_version"],
 288                                            filter["exclude_versions"])
 289       if revisions.empty?
 290         raise ArgumentError.
 291           new("error searching #{filter['repository']} from " +
 292               "'#{filter['min_version']}' to '#{filter['max_version']}', " +
 293               "excluding #{filter['exclude_versions']}")
 294       end
 295       @filters.append([attr, "in", revisions])
 296     end
 297   end
 298
 299   def load_filters_param
 300     begin
 301       super
 302       load_job_specific_filters
 303     rescue ArgumentError => error
 304       send_error(error.message)
 305       false
 306     else
 307       true
 308     end
 309   end
 310 end