services/api/app/models/container_request.rb

   1 require 'whitelist_update'
   2
   3 class ContainerRequest < ArvadosModel
   4   include HasUuid
   5   include KindAndEtag
   6   include CommonApiTemplate
   7   include WhitelistUpdate
   8
   9   serialize :properties, Hash
  10   serialize :environment, Hash
  11   serialize :mounts, Hash
  12   serialize :runtime_constraints, Hash
  13   serialize :command, Array
  14   serialize :scheduling_parameters, Hash
  15
  16   before_validation :fill_field_defaults, :if => :new_record?
  17   before_validation :validate_runtime_constraints
  18   before_validation :validate_scheduling_parameters
  19   before_validation :set_container
  20   validates :command, :container_image, :output_path, :cwd, :presence => true
  21   validate :validate_state_change
  22   validate :validate_change
  23   after_save :update_priority
  24   after_save :finalize_if_needed
  25   before_create :set_requesting_container_uuid
  26
  27   api_accessible :user, extend: :common do |t|
  28     t.add :command
  29     t.add :container_count
  30     t.add :container_count_max
  31     t.add :container_image
  32     t.add :container_uuid
  33     t.add :cwd
  34     t.add :description
  35     t.add :environment
  36     t.add :expires_at
  37     t.add :filters
  38     t.add :log_uuid
  39     t.add :mounts
  40     t.add :name
  41     t.add :output_name
  42     t.add :output_path
  43     t.add :output_uuid
  44     t.add :priority
  45     t.add :properties
  46     t.add :requesting_container_uuid
  47     t.add :runtime_constraints
  48     t.add :scheduling_parameters
  49     t.add :state
  50     t.add :use_existing
  51   end
  52
  53   # Supported states for a container request
  54   States =
  55     [
  56      (Uncommitted = 'Uncommitted'),
  57      (Committed = 'Committed'),
  58      (Final = 'Final'),
  59     ]
  60
  61   State_transitions = {
  62     nil => [Uncommitted, Committed],
  63     Uncommitted => [Committed],
  64     Committed => [Final]
  65   }
  66
  67   def state_transitions
  68     State_transitions
  69   end
  70
  71   def skip_uuid_read_permission_check
  72     # XXX temporary until permissions are sorted out.
  73     %w(modified_by_client_uuid container_uuid requesting_container_uuid)
  74   end
  75
  76   def finalize_if_needed
  77     if state == Committed && Container.find_by_uuid(container_uuid).final?
  78       reload
  79       act_as_system_user do
  80         finalize!
  81       end
  82     end
  83   end
  84
  85   # Finalize the container request after the container has
  86   # finished/cancelled.
  87   def finalize!
  88     out_coll = nil
  89     log_coll = nil
  90     c = Container.find_by_uuid(container_uuid)
  91     ['output', 'log'].each do |out_type|
  92       pdh = c.send(out_type)
  93       next if pdh.nil?
  94       if self.output_name and out_type == 'output'
  95         coll_name = self.output_name
  96       else
  97         coll_name = "Container #{out_type} for request #{uuid}"
  98       end
  99       manifest = Collection.unscoped do
 100         Collection.where(portable_data_hash: pdh).first.manifest_text
 101       end
 102       begin
 103         coll = Collection.create!(owner_uuid: owner_uuid,
 104                                   manifest_text: manifest,
 105                                   portable_data_hash: pdh,
 106                                   name: coll_name,
 107                                   properties: {
 108                                     'type' => out_type,
 109                                     'container_request' => uuid,
 110                                   })
 111       rescue ActiveRecord::RecordNotUnique => rn
 112         # In case this is executed as part of a transaction: When a Postgres exception happens,
 113         # the following statements on the same transaction become invalid, so a rollback is
 114         # needed. One example are Unit Tests, every test is enclosed inside a transaction so
 115         # that the database can be reverted before every new test starts.
 116         # See: http://api.rubyonrails.org/classes/ActiveRecord/Transactions/ClassMethods.html#module-ActiveRecord::Transactions::ClassMethods-label-Exception+handling+and+rolling+back
 117         ActiveRecord::Base.connection.execute 'ROLLBACK'
 118         raise unless out_type == 'output' and self.output_name
 119         # Postgres specific unique name check. See ApplicationController#create for
 120         # a detailed explanation.
 121         raise unless rn.original_exception.is_a? PG::UniqueViolation
 122         err = rn.original_exception
 123         detail = err.result.error_field(PG::Result::PG_DIAG_MESSAGE_DETAIL)
 124         raise unless /^Key \(owner_uuid, name\)=\([a-z0-9]{5}-[a-z0-9]{5}-[a-z0-9]{15}, .*?\) already exists\./.match detail
 125         # Output collection name collision detected: append a timestamp.
 126         coll_name = "#{self.output_name} #{Time.now.getgm.strftime('%FT%TZ')}"
 127         retry
 128       end
 129       if out_type == 'output'
 130         out_coll = coll.uuid
 131       else
 132         log_coll = coll.uuid
 133       end
 134     end
 135     update_attributes!(state: Final, output_uuid: out_coll, log_uuid: log_coll)
 136   end
 137
 138   def self.full_text_searchable_columns
 139     super - ["mounts"]
 140   end
 141
 142   protected
 143
 144   def fill_field_defaults
 145     self.state ||= Uncommitted
 146     self.environment ||= {}
 147     self.runtime_constraints ||= {}
 148     self.mounts ||= {}
 149     self.cwd ||= "."
 150     self.container_count_max ||= Rails.configuration.container_count_max
 151     self.scheduling_parameters ||= {}
 152   end
 153
 154   # Create a new container (or find an existing one) to satisfy this
 155   # request.
 156   def resolve
 157     c_mounts = mounts_for_container
 158     c_runtime_constraints = {
 159       'keep_cache_ram' =>
 160       Rails.configuration.container_default_keep_cache_ram,
 161     }.merge(runtime_constraints_for_container)
 162     c_container_image = container_image_for_container
 163     c = act_as_system_user do
 164       c_attrs = {command: self.command,
 165                  cwd: self.cwd,
 166                  environment: self.environment,
 167                  output_path: self.output_path,
 168                  container_image: c_container_image,
 169                  mounts: c_mounts,
 170                  runtime_constraints: c_runtime_constraints}
 171
 172       reusable = self.use_existing ? Container.find_reusable(c_attrs) : nil
 173       if not reusable.nil?
 174         reusable
 175       else
 176         c_attrs[:scheduling_parameters] = self.scheduling_parameters
 177         Container.create!(c_attrs)
 178       end
 179     end
 180     self.container_uuid = c.uuid
 181   end
 182
 183   # Return a runtime_constraints hash that complies with
 184   # self.runtime_constraints but is suitable for saving in a container
 185   # record, i.e., has specific values instead of ranges.
 186   #
 187   # Doing this as a step separate from other resolutions, like "git
 188   # revision range to commit hash", makes sense only when there is no
 189   # opportunity to reuse an existing container (e.g., container reuse
 190   # is not implemented yet, or we have already found that no existing
 191   # containers are suitable).
 192   def runtime_constraints_for_container
 193     rc = {}
 194     runtime_constraints.each do |k, v|
 195       if v.is_a? Array
 196         rc[k] = v[0]
 197       else
 198         rc[k] = v
 199       end
 200     end
 201     rc
 202   end
 203
 204   # Return a mounts hash suitable for a Container, i.e., with every
 205   # readonly collection UUID resolved to a PDH.
 206   def mounts_for_container
 207     c_mounts = {}
 208     mounts.each do |k, mount|
 209       mount = mount.dup
 210       c_mounts[k] = mount
 211       if mount['kind'] != 'collection'
 212         next
 213       end
 214       if (uuid = mount.delete 'uuid')
 215         c = Collection.
 216           readable_by(current_user).
 217           where(uuid: uuid).
 218           select(:portable_data_hash).
 219           first
 220         if !c
 221           raise ArvadosModel::UnresolvableContainerError.new "cannot mount collection #{uuid.inspect}: not found"
 222         end
 223         if mount['portable_data_hash'].nil?
 224           # PDH not supplied by client
 225           mount['portable_data_hash'] = c.portable_data_hash
 226         elsif mount['portable_data_hash'] != c.portable_data_hash
 227           # UUID and PDH supplied by client, but they don't agree
 228           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"
 229         end
 230       end
 231     end
 232     return c_mounts
 233   end
 234
 235   # Return a container_image PDH suitable for a Container.
 236   def container_image_for_container
 237     coll = Collection.for_latest_docker_image(container_image)
 238     if !coll
 239       raise ArvadosModel::UnresolvableContainerError.new "docker image #{container_image.inspect} not found"
 240     end
 241     coll.portable_data_hash
 242   end
 243
 244   def set_container
 245     if (container_uuid_changed? and
 246         not current_user.andand.is_admin and
 247         not container_uuid.nil?)
 248       errors.add :container_uuid, "can only be updated to nil."
 249       return false
 250     end
 251     if state_changed? and state == Committed and container_uuid.nil?
 252       resolve
 253     end
 254     if self.container_uuid != self.container_uuid_was
 255       if self.container_count_changed?
 256         errors.add :container_count, "cannot be updated directly."
 257         return false
 258       else
 259         self.container_count += 1
 260       end
 261     end
 262   end
 263
 264   def validate_runtime_constraints
 265     case self.state
 266     when Committed
 267       [['vcpus', true],
 268        ['ram', true],
 269        ['keep_cache_ram', false]].each do |k, required|
 270         if !required && !runtime_constraints.include?(k)
 271           next
 272         end
 273         v = runtime_constraints[k]
 274         unless (v.is_a?(Integer) && v > 0)
 275           errors.add(:runtime_constraints,
 276                      "[#{k}]=#{v.inspect} must be a positive integer")
 277         end
 278       end
 279     end
 280   end
 281
 282   def validate_scheduling_parameters
 283     if self.state == Committed
 284       if scheduling_parameters.include? 'partitions' and
 285          (!scheduling_parameters['partitions'].is_a?(Array) ||
 286           scheduling_parameters['partitions'].reject{|x| !x.is_a?(String)}.size !=
 287             scheduling_parameters['partitions'].size)
 288             errors.add :scheduling_parameters, "partitions must be an array of strings"
 289       end
 290     end
 291   end
 292
 293   def validate_change
 294     permitted = [:owner_uuid]
 295
 296     case self.state
 297     when Uncommitted
 298       # Permit updating most fields
 299       permitted.push :command, :container_count_max,
 300                      :container_image, :cwd, :description, :environment,
 301                      :filters, :mounts, :name, :output_path, :priority,
 302                      :properties, :requesting_container_uuid, :runtime_constraints,
 303                      :state, :container_uuid, :use_existing, :scheduling_parameters,
 304                      :output_name
 305
 306     when Committed
 307       if container_uuid.nil?
 308         errors.add :container_uuid, "has not been resolved to a container."
 309       end
 310
 311       if priority.nil?
 312         errors.add :priority, "cannot be nil"
 313       end
 314
 315       # Can update priority, container count, name and description
 316       permitted.push :priority, :container_count, :container_count_max, :container_uuid,
 317                      :name, :description
 318
 319       if self.state_changed?
 320         # Allow create-and-commit in a single operation.
 321         permitted.push :command, :container_image, :cwd, :description, :environment,
 322                        :filters, :mounts, :name, :output_path, :properties,
 323                        :requesting_container_uuid, :runtime_constraints,
 324                        :state, :container_uuid, :use_existing, :scheduling_parameters,
 325                        :output_name
 326       end
 327
 328     when Final
 329       if not current_user.andand.is_admin and not (self.name_changed? || self.description_changed?)
 330         errors.add :state, "of container request can only be set to Final by system."
 331       end
 332
 333       if self.state_changed? || self.name_changed? || self.description_changed? || self.output_uuid_changed? || self.log_uuid_changed?
 334           permitted.push :state, :name, :description, :output_uuid, :log_uuid
 335       else
 336         errors.add :state, "does not allow updates"
 337       end
 338
 339     else
 340       errors.add :state, "invalid value"
 341     end
 342
 343     check_update_whitelist permitted
 344   end
 345
 346   def update_priority
 347     if self.state_changed? or
 348         self.priority_changed? or
 349         self.container_uuid_changed?
 350       act_as_system_user do
 351         Container.
 352           where('uuid in (?)',
 353                 [self.container_uuid_was, self.container_uuid].compact).
 354           map(&:update_priority!)
 355       end
 356     end
 357   end
 358
 359   def set_requesting_container_uuid
 360     return !new_record? if self.requesting_container_uuid   # already set
 361
 362     token_uuid = current_api_client_authorization.andand.uuid
 363     container = Container.where('auth_uuid=?', token_uuid).order('created_at desc').first
 364     self.requesting_container_uuid = container.uuid if container
 365     true
 366   end
 367 end