1 class Job < ArvadosModel
4 include CommonApiTemplate
5 serialize :components, Hash
6 attr_protected :arvados_sdk_version, :docker_image_locator
7 serialize :script_parameters, Hash
8 serialize :runtime_constraints, Hash
9 serialize :tasks_summary, Hash
10 before_create :ensure_unique_submit_id
11 after_commit :trigger_crunch_dispatch_if_cancelled, :on => :update
12 before_validation :set_priority
13 before_validation :update_state_from_old_state_attrs
14 before_validation :update_script_parameters_digest
15 validate :ensure_script_version_is_commit
16 validate :find_docker_image_locator
17 validate :find_arvados_sdk_version
18 validate :validate_status
19 validate :validate_state_change
20 validate :ensure_no_collection_uuids_in_script_params
21 before_save :tag_version_in_internal_repository
22 before_save :update_timestamps_when_state_changes
24 has_many :commit_ancestors, :foreign_key => :descendant, :primary_key => :script_version
25 has_many(:nodes, foreign_key: :job_uuid, primary_key: :uuid)
27 class SubmitIdReused < StandardError
30 api_accessible :user, extend: :common do |t|
34 t.add :script_parameters
37 t.add :cancelled_by_client_uuid
38 t.add :cancelled_by_user_uuid
45 t.add :is_locked_by_uuid
47 t.add :runtime_constraints
49 t.add :nondeterministic
51 t.add :supplied_script_version
52 t.add :arvados_sdk_version
53 t.add :docker_image_locator
60 # Supported states for a job
63 (Running = 'Running'),
64 (Cancelled = 'Cancelled'),
66 (Complete = 'Complete'),
70 update_attributes(finished_at: finished_at || db_current_time,
71 success: success.nil? ? false : success,
80 self.where('state = ?', Queued).order('priority desc, created_at')
84 # We used to report this accurately, but the implementation made queue
85 # API requests O(n**2) for the size of the queue. See #8800.
86 # We've soft-disabled it because it's not clear we even want this
87 # functionality: now that we have Node Manager with support for multiple
88 # node sizes, "queue position" tells you very little about when a job will
90 state == Queued ? 0 : nil
94 self.where('running = ?', true).
95 order('priority desc, created_at')
98 def lock locked_by_uuid
100 unless self.state == Queued and self.is_locked_by_uuid.nil?
101 raise AlreadyLockedError
104 self.is_locked_by_uuid = locked_by_uuid
109 def update_script_parameters_digest
110 self.script_parameters_digest = self.class.sorted_hash_digest(script_parameters)
113 def self.searchable_columns operator
114 super - ["script_parameters_digest"]
117 def self.load_job_specific_filters attrs, orig_filters, read_users
118 # Convert Job-specific @filters entries into general SQL filters.
119 script_info = {"repository" => nil, "script" => nil}
120 git_filters = Hash.new do |hash, key|
121 hash[key] = {"max_version" => "HEAD", "exclude_versions" => []}
124 orig_filters.each do |attr, operator, operand|
125 if (script_info.has_key? attr) and (operator == "=")
126 if script_info[attr].nil?
127 script_info[attr] = operand
128 elsif script_info[attr] != operand
129 raise ArgumentError.new("incompatible #{attr} filters")
134 git_filters[attr]["min_version"] = operand
136 git_filters[attr]["exclude_versions"] += Array.wrap(operand)
137 when "in docker", "not in docker"
138 image_hashes = Array.wrap(operand).flat_map do |search_term|
139 image_search, image_tag = search_term.split(':', 2)
141 find_all_for_docker_image(image_search, image_tag, read_users).
142 map(&:portable_data_hash)
144 filters << [attr, operator.sub(/ docker$/, ""), image_hashes]
146 filters << [attr, operator, operand]
150 # Build a real script_version filter from any "not? in git" filters.
151 git_filters.each_pair do |attr, filter|
153 when "script_version"
154 script_info.each_pair do |key, value|
156 raise ArgumentError.new("script_version filter needs #{key} filter")
159 filter["repository"] = script_info["repository"]
160 if attrs[:script_version]
161 filter["max_version"] = attrs[:script_version]
163 # Using HEAD, set earlier by the hash default, is fine.
165 when "arvados_sdk_version"
166 filter["repository"] = "arvados"
168 raise ArgumentError.new("unknown attribute for git filter: #{attr}")
170 revisions = Commit.find_commit_range(filter["repository"],
171 filter["min_version"],
172 filter["max_version"],
173 filter["exclude_versions"])
176 new("error searching #{filter['repository']} from " +
177 "'#{filter['min_version']}' to '#{filter['max_version']}', " +
178 "excluding #{filter['exclude_versions']}")
180 filters.append([attr, "in", revisions])
186 def self.find_reusable attrs, params, filters, read_users
187 if filters.empty? # Translate older creation parameters into filters.
189 [["repository", "=", attrs[:repository]],
190 ["script", "=", attrs[:script]],
191 ["script_version", "not in git", params[:exclude_script_versions]],
192 ].reject { |filter| filter.last.nil? or filter.last.empty? }
193 if !params[:minimum_script_version].blank?
194 filters << ["script_version", "in git",
195 params[:minimum_script_version]]
197 filters += default_git_filters("script_version", attrs[:repository],
198 attrs[:script_version])
200 if image_search = attrs[:runtime_constraints].andand["docker_image"]
201 if image_tag = attrs[:runtime_constraints]["docker_image_tag"]
202 image_search += ":#{image_tag}"
204 image_locator = Collection.
205 for_latest_docker_image(image_search).andand.portable_data_hash
209 filters << ["docker_image_locator", "=", image_locator]
210 if sdk_version = attrs[:runtime_constraints].andand["arvados_sdk_version"]
211 filters += default_git_filters("arvados_sdk_version", "arvados", sdk_version)
213 filters = load_job_specific_filters(attrs, filters, read_users)
216 # Check specified filters for some reasonableness.
217 filter_names = filters.map { |f| f.first }.uniq
218 ["repository", "script"].each do |req_filter|
219 if not filter_names.include?(req_filter)
220 return send_error("#{req_filter} filter required")
224 # Search for a reusable Job, and return it if found.
226 readable_by(current_user).
227 where('state = ? or (owner_uuid = ? and state in (?))',
228 Job::Complete, current_user.uuid, [Job::Queued, Job::Running]).
229 where('script_parameters_digest = ?', Job.sorted_hash_digest(attrs[:script_parameters])).
230 where('nondeterministic is distinct from ?', true).
231 order('state desc, created_at') # prefer Running jobs over Queued
232 candidates = apply_filters candidates, filters
235 candidates.each do |j|
236 if j.state != Job::Complete
237 # We'll use this if we don't find a job that has completed
243 # We have already decided not to reuse any completed job
246 if chosen.output != j.output
247 # If two matching jobs produced different outputs, run a new
248 # job (or use one that's already running/queued) instead of
249 # choosing one arbitrarily.
252 # ...and that's the only thing we need to do once we've chosen
254 elsif !Collection.readable_by(current_user).find_by_portable_data_hash(j.output)
255 # As soon as the output we will end up returning (if any) is
256 # decided, check whether it will be visible to the user; if
257 # not, any further investigation of reusable jobs is futile.
263 chosen || incomplete_job
266 def self.default_git_filters(attr_name, repo_name, refspec)
267 # Add a filter to @filters for `attr_name` = the latest commit available
268 # in `repo_name` at `refspec`. No filter is added if refspec can't be
270 commits = Commit.find_commit_range(repo_name, nil, refspec, nil)
271 if commit_hash = commits.first
272 [[attr_name, "=", commit_hash]]
280 def self.sorted_hash_digest h
281 Digest::MD5.hexdigest(Oj.dump(deep_sort_hash(h)))
284 def self.deep_sort_hash x
286 x.sort.collect do |k, v|
287 [k, deep_sort_hash(v)]
290 x.collect { |v| deep_sort_hash(v) }
296 def foreign_key_attributes
297 super + %w(output log)
300 def skip_uuid_read_permission_check
301 super + %w(cancelled_by_client_uuid)
304 def skip_uuid_existence_check
305 super + %w(output log)
309 if self.priority.nil?
315 def ensure_script_version_is_commit
317 # Apparently client has already decided to go for it. This is
318 # needed to run a local job using a local working directory
319 # instead of a commit-ish.
322 if new_record? or repository_changed? or script_version_changed?
323 sha1 = Commit.find_commit_range(repository,
324 nil, script_version, nil).first
326 errors.add :script_version, "#{script_version} does not resolve to a commit"
329 if supplied_script_version.nil? or supplied_script_version.empty?
330 self.supplied_script_version = script_version
332 self.script_version = sha1
337 def tag_version_in_internal_repository
339 # No point now. See ensure_script_version_is_commit.
342 # Won't be saved, and script_version might not even be valid.
344 elsif new_record? or repository_changed? or script_version_changed?
348 Commit.tag_in_internal_repository repository, script_version, uuid
356 def ensure_unique_submit_id
358 if Job.where('submit_id=?',self.submit_id).first
359 raise SubmitIdReused.new
365 def resolve_runtime_constraint(key, attr_sym)
366 if ((runtime_constraints.is_a? Hash) and
367 (search = runtime_constraints[key]))
368 ok, result = yield search
370 ok, result = true, nil
373 send("#{attr_sym}=".to_sym, result)
375 errors.add(attr_sym, result)
380 def find_arvados_sdk_version
381 resolve_runtime_constraint("arvados_sdk_version",
382 :arvados_sdk_version) do |git_search|
383 commits = Commit.find_commit_range("arvados",
384 nil, git_search, nil)
386 [false, "#{git_search} does not resolve to a commit"]
387 elsif not runtime_constraints["docker_image"]
388 [false, "cannot be specified without a Docker image constraint"]
390 [true, commits.first]
395 def find_docker_image_locator
396 runtime_constraints['docker_image'] =
397 Rails.configuration.default_docker_image_for_jobs if ((runtime_constraints.is_a? Hash) and
398 (runtime_constraints['docker_image']).nil? and
399 Rails.configuration.default_docker_image_for_jobs)
400 resolve_runtime_constraint("docker_image",
401 :docker_image_locator) do |image_search|
402 image_tag = runtime_constraints['docker_image_tag']
403 if coll = Collection.for_latest_docker_image(image_search, image_tag)
404 [true, coll.portable_data_hash]
406 [false, "not found for #{image_search}"]
411 def permission_to_update
412 if is_locked_by_uuid_was and !(current_user and
413 (current_user.uuid == is_locked_by_uuid_was or
414 current_user.uuid == system_user.uuid))
415 if script_changed? or
416 script_parameters_changed? or
417 script_version_changed? or
418 (!cancelled_at_was.nil? and
419 (cancelled_by_client_uuid_changed? or
420 cancelled_by_user_uuid_changed? or
421 cancelled_at_changed?)) or
422 started_at_changed? or
423 finished_at_changed? or
428 tasks_summary_changed? or
431 logger.warn "User #{current_user.uuid if current_user} tried to change protected job attributes on locked #{self.class.to_s} #{uuid_was}"
435 if !is_locked_by_uuid_changed?
439 logger.warn "Anonymous user tried to change lock on #{self.class.to_s} #{uuid_was}"
441 elsif is_locked_by_uuid_was and is_locked_by_uuid_was != current_user.uuid
442 logger.warn "User #{current_user.uuid} tried to steal lock on #{self.class.to_s} #{uuid_was} from #{is_locked_by_uuid_was}"
444 elsif !is_locked_by_uuid.nil? and is_locked_by_uuid != current_user.uuid
445 logger.warn "User #{current_user.uuid} tried to lock #{self.class.to_s} #{uuid_was} with uuid #{is_locked_by_uuid}"
453 def update_modified_by_fields
454 if self.cancelled_at_changed?
455 # Ensure cancelled_at cannot be set to arbitrary non-now times,
456 # or changed once it is set.
457 if self.cancelled_at and not self.cancelled_at_was
458 self.cancelled_at = db_current_time
459 self.cancelled_by_user_uuid = current_user.uuid
460 self.cancelled_by_client_uuid = current_api_client.andand.uuid
461 @need_crunch_dispatch_trigger = true
463 self.cancelled_at = self.cancelled_at_was
464 self.cancelled_by_user_uuid = self.cancelled_by_user_uuid_was
465 self.cancelled_by_client_uuid = self.cancelled_by_client_uuid_was
471 def trigger_crunch_dispatch_if_cancelled
472 if @need_crunch_dispatch_trigger
473 File.open(Rails.configuration.crunch_refresh_trigger, 'wb') do
474 # That's all, just create/touch a file for crunch-job to see.
479 def update_timestamps_when_state_changes
480 return if not (state_changed? or new_record?)
484 self.started_at ||= db_current_time
485 when Failed, Complete
486 self.finished_at ||= db_current_time
488 self.cancelled_at ||= db_current_time
491 # TODO: Remove the following case block when old "success" and
492 # "running" attrs go away. Until then, this ensures we still
493 # expose correct success/running flags to older clients, even if
494 # some new clients are writing only the new state attribute.
502 when Cancelled, Failed
509 self.running ||= false # Default to false instead of nil.
511 @need_crunch_dispatch_trigger = true
516 def update_state_from_old_state_attrs
517 # If a client has touched the legacy state attrs, update the
518 # "state" attr to agree with the updated values of the legacy
521 # TODO: Remove this method when old "success" and "running" attrs
523 if cancelled_at_changed? or
528 self.state = Cancelled
529 elsif success == false
531 elsif success == true
532 self.state = Complete
533 elsif running == true
543 if self.state.in?(States)
546 errors.add :state, "#{state.inspect} must be one of: #{States.inspect}"
551 def validate_state_change
553 if self.state_changed?
554 ok = case self.state_was
556 # state isn't set yet
559 # Permit going from queued to any state
562 # From running, may only transition to a finished state
563 [Complete, Failed, Cancelled].include? self.state
564 when Complete, Failed, Cancelled
565 # Once in a finished state, don't permit any more state changes
568 # Any other state transition is also invalid
572 errors.add :state, "invalid change from #{self.state_was} to #{self.state}"
578 def ensure_no_collection_uuids_in_script_params
579 # recursive_hash_search searches recursively through hashes and
580 # arrays in 'thing' for string fields matching regular expression
581 # 'pattern'. Returns true if pattern is found, false otherwise.
582 def recursive_hash_search thing, pattern
585 return true if recursive_hash_search v, pattern
587 elsif thing.is_a? Array
589 return true if recursive_hash_search k, pattern
591 elsif thing.is_a? String
592 return true if thing.match pattern
597 # Fail validation if any script_parameters field includes a string containing a
598 # collection uuid pattern.
599 if self.script_parameters_changed?
600 if recursive_hash_search(self.script_parameters, Collection.uuid_regex)
601 self.errors.add :script_parameters, "must use portable_data_hash instead of collection uuid"