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 = runtime_constraints_for_container
 159     c_container_image = container_image_for_container
 160     c = act_as_system_user do
 161       c_attrs = {command: self.command,
 162                  cwd: self.cwd,
 163                  environment: self.environment,
 164                  output_path: self.output_path,
 165                  container_image: c_container_image,
 166                  mounts: c_mounts,
 167                  runtime_constraints: c_runtime_constraints}
 168
 169       reusable = self.use_existing ? Container.find_reusable(c_attrs) : nil
 170       if not reusable.nil?
 171         reusable
 172       else
 173         c_attrs[:scheduling_parameters] = self.scheduling_parameters
 174         Container.create!(c_attrs)
 175       end
 176     end
 177     self.container_uuid = c.uuid
 178   end
 179
 180   # Return a runtime_constraints hash that complies with
 181   # self.runtime_constraints but is suitable for saving in a container
 182   # record, i.e., has specific values instead of ranges.
 183   #
 184   # Doing this as a step separate from other resolutions, like "git
 185   # revision range to commit hash", makes sense only when there is no
 186   # opportunity to reuse an existing container (e.g., container reuse
 187   # is not implemented yet, or we have already found that no existing
 188   # containers are suitable).
 189   def runtime_constraints_for_container
 190     rc = {}
 191     runtime_constraints.each do |k, v|
 192       if v.is_a? Array
 193         rc[k] = v[0]
 194       else
 195         rc[k] = v
 196       end
 197     end
 198     rc
 199   end
 200
 201   # Return a mounts hash suitable for a Container, i.e., with every
 202   # readonly collection UUID resolved to a PDH.
 203   def mounts_for_container
 204     c_mounts = {}
 205     mounts.each do |k, mount|
 206       mount = mount.dup
 207       c_mounts[k] = mount
 208       if mount['kind'] != 'collection'
 209         next
 210       end
 211       if (uuid = mount.delete 'uuid')
 212         c = Collection.
 213           readable_by(current_user).
 214           where(uuid: uuid).
 215           select(:portable_data_hash).
 216           first
 217         if !c
 218           raise ArvadosModel::UnresolvableContainerError.new "cannot mount collection #{uuid.inspect}: not found"
 219         end
 220         if mount['portable_data_hash'].nil?
 221           # PDH not supplied by client
 222           mount['portable_data_hash'] = c.portable_data_hash
 223         elsif mount['portable_data_hash'] != c.portable_data_hash
 224           # UUID and PDH supplied by client, but they don't agree
 225           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"
 226         end
 227       end
 228     end
 229     return c_mounts
 230   end
 231
 232   # Return a container_image PDH suitable for a Container.
 233   def container_image_for_container
 234     coll = Collection.for_latest_docker_image(container_image)
 235     if !coll
 236       raise ArvadosModel::UnresolvableContainerError.new "docker image #{container_image.inspect} not found"
 237     end
 238     return Collection.docker_migration_pdh([current_user], coll.portable_data_hash)
 239   end
 240
 241   def set_container
 242     if (container_uuid_changed? and
 243         not current_user.andand.is_admin and
 244         not container_uuid.nil?)
 245       errors.add :container_uuid, "can only be updated to nil."
 246       return false
 247     end
 248     if state_changed? and state == Committed and container_uuid.nil?
 249       resolve
 250     end
 251     if self.container_uuid != self.container_uuid_was
 252       if self.container_count_changed?
 253         errors.add :container_count, "cannot be updated directly."
 254         return false
 255       else
 256         self.container_count += 1
 257       end
 258     end
 259   end
 260
 261   def validate_runtime_constraints
 262     case self.state
 263     when Committed
 264       ['vcpus', 'ram'].each do |k|
 265         if not (runtime_constraints.include? k and
 266                 runtime_constraints[k].is_a? Integer and
 267                 runtime_constraints[k] > 0)
 268           errors.add :runtime_constraints, "#{k} must be a positive integer"
 269         end
 270       end
 271
 272       if runtime_constraints.include? 'keep_cache_ram' and
 273          (!runtime_constraints['keep_cache_ram'].is_a?(Integer) or
 274           runtime_constraints['keep_cache_ram'] <= 0)
 275             errors.add :runtime_constraints, "keep_cache_ram must be a positive integer"
 276       elsif !runtime_constraints.include? 'keep_cache_ram'
 277         runtime_constraints['keep_cache_ram'] = Rails.configuration.container_default_keep_cache_ram
 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