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 (new_record? || state_changed?) &&
 330        state == Committed &&
 331        Rails.configuration.Containers.AlwaysUsePreemptibleInstances &&
 332        get_requesting_container_uuid() &&
 333        self.class.any_preemptible_instances?
 334       self.scheduling_parameters['preemptible'] = true
 335     end
 336   end
 337
 338   def validate_runtime_constraints
 339     case self.state
 340     when Committed
 341       ['vcpus', 'ram'].each do |k|
 342         v = runtime_constraints[k]
 343         if !v.is_a?(Integer) || v <= 0
 344           errors.add(:runtime_constraints,
 345                      "[#{k}]=#{v.inspect} must be a positive integer")
 346         end
 347       end
 348     end
 349   end
 350
 351   def validate_datatypes
 352     command.each do |c|
 353       if !c.is_a? String
 354         errors.add(:command, "must be an array of strings but has entry #{c.class}")
 355       end
 356     end
 357     environment.each do |k,v|
 358       if !k.is_a?(String) || !v.is_a?(String)
 359         errors.add(:environment, "must be an map of String to String but has entry #{k.class} to #{v.class}")
 360       end
 361     end
 362     [:mounts, :secret_mounts].each do |m|
 363       self[m].each do |k, v|
 364         if !k.is_a?(String) || !v.is_a?(Hash)
 365           errors.add(m, "must be an map of String to Hash but is has entry #{k.class} to #{v.class}")
 366         end
 367         if v["kind"].nil?
 368           errors.add(m, "each item must have a 'kind' field")
 369         end
 370         [[String, ["kind", "portable_data_hash", "uuid", "device_type",
 371                    "path", "commit", "repository_name", "git_url"]],
 372          [Integer, ["capacity"]]].each do |t, fields|
 373           fields.each do |f|
 374             if !v[f].nil? && !v[f].is_a?(t)
 375               errors.add(m, "#{k}: #{f} must be a #{t} but is #{v[f].class}")
 376             end
 377           end
 378         end
 379         ["writable", "exclude_from_output"].each do |f|
 380           if !v[f].nil? && !v[f].is_a?(TrueClass) && !v[f].is_a?(FalseClass)
 381             errors.add(m, "#{k}: #{f} must be a #{t} but is #{v[f].class}")
 382           end
 383         end
 384       end
 385     end
 386   end
 387
 388   def validate_scheduling_parameters
 389     if self.state == Committed
 390       if scheduling_parameters.include? 'partitions' and
 391          (!scheduling_parameters['partitions'].is_a?(Array) ||
 392           scheduling_parameters['partitions'].reject{|x| !x.is_a?(String)}.size !=
 393             scheduling_parameters['partitions'].size)
 394             errors.add :scheduling_parameters, "partitions must be an array of strings"
 395       end
 396       if scheduling_parameters['preemptible'] &&
 397          (new_record? || state_changed?) &&
 398          !self.class.any_preemptible_instances?
 399         errors.add :scheduling_parameters, "preemptible instances are not configured in InstanceTypes"
 400       end
 401       if scheduling_parameters.include? 'max_run_time' and
 402         (!scheduling_parameters['max_run_time'].is_a?(Integer) ||
 403           scheduling_parameters['max_run_time'] < 0)
 404           errors.add :scheduling_parameters, "max_run_time must be positive integer"
 405       end
 406     end
 407   end
 408
 409   def check_update_whitelist
 410     permitted = AttrsPermittedAlways.dup
 411
 412     if self.new_record? || self.state_was == Uncommitted
 413       # Allow create-and-commit in a single operation.
 414       permitted.push(*AttrsPermittedBeforeCommit)
 415     elsif mounts_changed? && mounts_was.keys.sort == mounts.keys.sort
 416       # Ignore the updated mounts if the only changes are default/zero
 417       # values as added by controller, see 17774
 418       only_defaults = true
 419       mounts.each do |path, mount|
 420         (mount.to_a - mounts_was[path].to_a).each do |k, v|
 421           if ![0, "", false, nil].index(v)
 422             only_defaults = false
 423           end
 424         end
 425       end
 426       if only_defaults
 427         clear_attribute_change("mounts")
 428       end
 429     end
 430
 431     case self.state
 432     when Committed
 433       permitted.push :priority, :container_count_max, :container_uuid
 434
 435       if self.priority.nil?
 436         self.errors.add :priority, "cannot be nil"
 437       end
 438
 439       # Allow container count to increment (not by client, only by us
 440       # -- see set_container)
 441       permitted.push :container_count
 442
 443       if current_user.andand.is_admin
 444         permitted.push :log_uuid
 445       end
 446
 447     when Final
 448       if self.state_was == Committed
 449         # "Cancel" means setting priority=0, state=Committed
 450         permitted.push :priority
 451
 452         if current_user.andand.is_admin
 453           permitted.push :output_uuid, :log_uuid
 454         end
 455       end
 456
 457     end
 458
 459     super(permitted)
 460   end
 461
 462   def secret_mounts_key_conflict
 463     secret_mounts.each do |k, v|
 464       if mounts.has_key?(k)
 465         errors.add(:secret_mounts, 'conflict with non-secret mounts')
 466         return false
 467       end
 468     end
 469   end
 470
 471   def validate_runtime_token
 472     if !self.runtime_token.nil? && self.runtime_token_changed?
 473       if !runtime_token[0..2] == "v2/"
 474         errors.add :runtime_token, "not a v2 token"
 475         return
 476       end
 477       if ApiClientAuthorization.validate(token: runtime_token).nil?
 478         errors.add :runtime_token, "failed validation"
 479       end
 480     end
 481   end
 482
 483   def scrub_secrets
 484     if self.state == Final
 485       self.secret_mounts = {}
 486       self.runtime_token = nil
 487     end
 488   end
 489
 490   def update_priority
 491     return unless saved_change_to_state? || saved_change_to_priority? || saved_change_to_container_uuid?
 492     act_as_system_user do
 493       Container.
 494         where('uuid in (?)', [container_uuid_before_last_save, self.container_uuid].compact).
 495         map(&:update_priority!)
 496     end
 497   end
 498
 499   def set_priority_zero
 500     self.update_attributes!(priority: 0) if self.state != Final
 501   end
 502
 503   def set_requesting_container_uuid
 504     if (self.requesting_container_uuid = get_requesting_container_uuid())
 505       # Determine the priority of container request for the requesting
 506       # container.
 507       self.priority = ContainerRequest.where(container_uuid: self.requesting_container_uuid).maximum("priority") || 0
 508     end
 509   end
 510
 511   def get_requesting_container_uuid
 512     return self.requesting_container_uuid || Container.for_current_token.andand.uuid
 513   end
 514 end