services/api/app/models/container.rb

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