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 skip_before_filter :render_404_if_no_object, :only => :queue
9 [:repository, :script, :script_version, :script_parameters].each do |r|
12 :error => "#{r} attribute must be specified"
13 }, status: :unprocessable_entity
17 r = Commit.find_commit_range(current_user,
18 resource_attrs[:repository],
19 resource_attrs[:minimum_script_version],
20 resource_attrs[:script_version],
21 resource_attrs[:exclude_script_versions])
22 if !resource_attrs[:nondeterministic] and !resource_attrs[:no_reuse]
23 # Search for jobs where the script_version is in the list of commits
24 # returned by find_commit_range
26 Job.readable_by(current_user).where(script: resource_attrs[:script],
29 if j.nondeterministic != true and
30 j.success != false and
31 j.script_parameters == resource_attrs[:script_parameters]
32 # Record the first job in the list
36 # Ensure that all candidate jobs actually did produce the same output
37 if @object.output != j.output
48 resource_attrs[:script_version] = r[0]
51 # Don't pass these on to activerecord
52 resource_attrs.delete(:repository)
53 resource_attrs.delete(:minimum_script_version)
54 resource_attrs.delete(:exclude_script_versions)
55 resource_attrs.delete(:no_reuse)
60 reload_object_before_update
61 @object.update_attributes cancelled_at: Time.now
66 Q_UPDATE_INTERVAL = 12
67 def initialize(job, opts={})
73 yield "#{@job.uuid} finished at #{@job.finished_at}\n"
76 while not @job.started_at
77 # send a summary (job queue + available nodes) to the client
78 # every few seconds while waiting for the job to start
79 last_ack_at ||= Time.now - Q_UPDATE_INTERVAL - 1
80 if Time.now - last_ack_at >= Q_UPDATE_INTERVAL
81 nodes_in_state = {idle: 0, alloc: 0}
82 ActiveRecord::Base.uncached do
83 Node.where('hostname is not ?', nil).collect do |n|
84 if n.info[:slurm_state]
85 nodes_in_state[n.info[:slurm_state]] ||= 0
86 nodes_in_state[n.info[:slurm_state]] += 1
91 n_queued_before_me = 0
93 break if j.uuid == @job.uuid
94 n_queued_before_me += 1
98 " queue_position #{n_queued_before_me}" \
99 " queue_size #{job_queue.size}" \
100 " nodes_idle #{nodes_in_state[:idle]}" \
101 " nodes_alloc #{nodes_in_state[:alloc]}\n"
102 last_ack_at = Time.now
105 ActiveRecord::Base.uncached do
109 @redis = Redis.new(:timeout => 0)
110 if @redis.exists @job.uuid
111 # A log buffer exists. Start by showing the last few KB.
113 getrange(@job.uuid, 0 - [@opts[:buffer_size], 1].max, -1).
114 sub(/^[^\n]*\n?/, '').
120 # TODO: avoid missing log entries between getrange() above and
122 @redis.subscribe(@job.uuid) do |event|
123 event.message do |channel, msg|
125 @redis.unsubscribe @job.uuid
134 def self._log_tail_follow_requires_parameters
136 buffer_size: {type: 'integer', required: false, default: 2**13}
140 if !@object.andand.uuid
141 return render_not_found
143 if client_accepts_plain_text_stream
144 self.response.headers['Last-Modified'] = Time.now.ctime.to_s
145 self.response_body = LogStreamer.new @object, {
146 buffer_size: (params[:buffer_size].to_i rescue 2**13)
150 href: url_for(uuid: @object.uuid),
151 comment: ('To retrieve the log stream as plain text, ' +
152 'use a request header like "Accept: text/plain"')
161 is_locked_by_uuid: nil,
165 params[:order] ||= 'priority desc, created_at'
166 find_objects_for_index
170 def self._queue_requires_parameters
171 self._index_requires_parameters