X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/fb3c02b38a24cda422de95f2f8b49002b841cc72..4f552c0187f8c31d94ff74485c57ef7f9888597e:/services/api/app/models/job.rb diff --git a/services/api/app/models/job.rb b/services/api/app/models/job.rb index e13446b4ae..0b2247bc21 100644 --- a/services/api/app/models/job.rb +++ b/services/api/app/models/job.rb @@ -1,37 +1,44 @@ -class Job < OrvosModel +class Job < ArvadosModel include AssignUuid include KindAndEtag include CommonApiTemplate serialize :script_parameters, Hash - serialize :resource_limits, Hash + serialize :runtime_constraints, Hash serialize :tasks_summary, Hash before_create :ensure_unique_submit_id before_create :ensure_script_version_is_commit + before_update :ensure_script_version_is_commit + after_commit :trigger_crunch_dispatch_if_cancelled, :on => :update has_many :commit_ancestors, :foreign_key => :descendant, :primary_key => :script_version class SubmitIdReused < StandardError end - api_accessible :superuser, :extend => :common do |t| + api_accessible :user, extend: :common do |t| t.add :submit_id t.add :priority t.add :script t.add :script_parameters t.add :script_version t.add :cancelled_at - t.add :cancelled_by_client - t.add :cancelled_by_user + t.add :cancelled_by_client_uuid + t.add :cancelled_by_user_uuid t.add :started_at t.add :finished_at t.add :output + t.add :output_is_persistent t.add :success t.add :running - t.add :is_locked_by + t.add :is_locked_by_uuid t.add :log - t.add :resource_limits + t.add :runtime_constraints t.add :tasks_summary t.add :dependencies + t.add :log_stream_href + t.add :log_buffer + t.add :nondeterministic + t.add :repository end def assert_finished @@ -40,14 +47,47 @@ class Job < OrvosModel running: false) end + def log_stream_href + unless self.finished_at + "#{current_api_base}/#{self.class.to_s.pluralize.underscore}/#{self.uuid}/log_tail_follow" + end + end + + def self.queue + self.where('started_at is ? and is_locked_by_uuid is ? and cancelled_at is ? and success is ?', + nil, nil, nil, nil). + order('priority desc, created_at') + end + + def self.running + self.where('running = ?', true). + order('priority desc, created_at') + end + protected + def foreign_key_attributes + super + %w(output log) + end + + def skip_uuid_read_permission_check + super + %w(cancelled_by_client_uuid) + end + def ensure_script_version_is_commit - sha1 = Commit.find_by_commit_ish(self.script_version) rescue nil - if sha1 - self.script_version = sha1 - else - raise ArgumentError.new("Specified script_version does not resolve to a commit") + if self.is_locked_by_uuid and self.started_at + # Apparently client has already decided to go for it. This is + # needed to run a local job using a local working directory + # instead of a commit-ish. + return true + end + if new_record? or script_version_changed? + sha1 = Commit.find_commit_range(current_user, nil, nil, self.script_version, nil)[0] rescue nil + if sha1 + self.script_version = sha1 + else + raise ArgumentError.new("Specified script_version does not resolve to a commit") + end end end @@ -62,25 +102,33 @@ class Job < OrvosModel def dependencies deps = {} - self.script_parameters.values.each do |v| - next unless v.is_a? String - v.match(/^(([0-9a-f]{32})\b(\+[^,]+)?,?)*$/) do |locator| - bare_locator = locator[0].gsub(/\+[^,]+/,'') - deps[bare_locator] = true + queue = self.script_parameters.values + while not queue.empty? + queue = queue.flatten.compact.collect do |v| + if v.is_a? Hash + v.values + elsif v.is_a? String + v.match(/^(([0-9a-f]{32})\b(\+[^,]+)?,?)*$/) do |locator| + deps[locator.to_s] = true + end + nil + end end end deps.keys end def permission_to_update - if is_locked_by_was and !(current_user and - current_user.uuid == is_locked_by_was) + if is_locked_by_uuid_was and !(current_user and + (current_user.uuid == is_locked_by_uuid_was or + current_user.uuid == system_user.uuid)) if script_changed? or script_parameters_changed? or script_version_changed? or - cancelled_by_client_changed? or - cancelled_by_user_changed? or - cancelled_at_changed? or + (!cancelled_at_was.nil? and + (cancelled_by_client_uuid_changed? or + cancelled_by_user_uuid_changed? or + cancelled_at_changed?)) or started_at_changed? or finished_at_changed? or running_changed? or @@ -92,21 +140,58 @@ class Job < OrvosModel return false end end - if !is_locked_by_changed? + if !is_locked_by_uuid_changed? super else if !current_user logger.warn "Anonymous user tried to change lock on #{self.class.to_s} #{uuid_was}" false - elsif is_locked_by_was and is_locked_by_was != current_user.uuid - logger.warn "User #{current_user.uuid} tried to steal lock on #{self.class.to_s} #{uuid_was} from #{is_locked_by_was}" + elsif is_locked_by_uuid_was and is_locked_by_uuid_was != current_user.uuid + logger.warn "User #{current_user.uuid} tried to steal lock on #{self.class.to_s} #{uuid_was} from #{is_locked_by_uuid_was}" false - elsif !is_locked_by.nil? and is_locked_by != current_user.uuid - logger.warn "User #{current_user.uuid} tried to lock #{self.class.to_s} #{uuid_was} with uuid #{is_locked_by}" + elsif !is_locked_by_uuid.nil? and is_locked_by_uuid != current_user.uuid + logger.warn "User #{current_user.uuid} tried to lock #{self.class.to_s} #{uuid_was} with uuid #{is_locked_by_uuid}" false else super end end end + + def update_modified_by_fields + if self.cancelled_at_changed? + # Ensure cancelled_at cannot be set to arbitrary non-now times, + # or changed once it is set. + if self.cancelled_at and not self.cancelled_at_was + self.cancelled_at = Time.now + self.cancelled_by_user_uuid = current_user.uuid + self.cancelled_by_client_uuid = current_api_client.uuid + @need_crunch_dispatch_trigger = true + else + self.cancelled_at = self.cancelled_at_was + self.cancelled_by_user_uuid = self.cancelled_by_user_uuid_was + self.cancelled_by_client_uuid = self.cancelled_by_client_uuid_was + end + end + super + end + + def trigger_crunch_dispatch_if_cancelled + if @need_crunch_dispatch_trigger + File.open(Rails.configuration.crunch_refresh_trigger, 'wb') do + # That's all, just create/touch a file for crunch-job to see. + end + end + end + + def log_buffer + begin + @@redis ||= Redis.new(:timeout => 0) + if @@redis.exists uuid + @@redis.getrange(uuid, 0 - 2**10, -1) + end + rescue Redis::CannotConnectError + return '(not available)' + end + end end