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