services/api/app/models/container_request.rb

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 require 'whitelist_update'
   6 require 'arvados/collection'
   7
   8 class ContainerRequest < ArvadosModel
   9   include ArvadosModelUpdates
  10   include HasUuid
  11   include KindAndEtag
  12   include CommonApiTemplate
  13   include WhitelistUpdate
  14
  15   belongs_to :container, foreign_key: :container_uuid, primary_key: :uuid
  16   belongs_to :requesting_container, {
  17                class_name: 'Container',
  18                foreign_key: :requesting_container_uuid,
  19                primary_key: :uuid,
  20              }
  21
  22   # Posgresql JSONB columns should NOT be declared as serialized, Rails 5
  23   # already know how to properly treat them.
  24   attribute :properties, :jsonbHash, default: {}
  25   attribute :secret_mounts, :jsonbHash, default: {}
  26
  27   serialize :environment, Hash
  28   serialize :mounts, Hash
  29   serialize :runtime_constraints, Hash
  30   serialize :command, Array
  31   serialize :scheduling_parameters, Hash
  32
  33   after_find :fill_container_defaults_after_find
  34   before_validation :fill_field_defaults, :if => :new_record?
  35   before_validation :fill_container_defaults
  36   before_validation :set_default_preemptible_scheduling_parameter
  37   before_validation :set_container
  38   validates :command, :container_image, :output_path, :cwd, :presence => true
  39   validates :output_ttl, numericality: { only_integer: true, greater_than_or_equal_to: 0 }
  40   validates :priority, numericality: { only_integer: true, greater_than_or_equal_to: 0, less_than_or_equal_to: 1000 }
  41   validate :validate_datatypes
  42   validate :validate_runtime_constraints
  43   validate :validate_scheduling_parameters
  44   validate :validate_state_change
  45   validate :check_update_whitelist
  46   validate :secret_mounts_key_conflict
  47   validate :validate_runtime_token
  48   before_save :scrub_secrets
  49   before_create :set_requesting_container_uuid
  50   before_destroy :set_priority_zero
  51   after_save :update_priority
  52   after_save :finalize_if_needed
  53
  54   api_accessible :user, extend: :common do |t|
  55     t.add :command
  56     t.add :container_count
  57     t.add :container_count_max
  58     t.add :container_image
  59     t.add :container_uuid
  60     t.add :cwd
  61     t.add :description
  62     t.add :environment
  63     t.add :expires_at
  64     t.add :filters
  65     t.add :log_uuid
  66     t.add :mounts
  67     t.add :name
  68     t.add :output_name
  69     t.add :output_path
  70     t.add :output_uuid
  71     t.add :output_ttl
  72     t.add :priority
  73     t.add :properties
  74     t.add :requesting_container_uuid
  75     t.add :runtime_constraints
  76     t.add :scheduling_parameters
  77     t.add :state
  78     t.add :use_existing
  79   end
  80
  81   # Supported states for a container request
  82   States =
  83     [
  84      (Uncommitted = 'Uncommitted'),
  85      (Committed = 'Committed'),
  86      (Final = 'Final'),
  87     ]
  88
  89   State_transitions = {
  90     nil => [Uncommitted, Committed],
  91     Uncommitted => [Committed],
  92     Committed => [Final]
  93   }
  94
  95   AttrsPermittedAlways = [:owner_uuid, :state, :name, :description, :properties]
  96   AttrsPermittedBeforeCommit = [:command, :container_count_max,
  97   :container_image, :cwd, :environment, :filters, :mounts,
  98   :output_path, :priority, :runtime_token,
  99   :runtime_constraints, :state, :container_uuid, :use_existing,
 100   :scheduling_parameters, :secret_mounts, :output_name, :output_ttl]
 101
 102   def self.limit_index_columns_read
 103     ["mounts"]
 104   end
 105
 106   def logged_attributes
 107     super.except('secret_mounts', 'runtime_token')
 108   end
 109
 110   def state_transitions
 111     State_transitions
 112   end
 113
 114   def skip_uuid_read_permission_check
 115     # The uuid_read_permission_check prevents users from making
 116     # references to objects they can't view.  However, in this case we
 117     # don't want to do that check since there's a circular dependency
 118     # where user can't view the container until the user has
 119     # constructed the container request that references the container.
 120     %w(container_uuid)
 121   end
 122
 123   def finalize_if_needed
 124     return if state != Committed
 125     while true
 126       # get container lock first, then lock current container request
 127       # (same order as Container#handle_completed). Locking always
 128       # reloads the Container and ContainerRequest records.
 129       c = Container.find_by_uuid(container_uuid)
 130       c.lock! if !c.nil?
 131       self.lock!
 132
 133       if !c.nil? && container_uuid != c.uuid
 134         # After locking, we've noticed a race, the container_uuid is
 135         # different than the container record we just loaded.  This
 136         # can happen if Container#handle_completed scheduled a new
 137         # container for retry and set container_uuid while we were
 138         # waiting on the container lock.  Restart the loop and get the
 139         # new container.
 140         redo
 141       end
 142
 143       if !c.nil?
 144         if state == Committed && c.final?
 145           # The current container is
 146           act_as_system_user do
 147             leave_modified_by_user_alone do
 148               finalize!
 149             end
 150           end
 151         end
 152       elsif state == Committed
 153         # Behave as if the container is cancelled
 154         update_attributes!(state: Final)
 155       end
 156       return true
 157     end
 158   end
 159
 160   # Finalize the container request after the container has
 161   # finished/cancelled.
 162   def finalize!
 163     container = Container.find_by_uuid(container_uuid)
 164     if !container.nil?
 165       update_collections(container: container)
 166
 167       if container.state == Container::Complete
 168         log_col = Collection.where(portable_data_hash: container.log).first
 169         if log_col
 170           # Need to save collection
 171           completed_coll = Collection.new(
 172             owner_uuid: self.owner_uuid,
 173             name: "Container log for container #{container_uuid}",
 174             properties: {
 175               'type' => 'log',
 176               'container_request' => self.uuid,
 177               'container_uuid' => container_uuid,
 178             },
 179             portable_data_hash: log_col.portable_data_hash,
 180             manifest_text: log_col.manifest_text)
 181           completed_coll.save_with_unique_name!
 182         end
 183       end
 184     end
 185     update_attributes!(state: Final)
 186   end
 187
 188   def update_collections(container:, collections: ['log', 'output'])
 189     collections.each do |out_type|
 190       pdh = container.send(out_type)
 191       next if pdh.nil?
 192       c = Collection.where(portable_data_hash: pdh).first
 193       next if c.nil?
 194       manifest = c.manifest_text
 195
 196       coll_name = "Container #{out_type} for request #{uuid}"
 197       trash_at = nil
 198       if out_type == 'output'
 199         if self.output_name and self.output_name != ""
 200           coll_name = self.output_name
 201         end
 202         if self.output_ttl > 0
 203           trash_at = db_current_time + self.output_ttl
 204         end
 205       end
 206
 207       coll_uuid = self.send(out_type + '_uuid')
 208       coll = coll_uuid.nil? ? nil : Collection.where(uuid: coll_uuid).first
 209       if !coll
 210         coll = Collection.new(
 211           owner_uuid: self.owner_uuid,
 212           name: coll_name,
 213           manifest_text: "",
 214           properties: {
 215             'type' => out_type,
 216             'container_request' => uuid,
 217           })
 218       end
 219
 220       if out_type == "log"
 221         # Copy the log into a merged collection
 222         src = Arv::Collection.new(manifest)
 223         dst = Arv::Collection.new(coll.manifest_text)
 224         dst.cp_r("./", ".", src)
 225         dst.cp_r("./", "log for container #{container.uuid}", src)
 226         manifest = dst.manifest_text
 227       end
 228
 229       coll.assign_attributes(
 230         portable_data_hash: Digest::MD5.hexdigest(manifest) + '+' + manifest.bytesize.to_s,
 231         manifest_text: manifest,
 232         trash_at: trash_at,
 233         delete_at: trash_at)
 234       coll.save_with_unique_name!
 235       self.send(out_type + '_uuid=', coll.uuid)
 236     end
 237   end
 238
 239   def self.full_text_searchable_columns
 240     super - ["mounts", "secret_mounts", "secret_mounts_md5", "runtime_token"]
 241   end
 242
 243   protected
 244
 245   def fill_field_defaults
 246     self.state ||= Uncommitted
 247     self.environment ||= {}
 248     self.runtime_constraints ||= {}
 249     self.mounts ||= {}
 250     self.secret_mounts ||= {}
 251     self.cwd ||= "."
 252     self.container_count_max ||= Rails.configuration.Containers.MaxRetryAttempts
 253     self.scheduling_parameters ||= {}
 254     self.output_ttl ||= 0
 255     self.priority ||= 0
 256   end
 257
 258   def set_container
 259     if (container_uuid_changed? and
 260         not current_user.andand.is_admin and
 261         not container_uuid.nil?)
 262       errors.add :container_uuid, "can only be updated to nil."
 263       return false
 264     end
 265     if state_changed? and state == Committed and container_uuid.nil?
 266       while true
 267         c = Container.resolve(self)
 268         c.lock!
 269         if c.state == Container::Cancelled
 270           # Lost a race, we have a lock on the container but the
 271           # container was cancelled in a different request, restart
 272           # the loop and resolve request to a new container.
 273           redo
 274         end
 275         self.container_uuid = c.uuid
 276         break
 277       end
 278     end
 279     if self.container_uuid != self.container_uuid_was
 280       if self.container_count_changed?
 281         errors.add :container_count, "cannot be updated directly."
 282         return false
 283       end
 284
 285       self.container_count += 1
 286       return if self.container_uuid_was.nil?
 287
 288       old_container = Container.find_by_uuid(self.container_uuid_was)
 289       return if old_container.nil?
 290
 291       old_logs = Collection.where(portable_data_hash: old_container.log).first
 292       return if old_logs.nil?
 293
 294       log_coll = self.log_uuid.nil? ? nil : Collection.where(uuid: self.log_uuid).first
 295       if self.log_uuid.nil?
 296         log_coll = Collection.new(
 297           owner_uuid: self.owner_uuid,
 298           name: coll_name = "Container log for request #{uuid}",
 299           manifest_text: "")
 300       end
 301
 302       # copy logs from old container into CR's log collection
 303       src = Arv::Collection.new(old_logs.manifest_text)
 304       dst = Arv::Collection.new(log_coll.manifest_text)
 305       dst.cp_r("./", "log for container #{old_container.uuid}", src)
 306       manifest = dst.manifest_text
 307
 308       log_coll.assign_attributes(
 309         portable_data_hash: Digest::MD5.hexdigest(manifest) + '+' + manifest.bytesize.to_s,
 310         manifest_text: manifest)
 311       log_coll.save_with_unique_name!
 312       self.log_uuid = log_coll.uuid
 313     end
 314   end
 315
 316   def set_default_preemptible_scheduling_parameter
 317     if Rails.configuration.Containers.UsePreemptibleInstances && state == Committed && get_requesting_container()
 318       self.scheduling_parameters['preemptible'] = true
 319     end
 320   end
 321
 322   def validate_runtime_constraints
 323     case self.state
 324     when Committed
 325       ['vcpus', 'ram'].each do |k|
 326         v = runtime_constraints[k]
 327         if !v.is_a?(Integer) || v <= 0
 328           errors.add(:runtime_constraints,
 329                      "[#{k}]=#{v.inspect} must be a positive integer")
 330         end
 331       end
 332     end
 333   end
 334
 335   def validate_datatypes
 336     command.each do |c|
 337       if !c.is_a? String
 338         errors.add(:command, "must be an array of strings but has entry #{c.class}")
 339       end
 340     end
 341     environment.each do |k,v|
 342       if !k.is_a?(String) || !v.is_a?(String)
 343         errors.add(:environment, "must be an map of String to String but has entry #{k.class} to #{v.class}")
 344       end
 345     end
 346     [:mounts, :secret_mounts].each do |m|
 347       self[m].each do |k, v|
 348         if !k.is_a?(String) || !v.is_a?(Hash)
 349           errors.add(m, "must be an map of String to Hash but is has entry #{k.class} to #{v.class}")
 350         end
 351         if v["kind"].nil?
 352           errors.add(m, "each item must have a 'kind' field")
 353         end
 354         [[String, ["kind", "portable_data_hash", "uuid", "device_type",
 355                    "path", "commit", "repository_name", "git_url"]],
 356          [Integer, ["capacity"]]].each do |t, fields|
 357           fields.each do |f|
 358             if !v[f].nil? && !v[f].is_a?(t)
 359               errors.add(m, "#{k}: #{f} must be a #{t} but is #{v[f].class}")
 360             end
 361           end
 362         end
 363         ["writable", "exclude_from_output"].each do |f|
 364           if !v[f].nil? && !v[f].is_a?(TrueClass) && !v[f].is_a?(FalseClass)
 365             errors.add(m, "#{k}: #{f} must be a #{t} but is #{v[f].class}")
 366           end
 367         end
 368       end
 369     end
 370   end
 371
 372   def validate_scheduling_parameters
 373     if self.state == Committed
 374       if scheduling_parameters.include? 'partitions' and
 375          (!scheduling_parameters['partitions'].is_a?(Array) ||
 376           scheduling_parameters['partitions'].reject{|x| !x.is_a?(String)}.size !=
 377             scheduling_parameters['partitions'].size)
 378             errors.add :scheduling_parameters, "partitions must be an array of strings"
 379       end
 380       if !Rails.configuration.Containers.UsePreemptibleInstances and scheduling_parameters['preemptible']
 381         errors.add :scheduling_parameters, "preemptible instances are not allowed"
 382       end
 383       if scheduling_parameters.include? 'max_run_time' and
 384         (!scheduling_parameters['max_run_time'].is_a?(Integer) ||
 385           scheduling_parameters['max_run_time'] < 0)
 386           errors.add :scheduling_parameters, "max_run_time must be positive integer"
 387       end
 388     end
 389   end
 390
 391   def check_update_whitelist
 392     permitted = AttrsPermittedAlways.dup
 393
 394     if self.new_record? || self.state_was == Uncommitted
 395       # Allow create-and-commit in a single operation.
 396       permitted.push(*AttrsPermittedBeforeCommit)
 397     elsif mounts_changed? && mounts_was.keys == mounts.keys
 398       # Ignore the updated mounts if the only changes are default/zero
 399       # values as added by controller, see 17774
 400       only_defaults = true
 401       mounts.each do |path, mount|
 402         (mount.to_a - mounts_was[path].to_a).each do |k, v|
 403           if !["", false, nil].index(v)
 404             only_defaults = false
 405           end
 406         end
 407       end
 408       if only_defaults
 409         clear_attribute_change("mounts")
 410       end
 411     end
 412
 413     case self.state
 414     when Committed
 415       permitted.push :priority, :container_count_max, :container_uuid
 416
 417       if self.container_uuid.nil?
 418         self.errors.add :container_uuid, "has not been resolved to a container."
 419       end
 420
 421       if self.priority.nil?
 422         self.errors.add :priority, "cannot be nil"
 423       end
 424
 425       # Allow container count to increment by 1
 426       if (self.container_uuid &&
 427           self.container_uuid != self.container_uuid_was &&
 428           self.container_count == 1 + (self.container_count_was || 0))
 429         permitted.push :container_count
 430       end
 431
 432       if current_user.andand.is_admin
 433         permitted.push :log_uuid
 434       end
 435
 436     when Final
 437       if self.state_was == Committed
 438         # "Cancel" means setting priority=0, state=Committed
 439         permitted.push :priority
 440
 441         if current_user.andand.is_admin
 442           permitted.push :output_uuid, :log_uuid
 443         end
 444       end
 445
 446     end
 447
 448     super(permitted)
 449   end
 450
 451   def secret_mounts_key_conflict
 452     secret_mounts.each do |k, v|
 453       if mounts.has_key?(k)
 454         errors.add(:secret_mounts, 'conflict with non-secret mounts')
 455         return false
 456       end
 457     end
 458   end
 459
 460   def validate_runtime_token
 461     if !self.runtime_token.nil? && self.runtime_token_changed?
 462       if !runtime_token[0..2] == "v2/"
 463         errors.add :runtime_token, "not a v2 token"
 464         return
 465       end
 466       if ApiClientAuthorization.validate(token: runtime_token).nil?
 467         errors.add :runtime_token, "failed validation"
 468       end
 469     end
 470   end
 471
 472   def scrub_secrets
 473     if self.state == Final
 474       self.secret_mounts = {}
 475       self.runtime_token = nil
 476     end
 477   end
 478
 479   def update_priority
 480     return unless saved_change_to_state? || saved_change_to_priority? || saved_change_to_container_uuid?
 481     act_as_system_user do
 482       Container.
 483         where('uuid in (?)', [container_uuid_before_last_save, self.container_uuid].compact).
 484         map(&:update_priority!)
 485     end
 486   end
 487
 488   def set_priority_zero
 489     self.update_attributes!(priority: 0) if self.state != Final
 490   end
 491
 492   def set_requesting_container_uuid
 493     c = get_requesting_container()
 494     if !c.nil?
 495       self.requesting_container_uuid = c.uuid
 496       # Determine the priority of container request for the requesting
 497       # container.
 498       self.priority = ContainerRequest.where(container_uuid: self.requesting_container_uuid).maximum("priority") || 0
 499     end
 500   end
 501
 502   def get_requesting_container
 503     return self.requesting_container_uuid if !self.requesting_container_uuid.nil?
 504     Container.for_current_token
 505   end
 506 end