X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/c402c5c3ef362f2848f0dc9081195ced9d7b84aa..6d73acc845a2dd7413fdde0742473d83bf3d0719:/services/api/app/controllers/arvados/v1/jobs_controller.rb diff --git a/services/api/app/controllers/arvados/v1/jobs_controller.rb b/services/api/app/controllers/arvados/v1/jobs_controller.rb index 41cbfd4b0e..178b48f173 100644 --- a/services/api/app/controllers/arvados/v1/jobs_controller.rb +++ b/services/api/app/controllers/arvados/v1/jobs_controller.rb @@ -3,43 +3,64 @@ class Arvados::V1::JobsController < ApplicationController accept_attribute_as_json :runtime_constraints, Hash accept_attribute_as_json :tasks_summary, Hash skip_before_filter :find_object_by_uuid, :only => :queue + skip_before_filter :render_404_if_no_object, :only => :queue - def index - want_ancestor = @where[:script_version_descends_from] - if want_ancestor - # Check for missing commit_ancestor rows, and create them if - # possible. - @objects. - dup. - includes(:commit_ancestors). # I wish Rails would let me - # specify here which - # commit_ancestors I am - # interested in. - each do |o| - if o.commit_ancestors. - select { |ca| ca.ancestor == want_ancestor }. - empty? and !o.script_version.nil? - begin - o.commit_ancestors << CommitAncestor.find_or_create_by_descendant_and_ancestor(o.script_version, want_ancestor) - rescue + def create + [:repository, :script, :script_version, :script_parameters].each do |r| + if !resource_attrs[r] + return render json: { + :error => "#{r} attribute must be specified" + }, status: :unprocessable_entity + end + end + + r = Commit.find_commit_range(current_user, + resource_attrs[:repository], + resource_attrs[:minimum_script_version], + resource_attrs[:script_version], + resource_attrs[:exclude_script_versions]) + if !resource_attrs[:nondeterministic] and !resource_attrs[:no_reuse] + # Search for jobs where the script_version is in the list of commits + # returned by find_commit_range + @object = nil + Job.readable_by(current_user).where(script: resource_attrs[:script], + script_version: r). + each do |j| + if j.nondeterministic != true and + j.success != false and + j.script_parameters == resource_attrs[:script_parameters] + # Record the first job in the list + if !@object + @object = j + end + # Ensure that all candidate jobs actually did produce the same output + if @object.output != j.output + @object = nil + break end end - o.commit_ancestors. - select { |ca| ca.ancestor == want_ancestor }. - select(&:is). - first + if @object + return show + end end - # Now it is safe to do an .includes().where() because we are no - # longer interested in jobs that have other ancestors but not - # want_ancestor. - @objects = @objects. - includes(:commit_ancestors). - where('commit_ancestors.ancestor = ? and commit_ancestors.is = ?', - want_ancestor, true) end + if r + resource_attrs[:script_version] = r[0] + end + + # Don't pass these on to activerecord + resource_attrs.delete(:minimum_script_version) + resource_attrs.delete(:exclude_script_versions) + resource_attrs.delete(:no_reuse) super end + def cancel + reload_object_before_update + @object.update_attributes cancelled_at: Time.now + show + end + class LogStreamer Q_UPDATE_INTERVAL = 12 def initialize(job, opts={}) @@ -57,10 +78,12 @@ class Arvados::V1::JobsController < ApplicationController last_ack_at ||= Time.now - Q_UPDATE_INTERVAL - 1 if Time.now - last_ack_at >= Q_UPDATE_INTERVAL nodes_in_state = {idle: 0, alloc: 0} - Node.where('hostname is not ?', nil).collect do |n| - if n.info[:slurm_state] - nodes_in_state[n.info[:slurm_state]] ||= 0 - nodes_in_state[n.info[:slurm_state]] += 1 + ActiveRecord::Base.uncached do + Node.where('hostname is not ?', nil).collect do |n| + if n.info[:slurm_state] + nodes_in_state[n.info[:slurm_state]] ||= 0 + nodes_in_state[n.info[:slurm_state]] += 1 + end end end job_queue = Job.queue @@ -78,23 +101,24 @@ class Arvados::V1::JobsController < ApplicationController last_ack_at = Time.now end sleep 3 - @job.reload + ActiveRecord::Base.uncached do + @job.reload + end end @redis = Redis.new(:timeout => 0) - @redis.subscribe(@job.uuid) do |event| - if @redis.exists @job.uuid - # A log buffer exists. Start by showing the last few KB. - @redis. - getrange(@job.uuid, 0 - [@opts[:buffer_size], 1].max, -1). - sub(/^[^\n]*\n?/, ''). - split("\n"). - each do |line| - yield "#{line}\n" - end + if @redis.exists @job.uuid + # A log buffer exists. Start by showing the last few KB. + @redis. + getrange(@job.uuid, 0 - [@opts[:buffer_size], 1].max, -1). + sub(/^[^\n]*\n?/, ''). + split("\n"). + each do |line| + yield "#{line}\n" end - # TODO: avoid duplicating the last few lines of the log - # file. Use the fact that timestamps are lexicographically - # ordered. + end + # TODO: avoid missing log entries between getrange() above and + # subscribe() below. + @redis.subscribe(@job.uuid) do |event| event.message do |channel, msg| if msg == "end" @redis.unsubscribe @job.uuid @@ -108,17 +132,25 @@ class Arvados::V1::JobsController < ApplicationController def self._log_tail_follow_requires_parameters { - buffer_size: {type: 'integer', required: false} + buffer_size: {type: 'integer', required: false, default: 2**13} } end def log_tail_follow if !@object.andand.uuid return render_not_found end - self.response.headers['Last-Modified'] = Time.now.ctime.to_s - self.response_body = LogStreamer.new @object, { - buffer_size: (params[:buffer_size] || 2**13) - } + if client_accepts_plain_text_stream + self.response.headers['Last-Modified'] = Time.now.ctime.to_s + self.response_body = LogStreamer.new @object, { + buffer_size: (params[:buffer_size].to_i rescue 2**13) + } + else + render json: { + href: url_for(uuid: @object.uuid), + comment: ('To retrieve the log stream as plain text, ' + + 'use a request header like "Accept: text/plain"') + } + end end def queue @@ -126,7 +158,8 @@ class Arvados::V1::JobsController < ApplicationController @where.merge!({ started_at: nil, is_locked_by_uuid: nil, - cancelled_at: nil + cancelled_at: nil, + success: nil }) params[:order] ||= 'priority desc, created_at' find_objects_for_index