1 require 'whitelist_update'
3 class Container < ArvadosModel
6 include CommonApiTemplate
7 include WhitelistUpdate
9 serialize :environment, Hash
10 serialize :mounts, Hash
11 serialize :runtime_constraints, Hash
12 serialize :command, Array
14 before_validation :fill_field_defaults, :if => :new_record?
15 before_validation :set_timestamps
16 validates :command, :container_image, :output_path, :cwd, :priority, :presence => true
17 validate :validate_state_change
18 validate :validate_change
19 validate :validate_lock
20 after_validation :assign_auth
21 before_save :sort_serialized_attrs
22 after_save :handle_completed
24 has_many :container_requests, :foreign_key => :container_uuid, :class_name => 'ContainerRequest', :primary_key => :uuid
25 belongs_to :auth, :class_name => 'ApiClientAuthorization', :foreign_key => :auth_uuid, :primary_key => :uuid
27 api_accessible :user, extend: :common do |t|
29 t.add :container_image
41 t.add :runtime_constraints
47 # Supported states for a container
52 (Running = 'Running'),
53 (Complete = 'Complete'),
54 (Cancelled = 'Cancelled')
59 Queued => [Locked, Cancelled],
60 Locked => [Queued, Running, Cancelled],
61 Running => [Complete, Cancelled]
69 if [Queued, Locked, Running].include? self.state
70 # Update the priority of this container to the maximum priority of any of
71 # its committed container requests and save the record.
72 self.priority = ContainerRequest.
73 where(container_uuid: uuid,
74 state: ContainerRequest::Committed).
80 def self.find_reusable(attrs)
81 candidates = Container.
82 where('command = ?', attrs[:command].to_yaml).
83 where('cwd = ?', attrs[:cwd]).
84 where('environment = ?', self.deep_sort_hash(attrs[:environment]).to_yaml).
85 where('output_path = ?', attrs[:output_path]).
86 where('container_image = ?', attrs[:container_image]).
87 where('mounts = ?', self.deep_sort_hash(attrs[:mounts]).to_yaml).
88 where('runtime_constraints = ?', self.deep_sort_hash(attrs[:runtime_constraints]).to_yaml)
90 # Check for Completed candidates that only had consistent outputs.
91 completed = candidates.where(state: Complete).where(exit_code: 0)
92 if completed.select("output").group('output').limit(2).length == 1
93 return completed.order('finished_at asc').limit(1).first
96 # Check for Running candidates and return the most likely to finish sooner.
97 running = candidates.where(state: Running).
98 order('progress desc, started_at asc').limit(1).first
99 return running if not running.nil?
101 # Check for Locked or Queued ones and return the most likely to start first.
102 locked_or_queued = candidates.where("state IN (?)", [Locked, Queued]).
103 order('state asc, priority desc, created_at asc').limit(1).first
104 return locked_or_queued if not locked_or_queued.nil?
106 # No suitable candidate found.
112 if self.state == Locked
113 raise AlreadyLockedError
122 if self.state == Queued
123 raise InvalidStateTransitionError
130 def self.readable_by(*users_list)
131 if users_list.select { |u| u.is_admin }.any?
134 user_uuids = users_list.map { |u| u.uuid }
135 uuid_list = user_uuids + users_list.flat_map { |u| u.groups_i_can(:read) }
137 permitted = "(SELECT head_uuid FROM links WHERE link_class='permission' AND tail_uuid IN (:uuids))"
138 joins(:container_requests).
139 where("container_requests.uuid IN #{permitted} OR "+
140 "container_requests.owner_uuid IN (:uuids)",
146 def fill_field_defaults
147 self.state ||= Queued
148 self.environment ||= {}
149 self.runtime_constraints ||= {}
155 def permission_to_create
156 current_user.andand.is_admin
159 def permission_to_update
160 current_user.andand.is_admin
164 if self.state_changed? and self.state == Running
165 self.started_at ||= db_current_time
168 if self.state_changed? and [Complete, Cancelled].include? self.state
169 self.finished_at ||= db_current_time
177 permitted.push(:owner_uuid, :command, :container_image, :cwd,
178 :environment, :mounts, :output_path, :priority,
179 :runtime_constraints)
184 permitted.push :priority
187 permitted.push :priority, :progress
188 if self.state_changed?
189 permitted.push :started_at
193 if self.state_was == Running
194 permitted.push :finished_at, :output, :log, :exit_code
200 permitted.push :finished_at, :output, :log
202 permitted.push :finished_at
206 # The state_transitions check will add an error message for this
210 check_update_whitelist permitted
214 # If the Container is already locked by someone other than the
215 # current api_client_auth, disallow all changes -- except
216 # priority, which needs to change to reflect max(priority) of
217 # relevant ContainerRequests.
218 if locked_by_uuid_was
219 if locked_by_uuid_was != Thread.current[:api_client_authorization].uuid
220 check_update_whitelist [:priority]
224 if [Locked, Running].include? self.state
225 # If the Container was already locked, locked_by_uuid must not
226 # changes. Otherwise, the current auth gets the lock.
227 need_lock = locked_by_uuid_was || Thread.current[:api_client_authorization].uuid
232 # The caller can provide a new value for locked_by_uuid, but only
233 # if it's exactly what we expect. This allows a caller to perform
234 # an update like {"state":"Unlocked","locked_by_uuid":null}.
235 if self.locked_by_uuid_changed?
236 if self.locked_by_uuid != need_lock
237 return errors.add :locked_by_uuid, "can only change to #{need_lock}"
240 self.locked_by_uuid = need_lock
244 if self.auth_uuid_changed?
245 return errors.add :auth_uuid, 'is readonly'
247 if not [Locked, Running].include? self.state
249 self.auth.andand.update_attributes(expires_at: db_current_time)
256 cr = ContainerRequest.
257 where('container_uuid=? and priority>0', self.uuid).
258 order('priority desc').
261 return errors.add :auth_uuid, "cannot be assigned because priority <= 0"
263 self.auth = ApiClientAuthorization.
264 create!(user_id: User.find_by_uuid(cr.modified_by_user_uuid).id,
268 def sort_serialized_attrs
269 if self.environment_changed?
270 self.environment = self.class.deep_sort_hash(self.environment)
272 if self.mounts_changed?
273 self.mounts = self.class.deep_sort_hash(self.mounts)
275 if self.runtime_constraints_changed?
276 self.runtime_constraints = self.class.deep_sort_hash(self.runtime_constraints)
281 # This container is finished so finalize any associated container requests
282 # that are associated with this container.
283 if self.state_changed? and [Complete, Cancelled].include? self.state
284 act_as_system_user do
286 if self.state == Cancelled
287 retryable_requests = ContainerRequest.where("priority > 0 and state = 'Committed' and container_count < container_count_max")
289 retryable_requests = []
292 if retryable_requests.any?
294 command: self.command,
296 environment: self.environment,
297 output_path: self.output_path,
298 container_image: self.container_image,
300 runtime_constraints: self.runtime_constraints
302 c = Container.create! c_attrs
303 retryable_requests.each do |cr|
305 # Use row locking because this increments container_count
306 cr.container_uuid = c.uuid
312 # Notify container requests associated with this container
313 ContainerRequest.where(container_uuid: uuid,
314 :state => ContainerRequest::Committed).each do |cr|
315 cr.container_completed!
318 # Try to cancel any outstanding container requests made by this container.
319 ContainerRequest.where(requesting_container_uuid: uuid,
320 :state => ContainerRequest::Committed).each do |cr|