require 'log_reuse_info'
require 'whitelist_update'
require 'safe_json'
+require 'update_priority'
class Container < ArvadosModel
include ArvadosModelUpdates
serialize :command, Array
serialize :scheduling_parameters, Hash
serialize :secret_mounts, Hash
+ serialize :runtime_status, Hash
before_validation :fill_field_defaults, :if => :new_record?
before_validation :set_timestamps
validates :command, :container_image, :output_path, :cwd, :priority, { presence: true }
validates :priority, numericality: { only_integer: true, greater_than_or_equal_to: 0 }
+ validate :validate_runtime_status
validate :validate_state_change
validate :validate_change
validate :validate_lock
before_save :sort_serialized_attrs
before_save :update_secret_mounts_md5
before_save :scrub_secret_mounts
+ before_save :clear_runtime_status_when_queued
after_save :handle_completed
after_save :propagate_priority
+ after_commit { UpdatePriority.run_update_thread }
has_many :container_requests, :foreign_key => :container_uuid, :class_name => 'ContainerRequest', :primary_key => :uuid
belongs_to :auth, :class_name => 'ApiClientAuthorization', :foreign_key => :auth_uuid, :primary_key => :uuid
t.add :priority
t.add :progress
t.add :runtime_constraints
+ t.add :runtime_status
t.add :started_at
t.add :state
t.add :auth_uuid
# Update the priority of child container requests to match new
# priority of the parent container (ignoring requests with no
# container assigned, because their priority doesn't matter).
- ActiveRecord::Base.connection.execute('LOCK container_requests, containers IN EXCLUSIVE MODE')
ContainerRequest.
where(requesting_container_uuid: self.uuid,
state: ContainerRequest::Committed).
def self.find_reusable(attrs)
log_reuse_info { "starting with #{Container.all.count} container records in database" }
- candidates = Container.where_serialized(:command, attrs[:command])
+ candidates = Container.where_serialized(:command, attrs[:command], md5: true)
log_reuse_info(candidates) { "after filtering on command #{attrs[:command].inspect}" }
candidates = candidates.where('cwd = ?', attrs[:cwd])
log_reuse_info(candidates) { "after filtering on cwd #{attrs[:cwd].inspect}" }
- candidates = candidates.where_serialized(:environment, attrs[:environment])
+ candidates = candidates.where_serialized(:environment, attrs[:environment], md5: true)
log_reuse_info(candidates) { "after filtering on environment #{attrs[:environment].inspect}" }
candidates = candidates.where('output_path = ?', attrs[:output_path])
candidates = candidates.where('container_image = ?', image)
log_reuse_info(candidates) { "after filtering on container_image #{image.inspect} (resolved from #{attrs[:container_image].inspect})" }
- candidates = candidates.where_serialized(:mounts, resolve_mounts(attrs[:mounts]))
+ candidates = candidates.where_serialized(:mounts, resolve_mounts(attrs[:mounts]), md5: true)
log_reuse_info(candidates) { "after filtering on mounts #{attrs[:mounts].inspect}" }
- candidates = candidates.where('secret_mounts_md5 = ?', Digest::MD5.hexdigest(SafeJSON.dump(self.deep_sort_hash(attrs[:secret_mounts]))))
- log_reuse_info(candidates) { "after filtering on mounts #{attrs[:mounts].inspect}" }
+ secret_mounts_md5 = Digest::MD5.hexdigest(SafeJSON.dump(self.deep_sort_hash(attrs[:secret_mounts])))
+ candidates = candidates.where('secret_mounts_md5 = ?', secret_mounts_md5)
+ log_reuse_info(candidates) { "after filtering on secret_mounts_md5 #{secret_mounts_md5.inspect}" }
- candidates = candidates.where_serialized(:runtime_constraints, resolve_runtime_constraints(attrs[:runtime_constraints]))
+ candidates = candidates.where_serialized(:runtime_constraints, resolve_runtime_constraints(attrs[:runtime_constraints]), md5: true)
log_reuse_info(candidates) { "after filtering on runtime_constraints #{attrs[:runtime_constraints].inspect}" }
log_reuse_info { "checking for state=Complete with readable output and log..." }
return usable
end
- # Check for Running candidates and return the most likely to finish sooner.
+ # Check for non-failing Running candidates and return the most likely to finish sooner.
log_reuse_info { "checking for state=Running..." }
running = candidates.where(state: Running).
+ where("(runtime_status->'error') is null").
order('progress desc, started_at asc').
limit(1).first
if running
# (because state might have changed while acquiring the lock).
check_lock_fail
transaction do
- begin
- reload(lock: 'FOR UPDATE NOWAIT')
- rescue
- raise LockFailedError.new("cannot lock: other transaction in progress")
- end
+ reload
check_lock_fail
update_attributes!(state: Locked)
end
current_user.andand.is_admin
end
- def permission_to_update
- # Override base permission check to allow auth_uuid to set progress and
- # output (only). Whether it is legal to set progress and output in the current
- # state has already been checked in validate_change.
- current_user.andand.is_admin ||
- (!current_api_client_authorization.nil? and
- [self.auth_uuid, self.locked_by_uuid].include? current_api_client_authorization.uuid)
- end
-
def ensure_owner_uuid_is_permitted
- # Override base permission check to allow auth_uuid to set progress and
- # output (only). Whether it is legal to set progress and output in the current
- # state has already been checked in validate_change.
- if !current_api_client_authorization.nil? and self.auth_uuid == current_api_client_authorization.uuid
- check_update_whitelist [:progress, :output]
- else
- super
- end
+ # validate_change ensures owner_uuid can't be changed at all --
+ # except during create, which requires admin privileges. Checking
+ # permission here would be superfluous.
+ true
end
def set_timestamps
end
end
+ # Check that well-known runtime status keys have desired data types
+ def validate_runtime_status
+ [
+ 'error', 'errorDetail', 'warning', 'warningDetail', 'activity'
+ ].each do |k|
+ if self.runtime_status.andand.include?(k) && !self.runtime_status[k].is_a?(String)
+ errors.add(:runtime_status, "'#{k}' value must be a string")
+ end
+ end
+ end
+
def validate_change
permitted = [:state]
+ progress_attrs = [:progress, :runtime_status, :log, :output]
+ final_attrs = [:exit_code, :finished_at]
if self.new_record?
permitted.push(:owner_uuid, :command, :container_image, :cwd,
end
case self.state
- when Queued, Locked
+ when Locked
+ permitted.push :priority, :runtime_status, :log
+
+ when Queued
permitted.push :priority
when Running
- permitted.push :priority, :progress, :output
+ permitted.push :priority, *progress_attrs
if self.state_changed?
permitted.push :started_at
end
when Complete
if self.state_was == Running
- permitted.push :finished_at, :output, :log, :exit_code
+ permitted.push *final_attrs, *progress_attrs
end
when Cancelled
case self.state_was
when Running
- permitted.push :finished_at, :output, :log
+ permitted.push :finished_at, *progress_attrs
when Queued, Locked
permitted.push :finished_at, :log
end
return false
end
+ if current_api_client_authorization.andand.uuid.andand == self.auth_uuid
+ # The contained process itself can update progress indicators,
+ # but can't change priority etc.
+ permitted = permitted & [:state, :progress, :output]
+ elsif self.locked_by_uuid && self.locked_by_uuid != current_api_client_authorization.andand.uuid
+ # When locked, progress fields cannot be updated by the wrong
+ # dispatcher, even though it has admin privileges.
+ permitted = permitted - progress_attrs
+ end
check_update_whitelist permitted
end
end
end
+ def clear_runtime_status_when_queued
+ # Avoid leaking status messages between different dispatch attempts
+ if self.state_was == Locked && self.state == Queued
+ self.runtime_status = {}
+ end
+ end
+
def handle_completed
# This container is finished so finalize any associated container requests
# that are associated with this container.
if self.state_changed? and self.final?
act_as_system_user do
- ActiveRecord::Base.connection.execute('LOCK container_requests, containers IN EXCLUSIVE MODE')
if self.state == Cancelled
retryable_requests = ContainerRequest.where("container_uuid = ? and priority > 0 and state = 'Committed' and container_count < container_count_max", uuid)
else
projects / arvados.git / blobdiff