# already know how to properly treat them.
attribute :secret_mounts, :jsonbHash, default: {}
attribute :runtime_status, :jsonbHash, default: {}
- attribute :runtime_auth_scopes, :jsonbHash, default: {}
+ attribute :runtime_auth_scopes, :jsonbArray, default: []
+ attribute :output_storage_classes, :jsonbArray, default: lambda { Rails.configuration.DefaultStorageClasses }
+ attribute :output_properties, :jsonbHash, default: {}
serialize :environment, Hash
serialize :mounts, Hash
serialize :command, Array
serialize :scheduling_parameters, Hash
+ after_find :fill_container_defaults_after_find
before_validation :fill_field_defaults, :if => :new_record?
before_validation :set_timestamps
+ before_validation :check_lock
+ before_validation :check_unlock
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
t.add :scheduling_parameters
t.add :runtime_user_uuid
t.add :runtime_auth_scopes
+ t.add :lock_count
+ t.add :gateway_address
+ t.add :interactive_session_started
+ t.add :output_storage_classes
+ t.add :output_properties
end
# Supported states for a container
nil => [Queued],
Queued => [Locked, Cancelled],
Locked => [Queued, Running, Cancelled],
- Running => [Complete, Cancelled]
+ Running => [Complete, Cancelled],
+ Complete => [Cancelled]
}
def self.limit_index_columns_read
end
def self.full_text_searchable_columns
- super - ["secret_mounts", "secret_mounts_md5", "runtime_token"]
+ super - ["secret_mounts", "secret_mounts_md5", "runtime_token", "gateway_address", "output_storage_classes"]
end
def self.searchable_columns *args
- super - ["secret_mounts_md5", "runtime_token"]
+ super - ["secret_mounts_md5", "runtime_token", "gateway_address", "output_storage_classes"]
end
def logged_attributes
end
def propagate_priority
- return true unless priority_changed?
+ return true unless saved_change_to_priority?
act_as_system_user do
# Update the priority of child container requests to match new
# priority of the parent container (ignoring requests with no
secret_mounts: req.secret_mounts,
runtime_token: req.runtime_token,
runtime_user_uuid: runtime_user.uuid,
- runtime_auth_scopes: runtime_auth_scopes
+ runtime_auth_scopes: runtime_auth_scopes,
+ output_storage_classes: req.output_storage_classes,
}
end
act_as_system_user do
# containers are suitable).
def self.resolve_runtime_constraints(runtime_constraints)
rc = {}
- defaults = {
- 'keep_cache_ram' =>
- Rails.configuration.Containers.DefaultKeepCacheRAM,
- }
- defaults.merge(runtime_constraints).each do |k, v|
+ runtime_constraints.each do |k, v|
if v.is_a? Array
rc[k] = v[0]
else
rc[k] = v
end
end
+ if rc['keep_cache_ram'] == 0
+ rc['keep_cache_ram'] = Rails.configuration.Containers.DefaultKeepCacheRAM
+ end
rc
end
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]), md5: true)
+ if attrs[:runtime_constraints]['cuda'].nil?
+ attrs[:runtime_constraints]['cuda'] = {
+ 'device_count' => 0,
+ 'driver_version' => '',
+ 'hardware_capability' => '',
+ }
+ end
+ resolved_runtime_constraints = [resolve_runtime_constraints(attrs[:runtime_constraints])]
+ if resolved_runtime_constraints[0]['cuda']['device_count'] == 0
+ # If no CUDA requested, extend search to include older container
+ # records that don't have a 'cuda' section in runtime_constraints
+ resolved_runtime_constraints << resolved_runtime_constraints[0].except('cuda')
+ end
+
+ candidates = candidates.where_serialized(:runtime_constraints, resolved_runtime_constraints, md5: true, multivalue: 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..." }
nil
end
- def check_lock_fail
- if self.state != Queued
- raise LockFailedError.new("cannot lock when #{self.state}")
- elsif self.priority <= 0
- raise LockFailedError.new("cannot lock when priority<=0")
+ def lock
+ self.with_lock do
+ if self.state != Queued
+ raise LockFailedError.new("cannot lock when #{self.state}")
+ end
+ self.update_attributes!(state: Locked)
end
end
- def lock
- # Check invalid state transitions once before getting the lock
- # (because it's cheaper that way) and once after getting the lock
- # (because state might have changed while acquiring the lock).
- check_lock_fail
- transaction do
- reload
- check_lock_fail
- update_attributes!(state: Locked, lock_count: self.lock_count+1)
+ def check_lock
+ if state_was == Queued and state == Locked
+ if self.priority <= 0
+ raise LockFailedError.new("cannot lock when priority<=0")
+ end
+ self.lock_count = self.lock_count+1
end
end
- def check_unlock_fail
- if self.state != Locked
- raise InvalidStateTransitionError.new("cannot unlock when #{self.state}")
- elsif self.locked_by_uuid != current_api_client_authorization.uuid
- raise InvalidStateTransitionError.new("locked by a different token")
+ def unlock
+ self.with_lock do
+ if self.state != Locked
+ raise InvalidStateTransitionError.new("cannot unlock when #{self.state}")
+ end
+ self.update_attributes!(state: Queued)
end
end
- def unlock
- # Check invalid state transitions twice (see lock)
- check_unlock_fail
- transaction do
- reload(lock: 'FOR UPDATE')
- check_unlock_fail
- if self.lock_count < Rails.configuration.Containers.MaxDispatchAttempts
- update_attributes!(state: Queued)
- else
- update_attributes!(state: Cancelled,
- runtime_status: {
- error: "Container exceeded 'max_container_dispatch_attempts' (lock_count=#{self.lock_count}."
- })
+ def check_unlock
+ if state_was == Locked and state == Queued
+ if self.locked_by_uuid != current_api_client_authorization.uuid
+ raise ArvadosModel::PermissionDeniedError.new("locked by a different token")
+ end
+ if self.lock_count >= Rails.configuration.Containers.MaxDispatchAttempts
+ self.state = Cancelled
+ self.runtime_status = {error: "Failed to start container. Cancelled after exceeding 'Containers.MaxDispatchAttempts' (lock_count=#{self.lock_count})"}
end
end
end
if users_list.select { |u| u.is_admin }.any?
return super
end
- Container.where(ContainerRequest.readable_by(*users_list).where("containers.uuid = container_requests.container_uuid").exists)
+ Container.where(ContainerRequest.readable_by(*users_list).where("containers.uuid = container_requests.container_uuid").arel.exists)
end
def final?
current_user.andand.is_admin
end
+ def permission_to_destroy
+ current_user.andand.is_admin
+ end
+
def ensure_owner_uuid_is_permitted
# validate_change ensures owner_uuid can't be changed at all --
# except during create, which requires admin privileges. Checking
:environment, :mounts, :output_path, :priority,
:runtime_constraints, :scheduling_parameters,
:secret_mounts, :runtime_token,
- :runtime_user_uuid, :runtime_auth_scopes)
+ :runtime_user_uuid, :runtime_auth_scopes,
+ :output_storage_classes, :output_properties)
end
case self.state
permitted.push :priority
when Running
- permitted.push :priority, *progress_attrs
+ permitted.push :priority, :output_properties, *progress_attrs
if self.state_changed?
- permitted.push :started_at
+ permitted.push :started_at, :gateway_address
+ end
+ if !self.interactive_session_started_was
+ permitted.push :interactive_session_started
end
when Complete
return false
end
- if self.state == Running &&
+ if self.state_was == Running &&
!current_api_client_authorization.nil? &&
(current_api_client_authorization.uuid == self.auth_uuid ||
current_api_client_authorization.token == self.runtime_token)
# change priority or log.
permitted.push *final_attrs
permitted = permitted - [:log, :priority]
+ elsif !current_user.andand.is_admin
+ raise PermissionDeniedError
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.
# If self.final?, this update is superfluous: the final log/output
# update will be done when handle_completed calls finalize! on
# each requesting CR.
- return if self.final? || !self.log_changed?
+ return if self.final? || !saved_change_to_log?
leave_modified_by_user_alone do
ContainerRequest.where(container_uuid: self.uuid).each do |cr|
cr.update_collections(container: self, collections: ['log'])
return errors.add :auth_uuid, 'is readonly'
end
if not [Locked, Running].include? self.state
- # don't need one
- self.auth.andand.update_attributes(expires_at: db_current_time)
+ # Don't need one. If auth already exists, expire it.
+ #
+ # We use db_transaction_time here (not db_current_time) to
+ # ensure the token doesn't validate later in the same
+ # transaction (e.g., in a test case) by satisfying expires_at >
+ # transaction timestamp.
+ self.auth.andand.update_attributes(expires_at: db_transaction_time)
self.auth = nil
return
elsif self.auth
self.runtime_auth_scopes = ["all"]
end
- # generate a new token
+ # Generate a new token. This runs with admin credentials as it's done by a
+ # dispatcher user, so expires_at isn't enforced by API.MaxTokenLifetime.
self.auth = ApiClientAuthorization.
create!(user_id: User.find_by_uuid(self.runtime_user_uuid).id,
api_client_id: 0,
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
-
- if self.state == Cancelled
- retryable_requests = ContainerRequest.where("container_uuid = ? and priority > 0 and state = 'Committed' and container_count < container_count_max", uuid)
- else
- retryable_requests = []
- end
+ if saved_change_to_state? and self.final?
+ # These get wiped out by with_lock (which reloads the record),
+ # so record them now in case we need to schedule a retry.
+ prev_secret_mounts = secret_mounts_before_last_save
+ prev_runtime_token = runtime_token_before_last_save
+
+ # Need to take a lock on the container to ensure that any
+ # concurrent container requests that might try to reuse this
+ # container will block until the container completion
+ # transaction finishes. This ensure that concurrent container
+ # requests that try to reuse this container are finalized (on
+ # Complete) or don't reuse it (on Cancelled).
+ self.with_lock do
+ act_as_system_user do
+ if self.state == Cancelled
+ retryable_requests = ContainerRequest.where("container_uuid = ? and priority > 0 and state = 'Committed' and container_count < container_count_max", uuid)
+ else
+ retryable_requests = []
+ end
- if retryable_requests.any?
- c_attrs = {
- command: self.command,
- cwd: self.cwd,
- environment: self.environment,
- output_path: self.output_path,
- container_image: self.container_image,
- mounts: self.mounts,
- runtime_constraints: self.runtime_constraints,
- scheduling_parameters: self.scheduling_parameters,
- secret_mounts: self.secret_mounts_was,
- runtime_token: self.runtime_token_was,
- runtime_user_uuid: self.runtime_user_uuid,
- runtime_auth_scopes: self.runtime_auth_scopes
- }
- c = Container.create! c_attrs
- retryable_requests.each do |cr|
- cr.with_lock do
- leave_modified_by_user_alone do
- # Use row locking because this increments container_count
- cr.container_uuid = c.uuid
- cr.save!
+ if retryable_requests.any?
+ c_attrs = {
+ command: self.command,
+ cwd: self.cwd,
+ environment: self.environment,
+ output_path: self.output_path,
+ container_image: self.container_image,
+ mounts: self.mounts,
+ runtime_constraints: self.runtime_constraints,
+ scheduling_parameters: self.scheduling_parameters,
+ secret_mounts: prev_secret_mounts,
+ runtime_token: prev_runtime_token,
+ runtime_user_uuid: self.runtime_user_uuid,
+ runtime_auth_scopes: self.runtime_auth_scopes
+ }
+ c = Container.create! c_attrs
+ retryable_requests.each do |cr|
+ cr.with_lock do
+ 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
- end
- # Notify container requests associated with this container
- ContainerRequest.where(container_uuid: uuid,
- state: ContainerRequest::Committed).each do |cr|
- leave_modified_by_user_alone do
- cr.finalize!
+ # Notify container requests associated with this container
+ ContainerRequest.where(container_uuid: uuid,
+ state: ContainerRequest::Committed).each do |cr|
+ leave_modified_by_user_alone do
+ cr.finalize!
+ end
end
- end
- # 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)
+ # 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
projects / arvados.git / blobdiff