+require 'log_reuse_info'
require 'safe_json'
class Job < ArvadosModel
include KindAndEtag
include CommonApiTemplate
extend CurrentApiClient
+ extend LogReuseInfo
serialize :components, Hash
attr_protected :arvados_sdk_version, :docker_image_locator
serialize :script_parameters, Hash
@need_crunch_dispatch_trigger = false
end
+ def self.limit_index_columns_read
+ ["components"]
+ end
+
def assert_finished
update_attributes(finished_at: finished_at || db_current_time,
success: success.nil? ? false : success,
image_hashes = Array.wrap(operand).flat_map do |search_term|
image_search, image_tag = search_term.split(':', 2)
Collection.
- find_all_for_docker_image(image_search, image_tag, read_users).
+ find_all_for_docker_image(image_search, image_tag, read_users, filter_compatible_format: false).
map(&:portable_data_hash)
end
filters << [attr, operator.sub(/ docker$/, ""), image_hashes]
else
image_locator = nil
end
- filters << ["docker_image_locator", "=",
- Collection.docker_migration_pdh(read_users, image_locator)]
+ filters << ["docker_image_locator", "=", image_locator]
if sdk_version = attrs[:runtime_constraints].andand["arvados_sdk_version"]
filters += default_git_filters("arvados_sdk_version", "arvados", sdk_version)
end
end
# Search for a reusable Job, and return it if found.
- candidates = Job.
- readable_by(current_user).
- where('state = ? or (owner_uuid = ? and state in (?))',
- Job::Complete, current_user.uuid, [Job::Queued, Job::Running]).
- where('script_parameters_digest = ?', Job.sorted_hash_digest(attrs[:script_parameters])).
- where('nondeterministic is distinct from ?', true).
- order('state desc, created_at') # prefer Running jobs over Queued
+ candidates = Job.readable_by(current_user)
+ log_reuse_info { "starting with #{candidates.count} jobs readable by current user #{current_user.uuid}" }
+
+ candidates = candidates.where(
+ 'state = ? or (owner_uuid = ? and state in (?))',
+ Job::Complete, current_user.uuid, [Job::Queued, Job::Running])
+ log_reuse_info(candidates) { "after filtering on job state ((state=Complete) or (state=Queued/Running and (submitted by current user)))" }
+
+ digest = Job.sorted_hash_digest(attrs[:script_parameters])
+ candidates = candidates.where('script_parameters_digest = ?', digest)
+ log_reuse_info(candidates) { "after filtering on script_parameters_digest #{digest}" }
+
+ candidates = candidates.where('nondeterministic is distinct from ?', true)
+ log_reuse_info(candidates) { "after filtering on !nondeterministic" }
+
+ # prefer Running jobs over Queued
+ candidates = candidates.order('state desc, created_at')
+
candidates = apply_filters candidates, filters
+ log_reuse_info(candidates) { "after filtering on repo, script, and custom filters #{filters.inspect}" }
+
chosen = nil
incomplete_job = nil
candidates.each do |j|
if j.state != Job::Complete
- # We'll use this if we don't find a job that has completed
- incomplete_job ||= j
- next
- end
-
- if chosen == false
- # We have already decided not to reuse any completed job
- next
+ if !incomplete_job
+ # We'll use this if we don't find a job that has completed
+ log_reuse_info { "job #{j.uuid} is reusable, but unfinished; continuing search for completed jobs" }
+ incomplete_job = j
+ else
+ log_reuse_info { "job #{j.uuid} is unfinished and we already have #{incomplete_job.uuid}; ignoring" }
+ end
+ elsif chosen == false
+ # Ignore: we have already decided not to reuse any completed
+ # job.
+ log_reuse_info { "job #{j.uuid} with output #{j.output} ignored, see above" }
+ elsif Rails.configuration.reuse_job_if_outputs_differ
+ if Collection.readable_by(current_user).find_by_portable_data_hash(j.output)
+ log_reuse_info { "job #{j.uuid} with output #{j.output} is reusable; decision is final." }
+ return j
+ else
+ # Ignore: keep locking for an incomplete job or one whose
+ # output is readable.
+ log_reuse_info { "job #{j.uuid} output #{j.output} unavailable to user; continuing search" }
+ end
elsif chosen
if chosen.output != j.output
# If two matching jobs produced different outputs, run a new
# job (or use one that's already running/queued) instead of
# choosing one arbitrarily.
+ log_reuse_info { "job #{j.uuid} output #{j.output} disagrees; forgetting about #{chosen.uuid} and ignoring any other finished jobs (see reuse_job_if_outputs_differ in application.default.yml)" }
chosen = false
+ else
+ log_reuse_info { "job #{j.uuid} output #{j.output} agrees with chosen #{chosen.uuid}; continuing search in case other candidates have different outputs" }
end
# ...and that's the only thing we need to do once we've chosen
# a job to reuse.
elsif !Collection.readable_by(current_user).find_by_portable_data_hash(j.output)
- # As soon as the output we will end up returning (if any) is
- # decided, check whether it will be visible to the user; if
- # not, any further investigation of reusable jobs is futile.
+ # This user cannot read the output of this job. Any other
+ # completed job will have either the same output (making it
+ # unusable) or a different output (making it unusable because
+ # reuse_job_if_outputs_different is turned off). Therefore,
+ # any further investigation of reusable jobs is futile.
+ log_reuse_info { "job #{j.uuid} output #{j.output} is unavailable to user; this means no finished job can be reused (see reuse_job_if_outputs_differ in application.default.yml)" }
chosen = false
else
+ log_reuse_info { "job #{j.uuid} with output #{j.output} can be reused; continuing search in case other candidates have different outputs" }
chosen = j
end
end
- chosen || incomplete_job
+ j = chosen || incomplete_job
+ if j
+ log_reuse_info { "done, #{j.uuid} was selected" }
+ else
+ log_reuse_info { "done, nothing suitable" }
+ end
+ return j
end
def self.default_git_filters(attr_name, repo_name, refspec)
protected
def self.sorted_hash_digest h
- Digest::MD5.hexdigest(SafeJSON.dump(deep_sort_hash(h)))
+ Digest::MD5.hexdigest(Oj.dump(deep_sort_hash(h)))
end
def foreign_key_attributes
[false, "not found for #{image_search}"]
end
end
- Rails.logger.info("docker_image_locator is #{docker_image_locator}")
- if docker_image_locator && docker_image_locator_changed?
- self.docker_image_locator =
- Collection.docker_migration_pdh([current_user], docker_image_locator)
- end
- Rails.logger.info("docker_image_locator is #{docker_image_locator}")
end
def permission_to_update
projects / arvados.git / blobdiff