Merge branch 'master' of git.curoverse.com:arvados into 11876-r-sdk
[arvados.git] / services / api / app / models / container.rb
1 # Copyright (C) The Arvados Authors. All rights reserved.
2 #
3 # SPDX-License-Identifier: AGPL-3.0
4
5 require 'log_reuse_info'
6 require 'whitelist_update'
7 require 'safe_json'
8
9 class Container < ArvadosModel
10   include HasUuid
11   include KindAndEtag
12   include CommonApiTemplate
13   include WhitelistUpdate
14   extend CurrentApiClient
15   extend DbCurrentTime
16   extend LogReuseInfo
17
18   serialize :environment, Hash
19   serialize :mounts, Hash
20   serialize :runtime_constraints, Hash
21   serialize :command, Array
22   serialize :scheduling_parameters, Hash
23
24   before_validation :fill_field_defaults, :if => :new_record?
25   before_validation :set_timestamps
26   validates :command, :container_image, :output_path, :cwd, :priority, :presence => true
27   validate :validate_state_change
28   validate :validate_change
29   validate :validate_lock
30   validate :validate_output
31   after_validation :assign_auth
32   before_save :sort_serialized_attrs
33   after_save :handle_completed
34   after_save :propagate_priority
35
36   has_many :container_requests, :foreign_key => :container_uuid, :class_name => 'ContainerRequest', :primary_key => :uuid
37   belongs_to :auth, :class_name => 'ApiClientAuthorization', :foreign_key => :auth_uuid, :primary_key => :uuid
38
39   api_accessible :user, extend: :common do |t|
40     t.add :command
41     t.add :container_image
42     t.add :cwd
43     t.add :environment
44     t.add :exit_code
45     t.add :finished_at
46     t.add :locked_by_uuid
47     t.add :log
48     t.add :mounts
49     t.add :output
50     t.add :output_path
51     t.add :priority
52     t.add :progress
53     t.add :runtime_constraints
54     t.add :started_at
55     t.add :state
56     t.add :auth_uuid
57     t.add :scheduling_parameters
58   end
59
60   # Supported states for a container
61   States =
62     [
63      (Queued = 'Queued'),
64      (Locked = 'Locked'),
65      (Running = 'Running'),
66      (Complete = 'Complete'),
67      (Cancelled = 'Cancelled')
68     ]
69
70   State_transitions = {
71     nil => [Queued],
72     Queued => [Locked, Cancelled],
73     Locked => [Queued, Running, Cancelled],
74     Running => [Complete, Cancelled]
75   }
76
77   def self.limit_index_columns_read
78     ["mounts"]
79   end
80
81   def state_transitions
82     State_transitions
83   end
84
85   def update_priority!
86     if [Queued, Locked, Running].include? self.state
87       # Update the priority of this container to the maximum priority of any of
88       # its committed container requests and save the record.
89       self.priority = ContainerRequest.
90         where(container_uuid: uuid,
91               state: ContainerRequest::Committed).
92         maximum('priority')
93       self.save!
94     end
95   end
96
97   def propagate_priority
98     if self.priority_changed?
99       act_as_system_user do
100          # Update the priority of child container requests to match new priority
101          # of the parent container.
102          ContainerRequest.where(requesting_container_uuid: self.uuid,
103                                 state: ContainerRequest::Committed).each do |cr|
104            cr.priority = self.priority
105            cr.save
106          end
107        end
108     end
109   end
110
111   # Create a new container (or find an existing one) to satisfy the
112   # given container request.
113   def self.resolve(req)
114     c_attrs = {
115       command: req.command,
116       cwd: req.cwd,
117       environment: req.environment,
118       output_path: req.output_path,
119       container_image: resolve_container_image(req.container_image),
120       mounts: resolve_mounts(req.mounts),
121       runtime_constraints: resolve_runtime_constraints(req.runtime_constraints),
122       scheduling_parameters: req.scheduling_parameters,
123     }
124     act_as_system_user do
125       if req.use_existing && (reusable = find_reusable(c_attrs))
126         reusable
127       else
128         Container.create!(c_attrs)
129       end
130     end
131   end
132
133   # Return a runtime_constraints hash that complies with requested but
134   # is suitable for saving in a container record, i.e., has specific
135   # values instead of ranges.
136   #
137   # Doing this as a step separate from other resolutions, like "git
138   # revision range to commit hash", makes sense only when there is no
139   # opportunity to reuse an existing container (e.g., container reuse
140   # is not implemented yet, or we have already found that no existing
141   # containers are suitable).
142   def self.resolve_runtime_constraints(runtime_constraints)
143     rc = {}
144     defaults = {
145       'keep_cache_ram' =>
146       Rails.configuration.container_default_keep_cache_ram,
147     }
148     defaults.merge(runtime_constraints).each do |k, v|
149       if v.is_a? Array
150         rc[k] = v[0]
151       else
152         rc[k] = v
153       end
154     end
155     rc
156   end
157
158   # Return a mounts hash suitable for a Container, i.e., with every
159   # readonly collection UUID resolved to a PDH.
160   def self.resolve_mounts(mounts)
161     c_mounts = {}
162     mounts.each do |k, mount|
163       mount = mount.dup
164       c_mounts[k] = mount
165       if mount['kind'] != 'collection'
166         next
167       end
168       if (uuid = mount.delete 'uuid')
169         c = Collection.
170           readable_by(current_user).
171           where(uuid: uuid).
172           select(:portable_data_hash).
173           first
174         if !c
175           raise ArvadosModel::UnresolvableContainerError.new "cannot mount collection #{uuid.inspect}: not found"
176         end
177         if mount['portable_data_hash'].nil?
178           # PDH not supplied by client
179           mount['portable_data_hash'] = c.portable_data_hash
180         elsif mount['portable_data_hash'] != c.portable_data_hash
181           # UUID and PDH supplied by client, but they don't agree
182           raise ArgumentError.new "cannot mount collection #{uuid.inspect}: current portable_data_hash #{c.portable_data_hash.inspect} does not match #{c['portable_data_hash'].inspect} in request"
183         end
184       end
185     end
186     return c_mounts
187   end
188
189   # Return a container_image PDH suitable for a Container.
190   def self.resolve_container_image(container_image)
191     coll = Collection.for_latest_docker_image(container_image)
192     if !coll
193       raise ArvadosModel::UnresolvableContainerError.new "docker image #{container_image.inspect} not found"
194     end
195     coll.portable_data_hash
196   end
197
198   def self.find_reusable(attrs)
199     log_reuse_info { "starting with #{Container.all.count} container records in database" }
200     candidates = Container.where_serialized(:command, attrs[:command])
201     log_reuse_info(candidates) { "after filtering on command #{attrs[:command].inspect}" }
202
203     candidates = candidates.where('cwd = ?', attrs[:cwd])
204     log_reuse_info(candidates) { "after filtering on cwd #{attrs[:cwd].inspect}" }
205
206     candidates = candidates.where_serialized(:environment, attrs[:environment])
207     log_reuse_info(candidates) { "after filtering on environment #{attrs[:environment].inspect}" }
208
209     candidates = candidates.where('output_path = ?', attrs[:output_path])
210     log_reuse_info(candidates) { "after filtering on output_path #{attrs[:output_path].inspect}" }
211
212     image = resolve_container_image(attrs[:container_image])
213     candidates = candidates.where('container_image = ?', image)
214     log_reuse_info(candidates) { "after filtering on container_image #{image.inspect} (resolved from #{attrs[:container_image].inspect})" }
215
216     candidates = candidates.where_serialized(:mounts, resolve_mounts(attrs[:mounts]))
217     log_reuse_info(candidates) { "after filtering on mounts #{attrs[:mounts].inspect}" }
218
219     candidates = candidates.where_serialized(:runtime_constraints, resolve_runtime_constraints(attrs[:runtime_constraints]))
220     log_reuse_info(candidates) { "after filtering on runtime_constraints #{attrs[:runtime_constraints].inspect}" }
221
222     log_reuse_info { "checking for state=Complete with readable output and log..." }
223
224     select_readable_pdh = Collection.
225       readable_by(current_user).
226       select(:portable_data_hash).
227       to_sql
228
229     usable = candidates.where(state: Complete, exit_code: 0)
230     log_reuse_info(usable) { "with state=Complete, exit_code=0" }
231
232     usable = usable.where("log IN (#{select_readable_pdh})")
233     log_reuse_info(usable) { "with readable log" }
234
235     usable = usable.where("output IN (#{select_readable_pdh})")
236     log_reuse_info(usable) { "with readable output" }
237
238     usable = usable.order('finished_at ASC').limit(1).first
239     if usable
240       log_reuse_info { "done, reusing container #{usable.uuid} with state=Complete" }
241       return usable
242     end
243
244     # Check for Running candidates and return the most likely to finish sooner.
245     log_reuse_info { "checking for state=Running..." }
246     running = candidates.where(state: Running).
247               order('progress desc, started_at asc').
248               limit(1).first
249     if running
250       log_reuse_info { "done, reusing container #{running.uuid} with state=Running" }
251       return running
252     else
253       log_reuse_info { "have no containers in Running state" }
254     end
255
256     # Check for Locked or Queued ones and return the most likely to start first.
257     locked_or_queued = candidates.
258                        where("state IN (?)", [Locked, Queued]).
259                        order('state asc, priority desc, created_at asc').
260                        limit(1).first
261     if locked_or_queued
262       log_reuse_info { "done, reusing container #{locked_or_queued.uuid} with state=#{locked_or_queued.state}" }
263       return locked_or_queued
264     else
265       log_reuse_info { "have no containers in Locked or Queued state" }
266     end
267
268     log_reuse_info { "done, no reusable container found" }
269     nil
270   end
271
272   def check_lock_fail
273     if self.state != Queued
274       raise LockFailedError.new("cannot lock when #{self.state}")
275     elsif self.priority <= 0
276       raise LockFailedError.new("cannot lock when priority<=0")
277     end
278   end
279
280   def lock
281     # Check invalid state transitions once before getting the lock
282     # (because it's cheaper that way) and once after getting the lock
283     # (because state might have changed while acquiring the lock).
284     check_lock_fail
285     transaction do
286       begin
287         reload(lock: 'FOR UPDATE NOWAIT')
288       rescue
289         raise LockFailedError.new("cannot lock: other transaction in progress")
290       end
291       check_lock_fail
292       update_attributes!(state: Locked)
293     end
294   end
295
296   def check_unlock_fail
297     if self.state != Locked
298       raise InvalidStateTransitionError.new("cannot unlock when #{self.state}")
299     elsif self.locked_by_uuid != current_api_client_authorization.uuid
300       raise InvalidStateTransitionError.new("locked by a different token")
301     end
302   end
303
304   def unlock
305     # Check invalid state transitions twice (see lock)
306     check_unlock_fail
307     transaction do
308       reload(lock: 'FOR UPDATE')
309       check_unlock_fail
310       update_attributes!(state: Queued)
311     end
312   end
313
314   def self.readable_by(*users_list)
315     # Load optional keyword arguments, if they exist.
316     if users_list.last.is_a? Hash
317       kwargs = users_list.pop
318     else
319       kwargs = {}
320     end
321     Container.where(ContainerRequest.readable_by(*users_list).where("containers.uuid = container_requests.container_uuid").exists)
322   end
323
324   def final?
325     [Complete, Cancelled].include?(self.state)
326   end
327
328   protected
329
330   def fill_field_defaults
331     self.state ||= Queued
332     self.environment ||= {}
333     self.runtime_constraints ||= {}
334     self.mounts ||= {}
335     self.cwd ||= "."
336     self.priority ||= 1
337     self.scheduling_parameters ||= {}
338   end
339
340   def permission_to_create
341     current_user.andand.is_admin
342   end
343
344   def permission_to_update
345     # Override base permission check to allow auth_uuid to set progress and
346     # output (only).  Whether it is legal to set progress and output in the current
347     # state has already been checked in validate_change.
348     current_user.andand.is_admin ||
349       (!current_api_client_authorization.nil? and
350        [self.auth_uuid, self.locked_by_uuid].include? current_api_client_authorization.uuid)
351   end
352
353   def ensure_owner_uuid_is_permitted
354     # Override base permission check to allow auth_uuid to set progress and
355     # output (only).  Whether it is legal to set progress and output in the current
356     # state has already been checked in validate_change.
357     if !current_api_client_authorization.nil? and self.auth_uuid == current_api_client_authorization.uuid
358       check_update_whitelist [:progress, :output]
359     else
360       super
361     end
362   end
363
364   def set_timestamps
365     if self.state_changed? and self.state == Running
366       self.started_at ||= db_current_time
367     end
368
369     if self.state_changed? and [Complete, Cancelled].include? self.state
370       self.finished_at ||= db_current_time
371     end
372   end
373
374   def validate_change
375     permitted = [:state]
376
377     if self.new_record?
378       permitted.push(:owner_uuid, :command, :container_image, :cwd,
379                      :environment, :mounts, :output_path, :priority,
380                      :runtime_constraints, :scheduling_parameters)
381     end
382
383     case self.state
384     when Queued, Locked
385       permitted.push :priority
386
387     when Running
388       permitted.push :priority, :progress, :output
389       if self.state_changed?
390         permitted.push :started_at
391       end
392
393     when Complete
394       if self.state_was == Running
395         permitted.push :finished_at, :output, :log, :exit_code
396       end
397
398     when Cancelled
399       case self.state_was
400       when Running
401         permitted.push :finished_at, :output, :log
402       when Queued, Locked
403         permitted.push :finished_at, :log
404       end
405
406     else
407       # The state_transitions check will add an error message for this
408       return false
409     end
410
411     check_update_whitelist permitted
412   end
413
414   def validate_lock
415     if [Locked, Running].include? self.state
416       # If the Container was already locked, locked_by_uuid must not
417       # changes. Otherwise, the current auth gets the lock.
418       need_lock = locked_by_uuid_was || current_api_client_authorization.andand.uuid
419     else
420       need_lock = nil
421     end
422
423     # The caller can provide a new value for locked_by_uuid, but only
424     # if it's exactly what we expect. This allows a caller to perform
425     # an update like {"state":"Unlocked","locked_by_uuid":null}.
426     if self.locked_by_uuid_changed?
427       if self.locked_by_uuid != need_lock
428         return errors.add :locked_by_uuid, "can only change to #{need_lock}"
429       end
430     end
431     self.locked_by_uuid = need_lock
432   end
433
434   def validate_output
435     # Output must exist and be readable by the current user.  This is so
436     # that a container cannot "claim" a collection that it doesn't otherwise
437     # have access to just by setting the output field to the collection PDH.
438     if output_changed?
439       c = Collection.unscoped do
440         Collection.
441             readable_by(current_user).
442             where(portable_data_hash: self.output).
443             first
444       end
445       if !c
446         errors.add :output, "collection must exist and be readable by current user."
447       end
448     end
449   end
450
451   def assign_auth
452     if self.auth_uuid_changed?
453       return errors.add :auth_uuid, 'is readonly'
454     end
455     if not [Locked, Running].include? self.state
456       # don't need one
457       self.auth.andand.update_attributes(expires_at: db_current_time)
458       self.auth = nil
459       return
460     elsif self.auth
461       # already have one
462       return
463     end
464     cr = ContainerRequest.
465       where('container_uuid=? and priority>0', self.uuid).
466       order('priority desc').
467       first
468     if !cr
469       return errors.add :auth_uuid, "cannot be assigned because priority <= 0"
470     end
471     self.auth = ApiClientAuthorization.
472       create!(user_id: User.find_by_uuid(cr.modified_by_user_uuid).id,
473               api_client_id: 0)
474   end
475
476   def sort_serialized_attrs
477     if self.environment_changed?
478       self.environment = self.class.deep_sort_hash(self.environment)
479     end
480     if self.mounts_changed?
481       self.mounts = self.class.deep_sort_hash(self.mounts)
482     end
483     if self.runtime_constraints_changed?
484       self.runtime_constraints = self.class.deep_sort_hash(self.runtime_constraints)
485     end
486     if self.scheduling_parameters_changed?
487       self.scheduling_parameters = self.class.deep_sort_hash(self.scheduling_parameters)
488     end
489   end
490
491   def handle_completed
492     # This container is finished so finalize any associated container requests
493     # that are associated with this container.
494     if self.state_changed? and self.final?
495       act_as_system_user do
496
497         if self.state == Cancelled
498           retryable_requests = ContainerRequest.where("container_uuid = ? and priority > 0 and state = 'Committed' and container_count < container_count_max", uuid)
499         else
500           retryable_requests = []
501         end
502
503         if retryable_requests.any?
504           c_attrs = {
505             command: self.command,
506             cwd: self.cwd,
507             environment: self.environment,
508             output_path: self.output_path,
509             container_image: self.container_image,
510             mounts: self.mounts,
511             runtime_constraints: self.runtime_constraints,
512             scheduling_parameters: self.scheduling_parameters
513           }
514           c = Container.create! c_attrs
515           retryable_requests.each do |cr|
516             cr.with_lock do
517               # Use row locking because this increments container_count
518               cr.container_uuid = c.uuid
519               cr.save
520             end
521           end
522         end
523
524         # Notify container requests associated with this container
525         ContainerRequest.where(container_uuid: uuid,
526                                state: ContainerRequest::Committed).each do |cr|
527           cr.finalize!
528         end
529
530         # Try to cancel any outstanding container requests made by this container.
531         ContainerRequest.where(requesting_container_uuid: uuid,
532                                state: ContainerRequest::Committed).each do |cr|
533           cr.priority = 0
534           cr.save
535         end
536       end
537     end
538   end
539
540 end