require 'safe_json'
class Container < ArvadosModel
+ include ArvadosModelUpdates
include HasUuid
include KindAndEtag
include CommonApiTemplate
serialize :runtime_constraints, Hash
serialize :command, Array
serialize :scheduling_parameters, Hash
+ serialize :secret_mounts, Hash
before_validation :fill_field_defaults, :if => :new_record?
before_validation :set_timestamps
- validates :command, :container_image, :output_path, :cwd, :priority, :presence => true
+ validates :command, :container_image, :output_path, :cwd, :priority, { presence: true }
+ validates :priority, numericality: { only_integer: true, greater_than_or_equal_to: 0 }
validate :validate_state_change
validate :validate_change
validate :validate_lock
validate :validate_output
after_validation :assign_auth
before_save :sort_serialized_attrs
+ before_save :update_secret_mounts_md5
+ before_save :scrub_secret_mounts
after_save :handle_completed
+ after_save :propagate_priority
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
["mounts"]
end
+ def self.full_text_searchable_columns
+ super - ["secret_mounts", "secret_mounts_md5"]
+ end
+
+ def self.searchable_columns *args
+ super - ["secret_mounts_md5"]
+ end
+
+ def logged_attributes
+ super.except('secret_mounts')
+ end
+
def state_transitions
State_transitions
end
+ # Container priority is the highest "computed priority" of any
+ # matching request. The computed priority of a container-submitted
+ # request is the priority of the submitting container. The computed
+ # priority of a user-submitted request is a function of
+ # user-assigned priority and request creation time.
def update_priority!
- if [Queued, Locked, Running].include? self.state
- # Update the priority of this container to the maximum priority of any of
- # its committed container requests and save the record.
- self.priority = ContainerRequest.
- where(container_uuid: uuid,
+ return if ![Queued, Locked, Running].include?(state)
+ p = ContainerRequest.
+ where('container_uuid=? and priority>0', uuid).
+ includes(:requesting_container).
+ lock(true).
+ map do |cr|
+ if cr.requesting_container
+ cr.requesting_container.priority
+ else
+ (cr.priority << 50) - (cr.created_at.to_time.to_f * 1000).to_i
+ end
+ end.max || 0
+ update_attributes!(priority: p)
+ end
+
+ def propagate_priority
+ return true unless priority_changed?
+ act_as_system_user do
+ # 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).
- maximum('priority')
- self.save!
+ where('container_uuid is not null').
+ includes(:container).
+ map(&:container).
+ map(&:update_priority!)
end
end
mounts: resolve_mounts(req.mounts),
runtime_constraints: resolve_runtime_constraints(req.runtime_constraints),
scheduling_parameters: req.scheduling_parameters,
+ secret_mounts: req.secret_mounts,
}
act_as_system_user do
if req.use_existing && (reusable = find_reusable(c_attrs))
candidates = candidates.where_serialized(:mounts, resolve_mounts(attrs[:mounts]))
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}" }
+
candidates = candidates.where_serialized(:runtime_constraints, resolve_runtime_constraints(attrs[:runtime_constraints]))
log_reuse_info(candidates) { "after filtering on runtime_constraints #{attrs[:runtime_constraints].inspect}" }
# (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
+ # Locking involves assigning auth_uuid, which involves looking
+ # up container requests, so we must lock both tables in the
+ # proper order to avoid deadlock.
+ ActiveRecord::Base.connection.execute('LOCK container_requests, containers IN EXCLUSIVE MODE')
+ reload
check_lock_fail
update_attributes!(state: Locked)
end
self.runtime_constraints ||= {}
self.mounts ||= {}
self.cwd ||= "."
- self.priority ||= 1
+ self.priority ||= 0
self.scheduling_parameters ||= {}
end
if self.new_record?
permitted.push(:owner_uuid, :command, :container_image, :cwd,
:environment, :mounts, :output_path, :priority,
- :runtime_constraints, :scheduling_parameters)
+ :runtime_constraints, :scheduling_parameters,
+ :secret_mounts)
end
case self.state
# that a container cannot "claim" a collection that it doesn't otherwise
# have access to just by setting the output field to the collection PDH.
if output_changed?
- c = Collection.unscoped do
- Collection.
- readable_by(current_user).
+ c = Collection.
+ readable_by(current_user, {include_trash: true}).
where(portable_data_hash: self.output).
first
- end
if !c
errors.add :output, "collection must exist and be readable by current user."
end
end
end
+ def update_secret_mounts_md5
+ if self.secret_mounts_changed?
+ self.secret_mounts_md5 = Digest::MD5.hexdigest(
+ SafeJSON.dump(self.class.deep_sort_hash(self.secret_mounts)))
+ end
+ end
+
+ def scrub_secret_mounts
+ # this runs after update_secret_mounts_md5, so the
+ # secret_mounts_md5 will still reflect the secrets that are being
+ # scrubbed here.
+ if self.state_changed? && self.final?
+ self.secret_mounts = {}
+ 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
c = Container.create! c_attrs
retryable_requests.each do |cr|
cr.with_lock do
- # Use row locking because this increments container_count
- cr.container_uuid = c.uuid
- cr.save
+ leave_modified_by_user_alone do
+ # Use row locking because this increments container_count
+ cr.container_uuid = c.uuid
+ cr.save!
+ end
end
end
end
# Notify container requests associated with this container
ContainerRequest.where(container_uuid: uuid,
state: ContainerRequest::Committed).each do |cr|
- cr.finalize!
+ leave_modified_by_user_alone do
+ cr.finalize!
+ end
end
- # Try to cancel any outstanding container requests made by this container.
- ContainerRequest.where(requesting_container_uuid: uuid,
- state: ContainerRequest::Committed).each do |cr|
- cr.priority = 0
- cr.save
+ # Cancel outstanding container requests made by this container.
+ ContainerRequest.
+ includes(:container).
+ where(requesting_container_uuid: uuid,
+ state: ContainerRequest::Committed).each do |cr|
+ leave_modified_by_user_alone do
+ cr.update_attributes!(priority: 0)
+ cr.container.reload
+ if cr.container.state == Container::Queued || cr.container.state == Container::Locked
+ # If the child container hasn't started yet, finalize the
+ # child CR now instead of leaving it "on hold", i.e.,
+ # Queued with priority 0. (OTOH, if the child is already
+ # running, leave it alone so it can get cancelled the
+ # usual way, get a copy of the log collection, etc.)
+ cr.update_attributes!(state: ContainerRequest::Final)
+ end
+ end
end
end
end
end
-
end