services/api/app/models/container.rb

   1 require 'whitelist_update'
   2
   3 class Container < ArvadosModel
   4   include HasUuid
   5   include KindAndEtag
   6   include CommonApiTemplate
   7   include WhitelistUpdate
   8
   9   serialize :environment, Hash
  10   serialize :mounts, Hash
  11   serialize :runtime_constraints, Hash
  12   serialize :command, Array
  13
  14   before_validation :fill_field_defaults, :if => :new_record?
  15   before_validation :set_timestamps
  16   validates :command, :container_image, :output_path, :cwd, :priority, :presence => true
  17   validate :validate_state_change
  18   validate :validate_change
  19   validate :validate_lock
  20   after_validation :assign_auth
  21   before_save :sort_serialized_attrs
  22   after_save :handle_completed
  23
  24   has_many :container_requests, :foreign_key => :container_uuid, :class_name => 'ContainerRequest', :primary_key => :uuid
  25   belongs_to :auth, :class_name => 'ApiClientAuthorization', :foreign_key => :auth_uuid, :primary_key => :uuid
  26
  27   api_accessible :user, extend: :common do |t|
  28     t.add :command
  29     t.add :container_image
  30     t.add :cwd
  31     t.add :environment
  32     t.add :exit_code
  33     t.add :finished_at
  34     t.add :locked_by_uuid
  35     t.add :log
  36     t.add :mounts
  37     t.add :output
  38     t.add :output_path
  39     t.add :priority
  40     t.add :progress
  41     t.add :runtime_constraints
  42     t.add :started_at
  43     t.add :state
  44     t.add :auth_uuid
  45   end
  46
  47   # Supported states for a container
  48   States =
  49     [
  50      (Queued = 'Queued'),
  51      (Locked = 'Locked'),
  52      (Running = 'Running'),
  53      (Complete = 'Complete'),
  54      (Cancelled = 'Cancelled')
  55     ]
  56
  57   State_transitions = {
  58     nil => [Queued],
  59     Queued => [Locked, Cancelled],
  60     Locked => [Queued, Running, Cancelled],
  61     Running => [Complete, Cancelled]
  62   }
  63
  64   def state_transitions
  65     State_transitions
  66   end
  67
  68   def update_priority!
  69     if [Queued, Locked, Running].include? self.state
  70       # Update the priority of this container to the maximum priority of any of
  71       # its committed container requests and save the record.
  72       self.priority = ContainerRequest.
  73         where(container_uuid: uuid,
  74               state: ContainerRequest::Committed).
  75         maximum('priority')
  76       self.save!
  77     end
  78   end
  79
  80   protected
  81
  82   def self.deep_sort_hash(x)
  83     if x.is_a? Hash
  84       x.sort.collect do |k, v|
  85         [k, deep_sort_hash(v)]
  86       end.to_h
  87     elsif x.is_a? Array
  88       x.collect { |v| deep_sort_hash(v) }
  89     else
  90       x
  91     end
  92   end
  93
  94   def fill_field_defaults
  95     self.state ||= Queued
  96     self.environment ||= {}
  97     self.runtime_constraints ||= {}
  98     self.mounts ||= {}
  99     self.cwd ||= "."
 100     self.priority ||= 1
 101   end
 102
 103   def permission_to_create
 104     current_user.andand.is_admin
 105   end
 106
 107   def permission_to_update
 108     current_user.andand.is_admin
 109   end
 110
 111   def set_timestamps
 112     if self.state_changed? and self.state == Running
 113       self.started_at ||= db_current_time
 114     end
 115
 116     if self.state_changed? and [Complete, Cancelled].include? self.state
 117       self.finished_at ||= db_current_time
 118     end
 119   end
 120
 121   def validate_change
 122     permitted = [:state]
 123
 124     if self.new_record?
 125       permitted.push(:owner_uuid, :command, :container_image, :cwd,
 126                      :environment, :mounts, :output_path, :priority,
 127                      :runtime_constraints)
 128     end
 129
 130     case self.state
 131     when Queued, Locked
 132       permitted.push :priority
 133
 134     when Running
 135       permitted.push :priority, :progress
 136       if self.state_changed?
 137         permitted.push :started_at
 138       end
 139
 140     when Complete
 141       if self.state_was == Running
 142         permitted.push :finished_at, :output, :log, :exit_code
 143       end
 144
 145     when Cancelled
 146       case self.state_was
 147       when Running
 148         permitted.push :finished_at, :output, :log
 149       when Queued, Locked
 150         permitted.push :finished_at
 151       end
 152
 153     else
 154       # The state_transitions check will add an error message for this
 155       return false
 156     end
 157
 158     check_update_whitelist permitted
 159   end
 160
 161   def validate_lock
 162     # If the Container is already locked by someone other than the
 163     # current api_client_auth, disallow all changes -- except
 164     # priority, which needs to change to reflect max(priority) of
 165     # relevant ContainerRequests.
 166     if locked_by_uuid_was
 167       if locked_by_uuid_was != Thread.current[:api_client_authorization].uuid
 168         check_update_whitelist [:priority]
 169       end
 170     end
 171
 172     if [Locked, Running].include? self.state
 173       # If the Container was already locked, locked_by_uuid must not
 174       # changes. Otherwise, the current auth gets the lock.
 175       need_lock = locked_by_uuid_was || Thread.current[:api_client_authorization].uuid
 176     else
 177       need_lock = nil
 178     end
 179
 180     # The caller can provide a new value for locked_by_uuid, but only
 181     # if it's exactly what we expect. This allows a caller to perform
 182     # an update like {"state":"Unlocked","locked_by_uuid":null}.
 183     if self.locked_by_uuid_changed?
 184       if self.locked_by_uuid != need_lock
 185         return errors.add :locked_by_uuid, "can only change to #{need_lock}"
 186       end
 187     end
 188     self.locked_by_uuid = need_lock
 189   end
 190
 191   def assign_auth
 192     if self.auth_uuid_changed?
 193       return errors.add :auth_uuid, 'is readonly'
 194     end
 195     if not [Locked, Running].include? self.state
 196       # don't need one
 197       self.auth.andand.update_attributes(expires_at: db_current_time)
 198       self.auth = nil
 199       return
 200     elsif self.auth
 201       # already have one
 202       return
 203     end
 204     cr = ContainerRequest.
 205       where('container_uuid=? and priority>0', self.uuid).
 206       order('priority desc').
 207       first
 208     if !cr
 209       return errors.add :auth_uuid, "cannot be assigned because priority <= 0"
 210     end
 211     self.auth = ApiClientAuthorization.
 212       create!(user_id: User.find_by_uuid(cr.modified_by_user_uuid).id,
 213               api_client_id: 0)
 214   end
 215
 216   def sort_serialized_attrs
 217     self.environment = self.class.deep_sort_hash(self.environment)
 218     self.mounts = self.class.deep_sort_hash(self.mounts)
 219     self.runtime_constraints = self.class.deep_sort_hash(self.runtime_constraints)
 220   end
 221
 222   def handle_completed
 223     # This container is finished so finalize any associated container requests
 224     # that are associated with this container.
 225     if self.state_changed? and [Complete, Cancelled].include? self.state
 226       act_as_system_user do
 227         # Notify container requests associated with this container
 228         ContainerRequest.where(container_uuid: uuid,
 229                                :state => ContainerRequest::Committed).each do |cr|
 230           cr.container_completed!
 231         end
 232
 233         # Try to cancel any outstanding container requests made by this container.
 234         ContainerRequest.where(requesting_container_uuid: uuid,
 235                                :state => ContainerRequest::Committed).each do |cr|
 236           cr.priority = 0
 237           cr.save
 238         end
 239       end
 240     end
 241   end
 242
 243 end